223 lines
8.4 KiB
Python
223 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
||
"""生产补证重组装离线测试：缺口 → 新 attempt / 新快照 / 创作输入变化。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
||
SKILLS_DIR = PROJECT_ROOT / ".agent" / "skills"
|
||
TEST_DIR = PROJECT_ROOT / "tests" / "skills" / "write-next-chapter"
|
||
CHECK_TEST_DIR = PROJECT_ROOT / "tests" / "skills" / "check-content-consistency"
|
||
SCRIPT_DIR = SKILLS_DIR / "write-next-chapter" / "scripts"
|
||
READ_CONTEXT_DIR = SKILLS_DIR / "assemble-context" / "scripts"
|
||
for path in (READ_CONTEXT_DIR, SCRIPT_DIR, TEST_DIR, CHECK_TEST_DIR):
|
||
sys.path.insert(0, str(path))
|
||
|
||
from production_evidence_reassemble import ( # noqa: E402
|
||
EvidenceReassembleError,
|
||
reassemble_writer_context_for_gaps,
|
||
)
|
||
from run_writer_pipeline import InMemoryCasStateStore, PipelineError, run_writer_pipeline # noqa: E402
|
||
from writer_contract import build_writer_creative_input, retrieval_identity # noqa: E402
|
||
|
||
from test_check_writer_candidate import _requirements, _valid_pair # noqa: E402
|
||
from test_run_writer_pipeline import _semantic_report, _writer_output # noqa: E402
|
||
|
||
|
||
class ProductionEvidenceReassembleTests(unittest.TestCase):
|
||
def test_zero_hit_still_changes_creative_input(self) -> None:
|
||
context, _ = _valid_pair()
|
||
gaps = [
|
||
{
|
||
"gapId": "gap-ch3-1",
|
||
"query": "污染链路深度数值含义",
|
||
"reason": "无正典依据",
|
||
"priority": "low",
|
||
}
|
||
]
|
||
with patch(
|
||
"production_evidence_reassemble._search_canonical_snippets",
|
||
return_value=[],
|
||
):
|
||
advanced = reassemble_writer_context_for_gaps(context, gaps, 2, work_id=12)
|
||
self.assertEqual(advanced["runId"], context["runId"])
|
||
self.assertEqual(advanced["attempt"], 2)
|
||
self.assertNotEqual(
|
||
advanced["contextSnapshot"]["contextSha256"],
|
||
context["contextSnapshot"]["contextSha256"],
|
||
)
|
||
self.assertEqual(
|
||
advanced["contextSnapshot"]["contextSha256"],
|
||
retrieval_identity(advanced),
|
||
)
|
||
self.assertNotEqual(
|
||
build_writer_creative_input(advanced),
|
||
build_writer_creative_input(context),
|
||
)
|
||
styles = " ".join(advanced.get("styleConstraints") or [])
|
||
self.assertNotIn("禁止写入", styles)
|
||
self.assertIn("补证轮元数据", styles)
|
||
self.assertIn("未解决", styles)
|
||
|
||
def test_prose_hit_appends_fact_evidence(self) -> None:
|
||
context, _ = _valid_pair()
|
||
gaps = [{"gapId": "gap-a", "query": "徽章", "reason": "需补证", "priority": "high"}]
|
||
hits = [
|
||
{
|
||
"kind": "prose",
|
||
"term": "徽章",
|
||
"chapter": 1,
|
||
"blockId": 99,
|
||
"revision": 3,
|
||
"start": 0,
|
||
"end": 12,
|
||
"text": "旧徽章属于林澈",
|
||
}
|
||
]
|
||
with patch(
|
||
"production_evidence_reassemble._search_canonical_snippets",
|
||
return_value=hits,
|
||
):
|
||
advanced = reassemble_writer_context_for_gaps(context, gaps, 2, work_id=12)
|
||
ids = {item["evidenceId"] for item in advanced["factEvidence"]}
|
||
self.assertIn("gap-gap-a-prose-2-0", ids)
|
||
styles = " ".join(advanced.get("styleConstraints") or [])
|
||
self.assertIn("已追加", styles)
|
||
|
||
def test_invalid_attempt_fails_closed(self) -> None:
|
||
context, _ = _valid_pair()
|
||
with self.assertRaises(EvidenceReassembleError):
|
||
reassemble_writer_context_for_gaps(
|
||
context,
|
||
[{"gapId": "g1", "query": "x", "reason": "y"}],
|
||
1,
|
||
work_id=12,
|
||
)
|
||
|
||
def test_pipeline_evidence_hit_stops_at_authorization(self) -> None:
|
||
"""needs_evidence + 检索命中 -> 授权终态（重写需人授权，以新运行继续）。"""
|
||
|
||
context, _ = _valid_pair()
|
||
calls: list[int] = []
|
||
hits = [
|
||
{
|
||
"kind": "prose",
|
||
"term": "徽章",
|
||
"chapter": 1,
|
||
"blockId": 99,
|
||
"revision": 3,
|
||
"start": 0,
|
||
"end": 12,
|
||
"text": "旧徽章属于林澈",
|
||
}
|
||
]
|
||
|
||
def writer(current: dict, version: int) -> dict:
|
||
calls.append(int(current["attempt"]))
|
||
return _writer_output(current, version)
|
||
|
||
def detector(current: dict, candidate: dict, _mechanical: dict) -> dict:
|
||
return _semantic_report(
|
||
current,
|
||
candidate,
|
||
status="needs_evidence",
|
||
gaps=[
|
||
{
|
||
"gapId": "gap-1",
|
||
"query": "旧徽章归属",
|
||
"reason": "缺口",
|
||
"priority": "high",
|
||
"candidateSha256": candidate["candidateSha256"],
|
||
"candidateQuote": "旧徽章",
|
||
"startCodePoint": 11,
|
||
"endCodePoint": 14,
|
||
}
|
||
],
|
||
)
|
||
|
||
def provider(current, gaps, attempt):
|
||
with patch(
|
||
"production_evidence_reassemble._search_canonical_snippets",
|
||
return_value=hits,
|
||
):
|
||
return reassemble_writer_context_for_gaps(
|
||
current, gaps, attempt, work_id=int(current["workId"])
|
||
)
|
||
|
||
with self.assertRaises(PipelineError) as caught:
|
||
run_writer_pipeline(
|
||
context=context,
|
||
requirements=_requirements(),
|
||
writer=writer,
|
||
evidence_provider=provider,
|
||
semantic_detector=detector,
|
||
state_store=InMemoryCasStateStore(),
|
||
)
|
||
self.assertEqual(caught.exception.code, "AUTHORIZATION_REQUIRED")
|
||
# 只跑单次收敛：写手只被调用一次，重写等人授权。
|
||
self.assertEqual(calls, [1])
|
||
self.assertEqual(caught.exception.details["nextAttempt"], 2)
|
||
self.assertTrue(caught.exception.details["reassembledContextSha256"].startswith("sha256:"))
|
||
|
||
def test_unfillable_gaps_pass_without_rewrite(self) -> None:
|
||
"""正典零命中的缺口是新设定提案：不重写，报告升格 passed，进人闸。"""
|
||
|
||
context, _ = _valid_pair()
|
||
calls: list[int] = []
|
||
|
||
def writer(current: dict, version: int) -> dict:
|
||
calls.append(int(current["attempt"]))
|
||
return _writer_output(current, version)
|
||
|
||
def detector(current: dict, candidate: dict, _mechanical: dict) -> dict:
|
||
return _semantic_report(
|
||
current,
|
||
candidate,
|
||
status="needs_evidence",
|
||
gaps=[
|
||
{
|
||
"gapId": "gap-ch3-01",
|
||
"query": "深渊洋流完整规则",
|
||
"reason": "尚无设定文档",
|
||
"priority": "medium",
|
||
"candidateSha256": candidate["candidateSha256"],
|
||
"candidateQuote": "旧徽章",
|
||
"startCodePoint": 11,
|
||
"endCodePoint": 14,
|
||
}
|
||
],
|
||
)
|
||
|
||
def provider(current, gaps, attempt):
|
||
with patch(
|
||
"production_evidence_reassemble._search_canonical_snippets",
|
||
return_value=[],
|
||
):
|
||
return reassemble_writer_context_for_gaps(
|
||
current, gaps, attempt, work_id=int(current["workId"])
|
||
)
|
||
|
||
result = run_writer_pipeline(
|
||
context=context,
|
||
requirements=_requirements(),
|
||
writer=writer,
|
||
evidence_provider=provider,
|
||
semantic_detector=detector,
|
||
state_store=InMemoryCasStateStore(),
|
||
)
|
||
self.assertEqual(result["status"], "PASSED")
|
||
self.assertEqual(calls, [1])
|
||
self.assertEqual(result.get("rewriteCount"), 0)
|
||
final_report = result["trace"][-1]["semanticReport"]
|
||
self.assertEqual(final_report["status"], "passed")
|
||
self.assertEqual(final_report["evidenceGaps"], [])
|
||
self.assertTrue(final_report["newSettingCandidates"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|