from __future__ import annotations from collections import OrderedDict from datetime import datetime import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch from typing import Mapping from zoneinfo import ZoneInfo from hibor_fast_collection.budget import Budget from hibor_fast_collection.models import Candidate, ContractError, TaskSpec from hibor_fast_collection.performance import ( EVIDENCE_REFERENCE_KEYS, PLAN_ENTRY_KEYS, PLAN_KEYS, PDFINFO_EXECUTABLE, PREPLAN_EVIDENCE_REFERENCE_KEYS, PREPLAN_KEYS, PREPLAN_STOP_CODES, build_performance_summary_from_evidence, write_performance_package, write_performance_plan, write_preplan_package, main as performance_main, ) import hibor_fast_collection.performance as performance_module from hibor_fast_collection.manifests import write_json_create_new from hibor_fast_collection.quota import AppQuotaObservation, QuotaLedger from hibor_fast_collection.selection import ScanOutcome from hibor_fast_collection.workflow import HiborWorkflow from helpers import task_spec from test_workflow_fake_e2e import FakeAdb, FakeUi def report(title: str, institution: str = "测试证券") -> OrderedDict: report_date = "2026-07-29" identity = hashlib.sha256(f"{title}|{institution}|{report_date}".encode("utf-8")).hexdigest() return OrderedDict((key, value) for key, value in ( ("report_identity", identity), ("title", title), ("institution", institution), ("report_date", report_date), ("page_count", 1), )) class PerformanceAdb(FakeAdb): def __init__(self, count: int): super().__init__() from hibor_fast_collection.adb import RemoteFile self.trigger_count = 0 self.remotes = tuple( RemoteFile( f"/sdcard/Android/data/cn.com.hibor/files/myfile/report-{index}.pdf", f"report-{index}.pdf", len(self._pdf(index)), f"2026-07-29 10:{index:02d}", ) for index in range(1, count + 1) ) @staticmethod def _pdf(index: int) -> bytes: data = bytearray(f"%PDF-1.4\n% ITEM {index}\n".encode("ascii")) objects = ( b"<< /Type /Catalog /Pages 2 0 R >>", b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>", b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Contents 4 0 R >>", b"<< /Length 0 >>\nstream\n\nendstream", ) offsets = [0] for number, body in enumerate(objects, 1): offsets.append(len(data)) data.extend(f"{number} 0 obj\n".encode("ascii")) data.extend(body) data.extend(b"\nendobj\n") xref_offset = len(data) data.extend(b"xref\n0 5\n0000000000 65535 f \n") for offset in offsets[1:]: data.extend(f"{offset:010d} 00000 n \n".encode("ascii")) data.extend( f"trailer\n<< /Size 5 /Root 1 0 R >>\nstartxref\n{xref_offset}\n%%EOF\n" .encode("ascii") ) return bytes(data) def list_cache(self): return self.remotes[:self.trigger_count] def remote_sha256(self, remote_path: str): index = next(i for i, row in enumerate(self.remotes, 1) if row.path == remote_path) return hashlib.sha256(self._pdf(index)).hexdigest() def pull(self, remote_path: str, local_staging: Path, *, timeout_ms: int): index = next(i for i, row in enumerate(self.remotes, 1) if row.path == remote_path) local_staging.parent.mkdir(parents=True, exist_ok=True) local_staging.write_bytes(self._pdf(index)) return object() class PerformanceUi: def __init__(self, adb: PerformanceAdb, reports: list[Mapping[str, object]]): self.adb = adb self.reports = reports self.candidates = tuple( Candidate(f"CANDIDATE-{index:03d}", str(row["title"]), None, None, score=120, bounds=(1, index, 2, index + 1)) for index, row in enumerate(reports, 1) ) def search(self, query: str): self.query = query def scan(self, **kwargs): return ScanOutcome("ENOUGH_CONFIRMED", self.candidates, 3, len(self.candidates), 0) def open_detail(self, candidate: Candidate): row = next(value for value in self.reports if value["title"] == candidate.title) return Candidate(candidate.candidate_id, candidate.title, str(row["institution"]), str(row["report_date"]), ("分析师甲",), int(row["page_count"]), candidate.score, candidate.bounds) def trigger_current_detail(self): self.adb.trigger_count += 1 self.adb.triggered = True class DiscoveryPerformanceTests(unittest.TestCase): def test_readonly_discovery_returns_exact_identity_without_quota_or_trigger(self): class DiscoveryUi(FakeUi): def trigger_current_detail(self): raise AssertionError("discovery must not trigger") with tempfile.TemporaryDirectory() as temp: root = Path(temp) spec = TaskSpec.from_mapping(task_spec(root, mode="discover")) adb = FakeAdb() result = HiborWorkflow(spec).discover(adb=adb, ui=DiscoveryUi(adb), execute=True) self.assertEqual(result["schema_version"], "HIBOR_FAST_DISCOVERY_V001") self.assertEqual(result["candidates"], [dict(report("中国中免深度研究"))]) self.assertFalse(result["quota_ledger_touched"]) self.assertFalse(result["trigger_attempted"]) self.assertFalse(spec.quota_path.exists()) self.assertFalse(adb.triggered) def test_expected_identity_drift_stops_before_reserve_and_tap(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") # Same visible title but a different institution yields a reachable # detail page whose exact identity cannot equal the frozen plan. raw["expected_reports"] = [report("中国中免深度研究", "另一机构")] spec = TaskSpec.from_mapping(raw) adb = FakeAdb() ui = FakeUi(adb) result = HiborWorkflow(spec).collect(adb=adb, ui=ui, execute=True, synthetic=True) self.assertEqual((result["status"], result["triggered"]), ("BLOCKED_INPUT", 0)) self.assertFalse(adb.triggered) ledger = spec.quota_path.read_text(encoding="utf-8") self.assertNotIn("RESERVATION", ledger) def test_item_windows_persist_previous_terminal_delta_with_fake_clock(self): class Clock: value = 0.0 def __call__(self): return self.value with tempfile.TemporaryDirectory() as temp: clock = Clock() spec = TaskSpec.from_mapping(task_spec(Path(temp), mode="batch", quantity=2)) budget = Budget(spec.total_budget_ms, spec.close_reserve_ms, batch_increment_ms=spec.batch_increment_budget_ms, clock=clock) workflow = HiborWorkflow(spec, budget=budget, monotonic_ns=lambda: int(clock.value * 1_000_000_000)) workflow._begin_item(1) clock.value = 10.0 self.assertTrue(workflow._complete_item(1)) workflow._begin_item(2) clock.value = 249.999 self.assertTrue(workflow._complete_item(2)) timing = workflow._timing_value("RUN-TEST") self.assertEqual(timing["schema_version"], "HIBOR_FAST_TIMING_V002") self.assertEqual(timing["item_windows"][0]["elapsed_ms"], 10_000) self.assertEqual(timing["item_windows"][1]["terminal_delta_from_previous_ms"], 239_999) self.assertTrue(all(row["deadline_met"] for row in timing["item_windows"])) def test_performance_plan_and_summary_exact_replay_and_drift(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) / "HIBOR-FAST-PERFORMANCE-TEST" queries = ["三环集团", "三环集团", "国瓷材料", "国瓷材料", "MLCC", "MLCC", "MLCC", "MLCC", "MLCC", "MLCC"] reports = [report(f"{query} 测试研报 {index}") for index, query in enumerate(queries, 1)] discovery_paths: list[Path] = [] discovery_hashes: list[str] = [] for index in range(6): value = { "schema_version": "HIBOR_FAST_DISCOVERY_V001", "candidates": [reports[index]], "quota_ledger_touched": False, "trigger_attempted": False, } path = root / "discovery" / f"single-{index + 1}.json" path.parent.mkdir(parents=True, exist_ok=True) raw = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8") path.write_bytes(raw) discovery_paths.append(path) discovery_hashes.append(hashlib.sha256(raw).hexdigest()) batch_discovery = { "schema_version": "HIBOR_FAST_DISCOVERY_V001", "candidates": reports[6:], "quota_ledger_touched": False, "trigger_attempted": False, } batch_discovery_path = root / "discovery" / "batch.json" batch_raw = json.dumps(batch_discovery, ensure_ascii=False, separators=(",", ":")).encode("utf-8") batch_discovery_path.write_bytes(batch_raw) batch_hash = hashlib.sha256(batch_raw).hexdigest() discovery_paths.extend([batch_discovery_path] * 4) discovery_hashes.extend([batch_hash] * 4) entries = [] for index, (query, expected) in enumerate(zip(queries, reports), 1): values = { "slot_id": f"PERF-{index:02d}", "mode": "batch" if index >= 7 else "collect-one", "batch_id": "PERF-B01" if index >= 7 else None, "item_order": index - 6 if index >= 7 else 1, "query": query, "expected_report": expected, "discovery_sha256": discovery_hashes[index - 1], "external_exclusion_allowed": True, "selected_from_readonly_discovery": True, } entries.append(OrderedDict((key, values[key]) for key in PLAN_ENTRY_KEYS)) quota_date = datetime.now(ZoneInfo("Asia/Shanghai")).date().isoformat() quota_control = root.parent / "report-collection-control" quota_path = quota_control / f"daily_quota_{quota_date}.csv" historical_path = quota_control / "daily_quota_2026-07-29.csv" if historical_path != quota_path: QuotaLedger(historical_path, quota_date="2026-07-29").initialize( task_id="HISTORICAL", requester_role="TEST", handoff_id="HISTORICAL", known_floor=3, evidence_ref="KNOWN-THREE", ) plan_raw = { "schema_version": "HIBOR_FAST_PERFORMANCE_PLAN_V002", "plan_id": "PERF-PLAN-TEST", "created_at_utc": "2026-07-30T00:00:00Z", "quota_date": quota_date, "quota_ledger": str(quota_path), "historical_quota_date": "2026-07-29", "historical_quota_ledger": str(historical_path), "trigger_cap": 10, "entries": entries, } plan = OrderedDict((key, plan_raw[key]) for key in PLAN_KEYS) plan_path = root / "control" / "performance_plan.json" first = write_performance_plan(plan_path, plan) self.assertEqual(write_performance_plan(plan_path, plan), first) drift = OrderedDict(plan) drift["created_at_utc"] = "2026-07-30T00:00:01Z" with self.assertRaises(ContractError): write_performance_plan(plan_path, drift) refs: list[OrderedDict] = [] task_paths: list[Path] = [] for index in range(6): raw = task_spec(root, mode="collect-one") raw["task_id"] = f"TASK-PERF-{index + 1:02d}" raw["handoff_id"] = f"HANDOFF-PERF-{index + 1:02d}" raw["query"] = queries[index] raw["expected_reports"] = [reports[index]] raw["performance_slot_id"] = f"PERF-{index + 1:02d}" raw["performance_plan_id"] = plan["plan_id"] raw["quota_ledger"] = str(quota_path) raw["destination"] = str(root / "archive") raw["pdfinfo_executable"] = str(PDFINFO_EXECUTABLE) task_path = root / "tasks" / f"single-{index + 1}.json" task_path.parent.mkdir(parents=True, exist_ok=True) task_path.write_text(json.dumps(raw, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") task_paths.append(task_path) spec = TaskSpec.from_mapping(raw) adb = PerformanceAdb(1) result = HiborWorkflow(spec).collect( adb=adb, ui=PerformanceUi(adb, [reports[index]]), execute=True, synthetic=False, ) self.assertEqual(result["status"], "SUCCESS", result) ref_values = { "slot_id": f"PERF-{index + 1:02d}", "task_path": str(task_path), "discovery_path": str(discovery_paths[index]), "terminal_path": result["terminal_path"], "timing_path": result["timing_path"], "manifest_path": result["manifest_path"], "item_id": "ITEM-001", } refs.append(OrderedDict((key, ref_values[key]) for key in EVIDENCE_REFERENCE_KEYS)) batch_task = task_spec(root, mode="batch", quantity=4) batch_task["task_id"] = "TASK-PERF-B01" batch_task["handoff_id"] = "HANDOFF-PERF-B01" batch_task["query"] = "MLCC" batch_task["expected_reports"] = reports[6:] batch_task["performance_slot_id"] = "PERF-B01" batch_task["performance_plan_id"] = plan["plan_id"] batch_task["quota_ledger"] = str(quota_path) batch_task["destination"] = str(root / "archive") batch_task["pdfinfo_executable"] = str(PDFINFO_EXECUTABLE) batch_task_path = root / "tasks" / "batch.json" batch_task_path.write_text(json.dumps(batch_task, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") batch_spec = TaskSpec.from_mapping(batch_task) batch_adb = PerformanceAdb(4) batch_result = HiborWorkflow(batch_spec).collect( adb=batch_adb, ui=PerformanceUi(batch_adb, reports[6:]), execute=True, synthetic=False, ) self.assertEqual(batch_result["status"], "SUCCESS", batch_result) for offset in range(4): ref_values = { "slot_id": f"PERF-{offset + 7:02d}", "task_path": str(batch_task_path), "discovery_path": str(batch_discovery_path), "terminal_path": batch_result["terminal_path"], "timing_path": batch_result["timing_path"], "manifest_path": batch_result["manifest_path"], "item_id": f"ITEM-{offset + 1:03d}", } refs.append(OrderedDict((key, ref_values[key]) for key in EVIDENCE_REFERENCE_KEYS)) summary, _ = build_performance_summary_from_evidence( plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=refs, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) self.assertEqual(summary["result"], "PERFORMANCE_PASS", summary) self.assertEqual(len(summary["batch_increment_ms"]), 3) summary_path = root / "performance_summary.json" evidence_manifest_path = root / "performance_manifest.csv" one = write_performance_package( summary_path=summary_path, evidence_manifest_path=evidence_manifest_path, plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=refs, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) self.assertEqual(write_performance_package( summary_path=summary_path, evidence_manifest_path=evidence_manifest_path, plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=refs, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ), one) with self.assertRaises(ContractError): write_performance_package( summary_path=summary_path, evidence_manifest_path=evidence_manifest_path, plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=refs, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:01Z", synthetic=False, ) cli_request = OrderedDict((key, value) for key, value in ( ("kind", "SUMMARY"), ("plan_path", str(plan_path)), ("plan_bytes", first[0]), ("plan_sha256", first[1]), ("evidence_root", str(root)), ("evidence_references", refs), ("started_at_utc", "2026-07-30T00:00:00Z"), ("ended_at_utc", "2026-07-30T01:00:00Z"), )) cli_request_path = root / "control" / "performance_summary_request.json" cli_request_path.write_bytes(json.dumps(cli_request, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) cli_summary = root / "package" / "performance_summary.json" cli_manifest = root / "package" / "performance_manifest.csv" with patch.object(performance_module, "PERFORMANCE_EVIDENCE_ROOT", root): self.assertEqual(performance_main([ "--request", str(cli_request_path), "--output", str(cli_summary), "--manifest-output", str(cli_manifest), ]), 0) self.assertEqual(json.loads(cli_summary.read_text(encoding="utf-8"))["result"], "PERFORMANCE_PASS") cli_manifest_text = cli_manifest.read_text(encoding="utf-8") self.assertIn(",performance_plan,", cli_manifest_text) self.assertIn(",runtime_pdfinfo,", cli_manifest_text) with self.assertRaises(ContractError): build_performance_summary_from_evidence( plan_path=plan_path, plan_bytes=first[0], plan_sha256="0" * 64, evidence_root=root, evidence_references=refs, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) partial, _ = build_performance_summary_from_evidence( plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=refs[:7], started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) self.assertEqual(partial["result"], "INSUFFICIENT_PERFORMANCE_SAMPLES") wrong_order = list(refs) wrong_order[0], wrong_order[1] = wrong_order[1], wrong_order[0] with self.assertRaises(ContractError): build_performance_summary_from_evidence( plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=wrong_order, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) # The predecessor's caller-asserted booleans API no longer exists. with self.assertRaises(TypeError): build_performance_summary_from_evidence( # type: ignore[call-arg] plan=plan, samples=[{"pdf_magic_pass": True}], quota={} ) missing = list(refs) missing[0] = OrderedDict(missing[0]) missing[0]["terminal_path"] = str(root / "missing-terminal.json") with self.assertRaises(ContractError): build_performance_summary_from_evidence( plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=missing, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) original_terminal = Path(refs[0]["terminal_path"]) fake_terminal = json.loads(original_terminal.read_text(encoding="utf-8")) fake_terminal["items"][0]["reused_without_new_trigger"] = True fake_path = root / "fake-pure-reuse-terminal.json" fake_path.write_bytes(json.dumps(fake_terminal, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) pure_reuse = list(refs) pure_reuse[0] = OrderedDict(pure_reuse[0]) pure_reuse[0]["terminal_path"] = str(fake_path) with self.assertRaises(ContractError): build_performance_summary_from_evidence( plan_path=plan_path, plan_bytes=first[0], plan_sha256=first[1], evidence_root=root, evidence_references=pure_reuse, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) bad_history = root / "bad" / "daily_quota_2026-07-29.csv" QuotaLedger(bad_history, quota_date="2026-07-29").initialize( task_id="BAD", requester_role="TEST", handoff_id="BAD", known_floor=0, evidence_ref="BAD-ZERO", ) bad_plan = OrderedDict(plan) bad_plan["historical_quota_ledger"] = str(bad_history) bad_plan_path = root / "bad-performance-plan.json" bad_plan_receipt = write_performance_plan(bad_plan_path, bad_plan) with self.assertRaises(ContractError): build_performance_summary_from_evidence( plan_path=bad_plan_path, plan_bytes=bad_plan_receipt[0], plan_sha256=bad_plan_receipt[1], evidence_root=root, evidence_references=refs, started_at_utc="2026-07-30T00:00:00Z", ended_at_utc="2026-07-30T01:00:00Z", synthetic=False, ) def test_preplan_all_stops_are_create_new_replayable_and_zero_trigger(self): with tempfile.TemporaryDirectory() as temp: parent = Path(temp) for index, stop_code in enumerate(PREPLAN_STOP_CODES): with self.subTest(stop_code=stop_code): root = (parent / f"case-{index}").resolve() control = root / "control" control.mkdir(parents=True) context_values = { "schema_version": "HIBOR_FAST_PERFORMANCE_EXECUTION_CONTEXT_V001", "task_id": "DEV-ANA-HIBOR-FAST-COLLECTION-20260729-001", "design_plan_id": "PERF-TEST-PLAN-ANA-HIBOR-FAST-COLLECTION-V003", "evidence_root": str(root), "execution_started_at_utc": "2026-07-30T00:00:00Z", "timezone": "Asia/Shanghai", "trigger_cap": 10, "subjects": ["三环集团", "国瓷材料", "MLCC"], } context = OrderedDict((key, context_values[key]) for key in ( "schema_version", "task_id", "design_plan_id", "evidence_root", "execution_started_at_utc", "timezone", "trigger_cap", "subjects", )) context_path = control / "execution_context.json" write_json_create_new(context_path, context) terminal_values = { "schema_version": "HIBOR_FAST_PREPLAN_TERMINAL_V001", "task_id": context["task_id"], "design_plan_id": context["design_plan_id"], "execution_started_at_utc": context["execution_started_at_utc"], "ended_at_utc": f"2026-07-30T00:00:{index + 1:02d}Z", "phase": "PREPLAN", "status": "STOPPED", "stop_code": stop_code, "exit_code": 27 if stop_code == "STATE_UNCERTAIN" else 10, "evidence_root": str(root), "plan_path": str(control / "performance_plan.json"), "plan_presence": "U" if stop_code == "STATE_UNCERTAIN" else "N", "plan_bytes": None, "plan_sha256": None, "discovery_completed": 0, "qualified_candidate_total": 0, "reserve_count": 0, "trigger_count": 0, "quota_write_count": 0, "blockers": [stop_code], } terminal = OrderedDict((key, terminal_values[key]) for key in PREPLAN_KEYS) evidence_values = { "evidence_type": "execution_context", "path": str(context_path), "status": "VALID", } evidence = [OrderedDict((key, evidence_values[key]) for key in PREPLAN_EVIDENCE_REFERENCE_KEYS)] terminal_path = root / "preplan" / "preplan_terminal.json" manifest_path = root / "preplan" / "preplan_manifest.csv" first = write_preplan_package( terminal_path=terminal_path, manifest_path=manifest_path, terminal_value=terminal, evidence_root=root, evidence_paths=evidence, ) self.assertEqual(write_preplan_package( terminal_path=terminal_path, manifest_path=manifest_path, terminal_value=terminal, evidence_root=root, evidence_paths=evidence, ), first) persisted = json.loads(terminal_path.read_text(encoding="utf-8")) self.assertEqual((persisted["reserve_count"], persisted["trigger_count"]), (0, 0)) self.assertNotIn("performance_plan", manifest_path.read_text(encoding="utf-8")) drift = OrderedDict(terminal) drift["ended_at_utc"] = "2026-07-30T00:01:00Z" with self.assertRaises(ContractError): write_preplan_package( terminal_path=terminal_path, manifest_path=manifest_path, terminal_value=drift, evidence_root=root, evidence_paths=evidence, ) invalid = OrderedDict(terminal) invalid["reserve_count"] = 1 with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "other" / "terminal.json", manifest_path=root / "other" / "manifest.csv", terminal_value=invalid, evidence_root=root, evidence_paths=evidence, ) if index == 0: request = OrderedDict((key, value) for key, value in ( ("kind", "PREPLAN"), ("terminal_value", terminal), ("evidence_root", str(root)), ("evidence_paths", evidence), )) request_path = control / "preplan_stop_request.json" request_path.write_bytes(json.dumps(request, ensure_ascii=False, separators=(",", ":")).encode("utf-8")) with patch.object(performance_module, "PERFORMANCE_EVIDENCE_ROOT", root): self.assertEqual(performance_main([ "--request", str(request_path), "--output", str(terminal_path), "--manifest-output", str(manifest_path), ]), 0) def test_preplan_recomputes_physical_evidence_and_rejects_semantic_lies(self): def prepare(root: Path) -> tuple[OrderedDict, Path]: control = root / "control" control.mkdir(parents=True) values = { "schema_version": "HIBOR_FAST_PERFORMANCE_EXECUTION_CONTEXT_V001", "task_id": "DEV-ANA-HIBOR-FAST-COLLECTION-20260729-001", "design_plan_id": "PERF-TEST-PLAN-ANA-HIBOR-FAST-COLLECTION-V003", "evidence_root": str(root), "execution_started_at_utc": "2026-07-30T00:00:00Z", "timezone": "Asia/Shanghai", "trigger_cap": 10, "subjects": ["三环集团", "国瓷材料", "MLCC"], } context = OrderedDict((key, values[key]) for key in ( "schema_version", "task_id", "design_plan_id", "evidence_root", "execution_started_at_utc", "timezone", "trigger_cap", "subjects", )) path = control / "execution_context.json" write_json_create_new(path, context) return context, path def terminal(root: Path, context: Mapping[str, object], **changes: object) -> OrderedDict: values: dict[str, object] = { "schema_version": "HIBOR_FAST_PREPLAN_TERMINAL_V001", "task_id": context["task_id"], "design_plan_id": context["design_plan_id"], "execution_started_at_utc": context["execution_started_at_utc"], "ended_at_utc": "2026-07-30T00:01:00Z", "phase": "PREPLAN", "status": "STOPPED", "stop_code": "CANDIDATES_INSUFFICIENT", "exit_code": 10, "evidence_root": str(root), "plan_path": str(root / "control" / "performance_plan.json"), "plan_presence": "N", "plan_bytes": None, "plan_sha256": None, "discovery_completed": 0, "qualified_candidate_total": 0, "reserve_count": 0, "trigger_count": 0, "quota_write_count": 0, "blockers": ["CANDIDATES_INSUFFICIENT"], } values.update(changes) return OrderedDict((key, values[key]) for key in PREPLAN_KEYS) def evidence(evidence_type: str, path: Path, status: str = "VALID") -> OrderedDict: values = {"evidence_type": evidence_type, "path": str(path), "status": status} return OrderedDict((key, values[key]) for key in PREPLAN_EVIDENCE_REFERENCE_KEYS) with tempfile.TemporaryDirectory() as temp: parent = Path(temp) # A physical discovery object, not caller counters, is authoritative. root = (parent / "physical-discovery").resolve() context, context_path = prepare(root) discovery_values = { "schema_version": "HIBOR_FAST_DISCOVERY_V001", "task_id": "TASK-PERF-S01", "handoff_id": "HANDOFF-PERF-S01", "performance_plan_id": "PERF-PLAN-TEST", "performance_slot_id": "PERF-S01", "query": "三环集团", "observed_at_utc": "2026-07-30T00:00:01Z", "completed_at_utc": "2026-07-30T00:00:02Z", "started_monotonic_ns": 1, "ended_monotonic_ns": 2, "screens_scanned": 3, "candidate_count": 1, "candidates": [report("三环集团 测试研报 1")], "quota_ledger_touched": False, "trigger_attempted": False, } discovery = OrderedDict( (key, discovery_values[key]) for key in performance_module.DISCOVERY_KEYS ) discovery_path = root / "discovery" / "PERF-S01.json" write_json_create_new(discovery_path, discovery) good_terminal = terminal( root, context, discovery_completed=1, qualified_candidate_total=1, ) write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=good_terminal, evidence_root=root, evidence_paths=[evidence("execution_context", context_path), evidence("discovery_PERF-S01", discovery_path)], ) root = (parent / "discovery-prefix-lie").resolve() context, context_path = prepare(root) discovery_values["performance_slot_id"] = "PERF-S02" discovery_values["task_id"] = "TASK-PERF-S02" discovery_values["handoff_id"] = "HANDOFF-PERF-S02" discovery = OrderedDict( (key, discovery_values[key]) for key in performance_module.DISCOVERY_KEYS ) discovery_path = root / "discovery" / "PERF-S02.json" write_json_create_new(discovery_path, discovery) with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal( root, context, discovery_completed=1, qualified_candidate_total=1, ), evidence_root=root, evidence_paths=[evidence("execution_context", context_path), evidence("discovery_PERF-S02", discovery_path)], ) # APP_RECONCILE is counted from the durable quota ledger itself. root = (parent / "physical-quota").resolve() context, context_path = prepare(root) quota_path = root / "control" / "daily_quota_2026-07-30.csv" quota = QuotaLedger(quota_path, quota_date="2026-07-30") quota.initialize( task_id="PERF-PREPLAN", requester_role="TEST", handoff_id="HANDOFF-PREPLAN", known_floor=3, evidence_ref="KNOWN-THREE", ) observation = AppQuotaObservation.build( device_serial="emulator-5554", quota_date="2026-07-30", captured_at_utc="2026-07-30T00:00:10Z", visible_remaining=20, ui_snapshot_fingerprint="f" * 64, ) quota.observe_app_remaining( task_id="PERF-PREPLAN", requester_role="TEST", handoff_id="HANDOFF-PREPLAN", observation=observation, ) write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal( root, context, stop_code="QUOTA_INSUFFICIENT", blockers=["QUOTA_INSUFFICIENT"], quota_write_count=1, ), evidence_root=root, evidence_paths=[evidence("execution_context", context_path), evidence("quota_ledger", quota_path)], ) # Caller-only discovery/candidate/quota claims cannot create a package. root = (parent / "counter-lie").resolve() context, context_path = prepare(root) with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal( root, context, discovery_completed=7, qualified_candidate_total=999, quota_write_count=999, ), evidence_root=root, evidence_paths=[evidence("execution_context", context_path)], ) # Arbitrary files cannot be laundered into the PREPLAN manifest. root = (parent / "arbitrary-evidence").resolve() context, context_path = prepare(root) arbitrary = root / "control" / "arbitrary.bin" arbitrary.write_bytes(b"not-reviewed-evidence") with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal(root, context), evidence_root=root, evidence_paths=[evidence("execution_context", context_path), evidence("arbitrary", arbitrary)], ) root = (parent / "status-lie").resolve() context, context_path = prepare(root) with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal(root, context), evidence_root=root, evidence_paths=[evidence("execution_context", context_path, "CALLER_SAYS_OK")], ) root = (parent / "unsafe-presence-on-known-stop").resolve() context, context_path = prepare(root) with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal(root, context, plan_presence="U"), evidence_root=root, evidence_paths=[evidence("execution_context", context_path)], ) # A valid physical 10-row plan cannot be relabelled as plan_presence=I. root = (parent / "valid-plan-lie").resolve() context, context_path = prepare(root) queries = ["三环集团", "三环集团", "国瓷材料", "国瓷材料", "MLCC", "MLCC", "MLCC", "MLCC", "MLCC", "MLCC"] entries = [] for index, query in enumerate(queries, 1): entry_values = { "slot_id": f"PERF-{index:02d}", "mode": "batch" if index >= 7 else "collect-one", "batch_id": "PERF-B01" if index >= 7 else None, "item_order": index - 6 if index >= 7 else 1, "query": query, "expected_report": report(f"{query} 计划研报 {index}"), "discovery_sha256": hashlib.sha256(f"D{index}".encode()).hexdigest(), "external_exclusion_allowed": True, "selected_from_readonly_discovery": True, } entries.append(OrderedDict( (key, entry_values[key]) for key in PLAN_ENTRY_KEYS )) plan_values = { "schema_version": "HIBOR_FAST_PERFORMANCE_PLAN_V002", "plan_id": "PERF-PLAN-VALID-LIE", "created_at_utc": "2026-07-30T00:00:30Z", "quota_date": "2026-07-30", "quota_ledger": str(root / "daily_quota_2026-07-30.csv"), "historical_quota_date": "2026-07-29", "historical_quota_ledger": str(root / "daily_quota_2026-07-29.csv"), "trigger_cap": 10, "entries": entries, } plan = OrderedDict((key, plan_values[key]) for key in PLAN_KEYS) plan_path = root / "control" / "performance_plan.json" plan_receipt = write_performance_plan(plan_path, plan) with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal( root, context, stop_code="PLAN_CREATE_FAILED", blockers=["PLAN_CREATE_FAILED"], plan_presence="I", plan_bytes=plan_receipt[0], plan_sha256=plan_receipt[1], ), evidence_root=root, evidence_paths=[evidence("execution_context", context_path), evidence("performance_plan", plan_path, "INVALID")], ) with self.assertRaises(ContractError): write_preplan_package( terminal_path=root / "preplan" / "preplan_terminal.json", manifest_path=root / "preplan" / "preplan_manifest.csv", terminal_value=terminal( root, context, stop_code="STATE_UNCERTAIN", exit_code=27, blockers=["STATE_UNCERTAIN"], plan_presence="U", ), evidence_root=root, evidence_paths=[evidence("execution_context", context_path)], ) if __name__ == "__main__": unittest.main()