136 lines
8.8 KiB
Python
136 lines
8.8 KiB
Python
"""Independent, entirely in-memory mapping review; never imports a live driver."""
|
|
import ast
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
OUT = ROOT / "app/artifacts/ai_concurrency"
|
|
spec = importlib.util.spec_from_file_location("r26_mapper_independent_review", OUT / "clinical_ttft_r26_review.py")
|
|
mapper = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mapper)
|
|
|
|
|
|
class MapperIdentityReview(unittest.TestCase):
|
|
def setUp(self):
|
|
self.body = "一、四诊\n\n望诊：神疲、面色少华。\n"
|
|
self.offset = self.body.index("望诊")
|
|
self.byte_offset = len(self.body[:self.offset].encode())
|
|
probe = (OUT / "private_tcm_r26_acceptance_probe.php").read_text(encoding="utf-8")
|
|
self.contract = re.search(r"'harness_contract'=>'([^']+)'", probe)[1]
|
|
self.row = dict(
|
|
test_label="v43r26_offline_0", profile="qwen", ok=True, error_code="",
|
|
synthetic=True, preparation_isolated=True, production_acceptance_path=True,
|
|
preparation_isolation_verified=True, diagnostic_only=False, case_id=11, mode="chat",
|
|
attachment_count=0, generation_budget_seconds=90, deadline_probe=False,
|
|
harness_contract=self.contract, acceptance_driver_sha256=mapper.AUDITED_ACCEPTANCE_DRIVER_SHA256,
|
|
actual_probe_file="private_tcm_r26_acceptance_probe.php",
|
|
actual_probe_sha256=mapper.AUDITED_ACCEPTANCE_PROBE_SHA256,
|
|
production_hashes=dict(probe=mapper.AUDITED_ACCEPTANCE_PROBE_SHA256,
|
|
logic=mapper.AUDITED_REDACTION_LOGIC_SHA256_R26),
|
|
stream_matches_final=True, stream_characters=len(self.body), answer_characters=len(self.body),
|
|
timing_clock="hrtime_monotonic", timing_origin="after_auth_authorization_and_control_begin",
|
|
delivery_prefix_timing_truncated=False,
|
|
delivery_prefix_timing=[dict(utf8_byte_start=0, utf8_byte_end_exclusive=self.byte_offset, at_ms=3.5),
|
|
dict(utf8_byte_start=self.byte_offset, utf8_byte_end_exclusive=len(self.body.encode()), at_ms=17.25)],
|
|
synthetic_review_file="tcm_complete_v43r26_offline_0_qwen.md")
|
|
self.raw = (mapper.HEADER + mapper.SUCCESS + self.body).encode()
|
|
|
|
def test_exact_new_source_tuple_pins(self):
|
|
for relative, expected in [
|
|
("app/artifacts/ai_concurrency/private_tcm_r26_acceptance_driver.py", mapper.AUDITED_ACCEPTANCE_DRIVER_SHA256),
|
|
("app/artifacts/ai_concurrency/private_tcm_r26_acceptance_probe.php", mapper.AUDITED_ACCEPTANCE_PROBE_SHA256),
|
|
("server/app/adminapi/logic/tcm/DiagnosisAiLogic.php", mapper.AUDITED_REDACTION_LOGIC_SHA256_R26),
|
|
]:
|
|
self.assertEqual(hashlib.sha256((ROOT / relative).read_bytes()).hexdigest(), expected)
|
|
self.assertEqual(self.contract, "v43r26_private_run_wire")
|
|
|
|
def test_mechanical_copy_preserves_accounting_and_body_export(self):
|
|
old = (OUT / "clinical_ttft_r25_review.py").read_text(encoding="utf-8")
|
|
current = (OUT / "clinical_ttft_r26_review.py").read_text(encoding="utf-8")
|
|
for actual, previous in [
|
|
(mapper.AUDITED_ACCEPTANCE_DRIVER_SHA256, "263a1a10fa0285dabffe26bb05ebc9683919e7f6b50552949a52d7ed2205e849"),
|
|
(mapper.AUDITED_ACCEPTANCE_PROBE_SHA256, "8858793fe6f189782749345867613a8aa0d9a8f622089173448034874b87a0de"),
|
|
(mapper.AUDITED_REDACTION_LOGIC_SHA256_R26, "1fa058aa67aec0ba0bf9d41b34e53ed125ced1c3eeb888f97a25b56b349d1a9a"),
|
|
]:
|
|
current = current.replace(actual, previous)
|
|
current = current.replace("v43r26_private_run_wire", "v43r25_private_run_wire").replace("R26", "R25").replace("r26", "r25")
|
|
self.assertEqual(ast.dump(ast.parse(current)), ast.dump(ast.parse(old)), "Only revision/contract identity and the exact three approved source hashes may differ.")
|
|
sources = [(OUT / ("private_tcm_" + rev + "_acceptance_driver.py")).read_text(encoding="utf-8") for rev in ("r25", "r26")]
|
|
marker = ' if result.get("preparation_isolation_verified") is True and save_body:'
|
|
blocks = [text[text.index(marker):text.index(" return result", text.index(marker))] for text in sources]
|
|
self.assertEqual(blocks[0], blocks[1], "Complete accepted/partial-body and rejection exporters are unchanged.")
|
|
|
|
def test_actual_new_exporter_contract_maps_original_body(self):
|
|
with patch.object(mapper, "fixed_file", return_value=self.raw):
|
|
result = mapper.row_review(self.row, "望诊", None)
|
|
self.assertEqual(result["measurement_status"], "MAPPED_SELECTED_CLINICAL_ANCHOR")
|
|
self.assertEqual((result["clinically_anchored_ttft_ms"], result["body_utf8_byte_offset"], result["delivery_index"]),
|
|
(17.25, self.byte_offset, 1))
|
|
|
|
def test_different_contract_is_not_accepted_as_original_exporter(self):
|
|
invalid = {**self.row, "harness_contract": "v43r26_run_aware"}
|
|
with patch.object(mapper, "fixed_file", return_value=self.raw):
|
|
result = mapper.row_review(invalid, "望诊", None)
|
|
self.assertEqual(result["measurement_status"], "R26_FROZEN_EXPORT_IDENTITY_UNPROVEN")
|
|
|
|
def test_low_level_offsets_and_windows_writer_inversion(self):
|
|
self.assertEqual(mapper.select_offset(self.body, "望诊", None), self.offset)
|
|
self.assertEqual(mapper.timing_at(self.row, self.body, self.offset), (17.25, self.byte_offset, 1))
|
|
original = "望诊：记录含原始\r\n换行。\n"
|
|
windows = (mapper.HEADER + mapper.SUCCESS + original).replace("\n", "\r\n").encode()
|
|
self.assertEqual(mapper.review_body(windows, True), original)
|
|
for marker in mapper.REDACTION_MARKERS:
|
|
marked = self.body + marker
|
|
row = {**self.row, "stream_characters": len(marked), "answer_characters": len(marked)}
|
|
with self.assertRaisesRegex(mapper.EvidenceError, "FINAL_OR_REDACTED_BODY_STREAM_IDENTITY_UNPROVEN"):
|
|
mapper.timing_at(row, marked, self.offset)
|
|
|
|
def test_negative_byte_and_identity_controls(self):
|
|
mutations = [
|
|
lambda row: row.update(stream_characters=len(self.body) + 1),
|
|
lambda row: row.update(stream_matches_final=False),
|
|
lambda row: row.update(actual_probe_sha256="0" * 64),
|
|
lambda row: row["production_hashes"].update(logic="0" * 64),
|
|
lambda row: row["delivery_prefix_timing"][1].update(utf8_byte_start=self.byte_offset + 1),
|
|
lambda row: row["delivery_prefix_timing"][0].update(utf8_byte_end_exclusive=1),
|
|
lambda row: row["delivery_prefix_timing"][1].update(at_ms=1),
|
|
lambda row: row["delivery_prefix_timing"][1].update(at_ms=math.nan),
|
|
lambda row: row["delivery_prefix_timing"][1].update(utf8_byte_end_exclusive=len(self.body.encode()) - 1),
|
|
]
|
|
for mutate in mutations:
|
|
row = copy.deepcopy(self.row); mutate(row)
|
|
with self.assertRaises(mapper.EvidenceError):
|
|
mapper.timing_at(row, self.body, self.offset)
|
|
for relative in ("../outside.md", "tcm_complete_wrong_qwen.md"):
|
|
with self.assertRaisesRegex(mapper.EvidenceError, "UNSAFE_OR_UNEXPECTED_SYNTHETIC_FILENAME"):
|
|
mapper.fixed_file(relative, "tcm_complete_v43r26_offline_0_qwen.md")
|
|
|
|
def test_anchor_and_manifest_negative_controls(self):
|
|
body = "一、四诊\n\n已收到资料。\n<think>望诊：推理文字。</think>\n```\n望诊：代码。\n```\n望诊：实采文字。\n"
|
|
self.assertEqual(mapper.select_offset(body, "望诊", None), body.index("望诊：实采"))
|
|
with self.assertRaises(mapper.EvidenceError):
|
|
mapper.select_offset(body, "望诊", body.index("望诊：推理"))
|
|
stage = "v43r26_offline"
|
|
doc = dict(stage=stage, synthetic_only=True, preparation_isolated=True, production_acceptance_path=True,
|
|
diagnostic_only=False, expected_requests=[dict(test_label=self.row["test_label"], profile="qwen"),
|
|
dict(test_label=stage + "_1", profile="openai")],
|
|
rows=[self.row], completed_requests=1, suite_complete=False)
|
|
with patch.object(mapper, "fixed_file", return_value=json.dumps(doc).encode()), \
|
|
patch.object(mapper, "row_review", return_value={"ok": True}):
|
|
review = mapper.review_stage(stage)
|
|
self.assertEqual((review["completed_requests"], review["unrecorded_requests"], review["suite_complete"]), (1, 1, False))
|
|
doc["suite_complete"] = True
|
|
with patch.object(mapper, "fixed_file", return_value=json.dumps(doc).encode()), self.assertRaisesRegex(mapper.EvidenceError, "MANIFEST_COMPLETION_MISMATCH"):
|
|
mapper.review_stage(stage)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|