fix: resolve legacy evidence targets without guessing and guide retries

backend/tests/test_optimization.py

@@ -563,6 +563,71 @@ def test_valid_target_cannot_be_attached_to_another_targets_chunk():
         validate_score(result, criterion, [chunk])
 
 
+def test_legacy_missing_target_uses_unique_chunk_association():
+    criterion = {"criterion_id": "x", "max_score": 10, "required_evidence": ["实验数据", "实现方法"]}
+    chunk = {"chunk_id": 1, "page": 1, "page_type": "body", "text": "采用视觉传感器实现位置反馈。", "target_ids": ["t1"]}
+    result = ScoreResult(criterion_id="x", score=8, max_score=10, reason="有实现方法", confidence=.8,
+                        evidence=[{"chunk_id": 1, "page": 1, "quote": chunk["text"]}])
+    assert validate_score(result, criterion, [chunk]).evidence[0].target_id == "t1"
+
+
+def test_target_mismatch_feedback_identifies_the_legal_chunk_target():
+    criterion = {"criterion_id": "x", "max_score": 10, "required_evidence": ["实验数据", "实现方法"]}
+    chunk = {"chunk_id": 42, "page": 1, "page_type": "body", "text": "采用视觉传感器实现位置反馈。", "target_ids": ["t1"]}
+    result = ScoreResult(criterion_id="x", score=8, max_score=10, reason="错误关联", confidence=.8,
+                        evidence=[{"chunk_id": 42, "page": 1, "quote": chunk["text"], "target_id": "t0"}])
+    with pytest.raises(ModelError) as failure:
+        validate_score(result, criterion, [chunk])
+    assert "chunk_id=42" in str(failure.value)
+    assert "允许 target_id=t1" in str(failure.value)
+
+
+@pytest.mark.parametrize("rich", [False, True])
+def test_missing_target_is_not_guessed_when_ambiguous_or_required(rich):
+    required = [{"label": "实验数据"}, {"label": "实现方法"}] if rich else ["实验数据", "实现方法"]
+    criterion = {"criterion_id": "x", "max_score": 10, "required_evidence": required}
+    chunk = {"chunk_id": 1, "page": 1, "page_type": "body", "text": "采用视觉传感器实现位置反馈。", "target_ids": ["t0", "t1"]}
+    result = ScoreResult(criterion_id="x", score=8, max_score=10, reason="需要明确目标", confidence=.8,
+                        evidence=[{"chunk_id": 1, "page": 1, "quote": chunk["text"]}])
+    with pytest.raises(ModelError, match="缺少 target_id"):
+        validate_score(result, criterion, [chunk])
+
+
+def test_scoring_retry_carries_target_contract_and_specific_repair_feedback(sqlite_db):
+    db = sqlite_db
+    item = project(db)
+    file = SubmissionFile(project_id=item.id, category="stage3", relative_path="test.pdf", mime_type="application/pdf",
+                          size=1, mtime=1, sha256="test", recognition_status="recognized", parse_status="ready")
+    db.add(file)
+    db.commit()
+    doc = ingest(db, course_version_id=1, project_id=item.id, submission_file_id=file.id, title="test.pdf",
+                 category="stage3", source_hash="test", pages=[{"page_number": 1, "text": "采用视觉传感器实现位置反馈。"}], embedder=FLAT)
+    chunk = db.scalar(select(KnowledgeChunk).where(KnowledgeChunk.document_id == doc.id))
+    criterion = {"criterion_id": "stage3.design", "category": "stage3", "max_score": 20,
+                 "description": "实现方法", "required_evidence": ["实验数据", "实现方法"]}
+    created = services.create_batch(SimpleNamespace(course_version_id=1, rubric_version_id=1), db)
+    batch = db.get(ScoringBatch, created["id"])
+    class Client:
+        calls = 0
+        def json_call(self, prompt, payload, *args, **kwargs):
+            self.calls += 1
+            assert "每条引文的 target_id 必须属于该片段的 target_ids" in prompt
+            if self.calls == 2:
+                assert payload["validation_feedback"]["error"].endswith("允许 target_id=t1")
+            return ScoreResult(criterion_id=criterion["criterion_id"], score=15, max_score=20,
+                               reason="有实现方法", confidence=.8, evidence=[{
+                                   "chunk_id": chunk.id, "page": 1, "quote": chunk.text,
+                                   "target_id": "t0" if self.calls == 1 else "t1"}])
+    client = Client()
+    def selected(*args, **kwargs):
+        return {"evidence": [{"chunk_id": chunk.id, "document_id": doc.id, "page": 1,
+                              "text": chunk.text, "page_type": "body", "target_ids": ["t1"]}],
+                "uncovered_targets": []}
+    score = _score_criterion(db, batch, item, criterion, client,
+                             retriever=lambda *args, **kwargs: [], multi_retriever=selected)
+    assert score.status == "scored" and client.calls == 2
+
+
 @pytest.mark.parametrize("version", [None, 1])
 def test_worker_rejects_old_policy_without_touching_scores(sqlite_db, version):
     from wisegrade.scoring import run_batch