"""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("r25_mapper_independent_review", OUT / "clinical_ttft_r25_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_r25_acceptance_probe.php").read_text(encoding="utf-8") self.contract = re.search(r"'harness_contract'=>'([^']+)'", probe)[1] self.row = dict( test_label="v43r25_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_r25_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_R25), 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_v43r25_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_r25_acceptance_driver.py", mapper.AUDITED_ACCEPTANCE_DRIVER_SHA256), ("app/artifacts/ai_concurrency/private_tcm_r25_acceptance_probe.php", mapper.AUDITED_ACCEPTANCE_PROBE_SHA256), ("server/app/adminapi/logic/tcm/DiagnosisAiLogic.php", mapper.AUDITED_REDACTION_LOGIC_SHA256_R25), ]: self.assertEqual(hashlib.sha256((ROOT / relative).read_bytes()).hexdigest(), expected) self.assertEqual(self.contract, "v43r25_private_run_wire") def test_mechanical_copy_preserves_accounting_and_body_export(self): old = (OUT / "clinical_ttft_r24_review.py").read_text(encoding="utf-8") current = (OUT / "clinical_ttft_r25_review.py").read_text(encoding="utf-8") for actual, previous in [ (mapper.AUDITED_ACCEPTANCE_DRIVER_SHA256, "ddfb14bbb1362705ec246e803dfbbfc0252f0008f86db95eb86ab82ca9f20453"), (mapper.AUDITED_ACCEPTANCE_PROBE_SHA256, "29f6b04b4ad5bfb93d19350e9181688fe42f29cedb516b2ec2303e92b6f9e79f"), (mapper.AUDITED_REDACTION_LOGIC_SHA256_R25, "91f01bad7fb2b3890cd2425371cb4eb5c02bd3878803f06e5949b8ecdd3b2fb7"), ]: current = current.replace(actual, previous) current = current.replace("v43r25_private_run_wire", "v43r24_run_aware").replace("R25", "R24").replace("r25", "r24") 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 ("r24", "r25")] 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": "v43r25_run_aware"} with patch.object(mapper, "fixed_file", return_value=self.raw): result = mapper.row_review(invalid, "望诊", None) self.assertEqual(result["measurement_status"], "R25_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_v43r25_offline_0_qwen.md") def test_anchor_and_manifest_negative_controls(self): body = "一、四诊\n\n已收到资料。\n望诊：推理文字。\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 = "v43r25_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()