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()
|
