from __future__ import annotations import csv import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch import hibor_fast_collection.workflow as workflow_module from hibor_fast_collection.adb import RemoteFile from hibor_fast_collection.budget import Budget from hibor_fast_collection.cache import CacheMatch from hibor_fast_collection.models import Candidate, ContractError, TaskSpec from hibor_fast_collection.quota import AppQuotaObservation, QuotaLedger from hibor_fast_collection.selection import ScanOutcome from hibor_fast_collection.workflow import HiborWorkflow, _identity from helpers import task_spec PDF = b"%PDF-1.4\n1 0 obj<<>>endobj\ntrailer<<>>\n%%EOF\n" class FakeAdb: def __init__(self): self.triggered = False self.remote = RemoteFile( "/sdcard/Android/data/cn.com.hibor/files/myfile/report.pdf", "report.pdf", len(PDF), "2026-07-29 10:00", ) def preflight(self): return object() def screenshot(self, target: Path): target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"\x89PNG\r\n\x1a\nFAKE") def list_cache(self): return (self.remote,) if self.triggered else () def remote_sha256(self, remote_path: str): self.assert_path(remote_path) return hashlib.sha256(PDF).hexdigest() def pull(self, remote_path: str, local_staging: Path, *, timeout_ms: int): self.assert_path(remote_path) local_staging.parent.mkdir(parents=True, exist_ok=True) local_staging.write_bytes(PDF) return object() def assert_path(self, value: str): if value != self.remote.path: raise AssertionError(value) class FakeUi: def __init__(self, adb: FakeAdb): self.adb = adb self.candidate = Candidate("CANDIDATE-001", "中国中免深度研究", None, None, score=120, bounds=(1, 1, 2, 2)) def search(self, query: str): self.query = query def scan(self, **kwargs): return ScanOutcome("ENOUGH_CONFIRMED", (self.candidate,), 3, 1, 0) def open_detail(self, candidate: Candidate): return Candidate(candidate.candidate_id, candidate.title, "测试证券", "2026-07-29", ("分析师甲",), 1, candidate.score, candidate.bounds) def trigger_current_detail(self): self.adb.triggered = True self.trigger_calls = getattr(self, "trigger_calls", 0) + 1 class WorkflowFakeE2ETests(unittest.TestCase): def test_one_report_complete_without_external_process(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) adb = FakeAdb() result = HiborWorkflow(spec).collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) self.assertEqual((result["status"], result["exit_code"]), ("SUCCESS", 0)) self.assertEqual((result["triggered"], result["succeeded"]), (1, 1)) self.assertEqual(len(result["items"]), 1) self.assertEqual(result["items"][0]["sha256"], hashlib.sha256(PDF).hexdigest()) self.assertTrue(Path(result["manifest_path"]).is_file()) self.assertTrue(Path(result["delivery_path"]).is_file()) self.assertTrue(Path(result["timing_path"]).is_file()) self.assertTrue(Path(result["terminal_path"]).is_file()) parsed = json.loads(Path(result["terminal_path"]).read_text(encoding="utf-8")) self.assertEqual(parsed["status"], "SUCCESS") ledger = spec.quota_path.read_text(encoding="utf-8") self.assertIn("BASELINE_ESTIMATE", ledger) self.assertIn("CONSUME_CONFIRMED", ledger) self.assertIn("ARTIFACT_SUCCESS", ledger) def test_quota_exhaustion_stops_before_trigger_and_writes_failure_manifest(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) QuotaLedger(spec.quota_path).initialize( task_id=spec.task_id, requester_role=spec.requester, handoff_id=spec.handoff_id, known_floor=25, evidence_ref="TEST-FLOOR-25", ) adb = FakeAdb() result = HiborWorkflow(spec).collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) self.assertEqual(result["status"], "PARTIAL_QUOTA_STOP") self.assertEqual((result["triggered"], adb.triggered), (0, False)) self.assertEqual(result["items"][0]["stop_code"], "QUOTA_EXHAUSTED") self.assertTrue(Path(result["manifest_path"]).is_file()) def test_cache_timeout_marks_consumption_uncertain(self): class ImmediateTimeout: def __init__(self, list_files): self.list_files = list_files def baseline(self): return {} def wait_for_unique_stable(self, baseline, *, timeout_ms): return CacheMatch("NONE_TIMEOUT", None, 1, ()) with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) adb = FakeAdb() result = HiborWorkflow(spec).collect( adb=adb, ui=FakeUi(adb), execute=True, synthetic=True, watcher_factory=ImmediateTimeout, ) self.assertEqual((result["status"], result["exit_code"]), ("STATE_UNCERTAIN", 27)) self.assertEqual((result["triggered"], result["quota_uncertain"]), (0, 1)) self.assertEqual(result["items"][0]["quota_state"], "UNCERTAIN") self.assertTrue(Path(result["manifest_path"]).is_file()) def test_confirmed_quota_survives_post_confirm_validation_failure(self): class BadRemoteHashAdb(FakeAdb): def remote_sha256(self, remote_path: str): self.assert_path(remote_path) return "0" * 64 with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) adb = BadRemoteHashAdb() result = HiborWorkflow(spec).collect( adb=adb, ui=FakeUi(adb), execute=True, synthetic=True, ) item = result["items"][0] self.assertEqual((item["quota_state"], item["stop_code"]), ("CONFIRMED", "HASH_MISMATCH")) self.assertIsNotNone(item["quota_terminal_event_id"]) self.assertIsNotNone(item["quota_artifact_event_id"]) ledger = spec.quota_path.read_text(encoding="utf-8") self.assertIn("CONSUME_CONFIRMED", ledger) self.assertIn("ARTIFACT_DUPLICATE_OR_FAILED", ledger) self.assertNotIn(",RELEASE,", ledger) with Path(result["manifest_path"]).open("r", encoding="utf-8", newline="") as stream: row = next(csv.DictReader(stream)) self.assertEqual(row["quota_terminal_event_id"], item["quota_terminal_event_id"]) self.assertEqual(row["quota_artifact_event_id"], item["quota_artifact_event_id"]) self.assertEqual((row["bytes"], row["sha256"], row["relative_path"]), ("", "", "")) def test_artifact_success_facts_survive_item_close_deadline(self): class Clock: value = 0.0 def __call__(self): return self.value clock = Clock() class AdvanceAfterArtifactQuota(QuotaLedger): def artifact(self, **kwargs): event = super().artifact(**kwargs) clock.value = 600.001 return event with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="batch", quantity=2) raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) budget = Budget(spec.total_budget_ms, spec.close_reserve_ms, batch_increment_ms=spec.batch_increment_budget_ms, clock=clock) adb = FakeAdb() with patch.object(workflow_module, "QuotaLedger", AdvanceAfterArtifactQuota): result = HiborWorkflow(spec, budget=budget).collect( adb=adb, ui=FakeUi(adb), execute=True, synthetic=True, ) item = result["items"][0] expected_sha = hashlib.sha256(PDF).hexdigest() self.assertEqual((result["status"], item["stop_code"], item["quota_state"]), ("TIME_BUDGET_STOP", "DEADLINE_EXPIRED", "CONFIRMED"), result) self.assertEqual((item["bytes"], item["sha256"]), (len(PDF), expected_sha)) self.assertTrue(Path(item["final_path"]).is_file()) self.assertIsNotNone(item["quota_artifact_event_id"]) ledger = spec.quota_path.read_text(encoding="utf-8") self.assertEqual(ledger.count("ARTIFACT_SUCCESS"), 1) self.assertNotIn("ARTIFACT_DUPLICATE_OR_FAILED", ledger) with Path(result["manifest_path"]).open("r", encoding="utf-8", newline="") as stream: row = next(csv.DictReader(stream)) self.assertEqual((row["download_status"], row["status"], row["stop_code"]), ("SUCCESS", "SUCCESS", "DEADLINE_EXPIRED")) self.assertEqual((int(row["bytes"]), row["sha256"], row["relative_path"]), (len(PDF), expected_sha, item["final_path"])) self.assertEqual(row["quota_artifact_event_id"], item["quota_artifact_event_id"]) def test_execution_requires_explicit_gate(self): with tempfile.TemporaryDirectory() as temp: spec = TaskSpec.from_mapping(task_spec(Path(temp), mode="collect-one")) with self.assertRaisesRegex(Exception, "explicit True"): HiborWorkflow(spec).collect(adb=object(), ui=object()) def test_hard_filter_rejects_excluded_title(self): with tempfile.TemporaryDirectory() as temp: raw = task_spec(Path(temp), mode="collect-one") raw["exclude"] = ["禁止词"] spec = TaskSpec.from_mapping(raw) candidate = Candidate("C", "中国中免禁止词报告", "测试证券", "2026-07-29", ("分析师甲",), 8, 120, (1, 1, 2, 2)) with self.assertRaises(ContractError): HiborWorkflow(spec)._enforce_hard_filters(candidate) def test_completed_request_replay_never_taps_or_mutates_ledger(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) adb = FakeAdb() ui = FakeUi(adb) first = HiborWorkflow(spec).collect(adb=adb, ui=ui, execute=True, synthetic=True) ledger_before = spec.quota_path.read_bytes() second_adb = FakeAdb() second_ui = FakeUi(second_adb) second = HiborWorkflow(spec).collect(adb=second_adb, ui=second_ui, execute=True, synthetic=True) self.assertEqual((first["run_id"], second["run_id"]), (second["run_id"], second["run_id"])) self.assertFalse(second_adb.triggered) self.assertFalse(hasattr(second_ui, "trigger_calls")) self.assertEqual(spec.quota_path.read_bytes(), ledger_before) def test_project_pure_reuse_passes_at_safe_zero_without_adb_or_ledger_write(self): class NoAdb: def set_timeout_provider(self, provider): self.provider = provider def preflight(self): raise AssertionError("pure project reuse must not use ADB") with tempfile.TemporaryDirectory() as temp: root = Path(temp) first_raw = task_spec(root, mode="collect-one") first_raw["destination"] = str(root / "archive") first_spec = TaskSpec.from_mapping(first_raw) adb = FakeAdb() first = HiborWorkflow(first_spec).collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) quota_date = QuotaLedger(first_spec.quota_path).snapshot().quota_date ledger = QuotaLedger(first_spec.quota_path, quota_date=quota_date) observation = AppQuotaObservation.build( device_serial="emulator-5554", quota_date=ledger.quota_date, captured_at_utc="2026-07-29T02:00:00Z", visible_remaining=0, ui_snapshot_fingerprint="a" * 64, ) ledger.observe_app_remaining(task_id="OBS", requester_role="R", handoff_id="H", observation=observation) before = first_spec.quota_path.read_bytes() second_raw = task_spec(root, mode="collect-one") second_raw["task_id"] = "TASK-TEST-REUSE" second_raw["handoff_id"] = "HANDOFF-TEST-REUSE" second_raw["performance_slot_id"] = "SLOT-REUSE" second_raw["destination"] = str(root / "archive") second_spec = TaskSpec.from_mapping(second_raw) result = HiborWorkflow(second_spec).collect( adb=NoAdb(), ui=object(), execute=True, synthetic=True, ) self.assertEqual((result["status"], result["triggered"], result["succeeded"]), ("SUCCESS", 0, 1)) self.assertTrue(result["items"][0]["reused_without_new_trigger"]) self.assertEqual(result["quota_safe_available"], 0) self.assertEqual(second_spec.quota_path.read_bytes(), before) def test_manifest_conflict_is_not_exposed_as_verified_path(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) workflow = HiborWorkflow(spec) conflict = spec.output_path / workflow._run_id() / "manifest.csv" conflict.parent.mkdir(parents=True) conflict.write_bytes(b"CONFLICT") adb = FakeAdb() result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) self.assertNotEqual(result["status"], "SUCCESS") self.assertIsNone(result["manifest_path"]) persisted = json.loads(Path(result["terminal_path"]).read_text(encoding="utf-8")) self.assertIsNone(persisted["manifest_path"]) def test_delivery_and_timing_conflicts_preserve_only_verified_prefix(self): for conflict_name, expected in (("delivery.md", (True, False, False)), ("timing.json", (True, True, False))): with self.subTest(conflict_name=conflict_name), tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["task_id"] = f"TASK-{conflict_name}" raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) workflow = HiborWorkflow(spec) run_root = spec.output_path / workflow._run_id() path = run_root / conflict_name path.parent.mkdir(parents=True) path.write_bytes(b"CONFLICT") adb = FakeAdb() result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) actual = (result["manifest_path"] is not None, result["delivery_path"] is not None, result["timing_path"] is not None) self.assertEqual(actual, expected) self.assertNotEqual(result["status"], "SUCCESS") def test_existing_invalid_terminal_stops_before_adb_and_trigger(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="collect-one") raw["destination"] = str(root / "archive") spec = TaskSpec.from_mapping(raw) workflow = HiborWorkflow(spec) terminal = spec.output_path / workflow._run_id() / "report_collection_terminal.json" terminal.parent.mkdir(parents=True) terminal.write_bytes(b"CONFLICT") adb = FakeAdb() result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) self.assertEqual(result["terminal_presence"], "I") self.assertEqual(result["status"], "INTERNAL_ERROR") self.assertFalse(adb.triggered) def test_expired_work_budget_emits_terminal_before_adb(self): class Clock: value = 0.0 def __call__(self): return self.value with tempfile.TemporaryDirectory() as temp: root = Path(temp) spec = TaskSpec.from_mapping(task_spec(root, mode="collect-one")) clock = Clock() budget = Budget(600, 100, clock=clock) workflow = HiborWorkflow(spec, budget=budget) clock.value = 0.51 adb = FakeAdb() result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) self.assertEqual((result["status"], result["stop_code"]), ("TIME_BUDGET_STOP", "DEADLINE_EXPIRED")) self.assertFalse(adb.triggered) def test_expired_batch_close_returns_in_memory_t00b_without_raise(self): class Clock: value = 0.0 def __call__(self): return self.value with tempfile.TemporaryDirectory() as temp: root = Path(temp) spec = TaskSpec.from_mapping(task_spec(root, mode="collect-one")) clock = Clock() budget = Budget(600, 100, clock=clock) workflow = HiborWorkflow(spec, budget=budget) clock.value = 0.601 adb = FakeAdb() result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True) self.assertEqual((result["status"], result["capability_status"], result["exit_code"], result["stop_code"]), ("TIME_BUDGET_STOP", "TIME_BUDGET_STOP", 10, "DEADLINE_EXPIRED")) self.assertEqual((result["terminal_presence"], result["terminal_persist_attempted"]), ("N", False)) self.assertFalse(adb.triggered) def test_recorded_cache_pure_reuse_does_not_touch_quota_or_ui(self): class CacheAdb(FakeAdb): def set_timeout_provider(self, provider): self.provider = provider def list_cache(self): return (self.remote,) with tempfile.TemporaryDirectory() as temp: root = Path(temp) raw = task_spec(root, mode="resume-postprocess") raw["destination"] = str(root / "archive") candidate = Candidate("CANDIDATE-001", f"{raw['query']} FAST REPORT", "娴嬭瘯璇佸埜", "2026-07-29", ("鍒嗘瀽甯堢敳",), 1) remote_path = "/sdcard/Android/data/cn.com.hibor/files/myfile/report.pdf" raw["source_scope"] = {"resume_items": [{ "candidate_id": candidate.candidate_id, "report_identity": _identity(candidate), "title": candidate.title, "institution": candidate.institution, "report_date": candidate.report_date, "analysts": list(candidate.analysts), "page_count": candidate.page_count, "remote_path": remote_path, "remote_bytes": len(PDF), "remote_sha256": hashlib.sha256(PDF).hexdigest(), }]} spec = TaskSpec.from_mapping(raw) adb = CacheAdb() result = HiborWorkflow(spec).collect(adb=adb, ui=object(), execute=True, synthetic=True) self.assertEqual((result["status"], result["triggered"], result["succeeded"]), ("SUCCESS", 0, 1)) self.assertTrue(result["items"][0]["reused_without_new_trigger"]) self.assertFalse(spec.quota_path.exists()) def test_mixed_batch_keeps_reuse_then_quota_stops_new_item_without_tap(self): class NewCandidateUi(FakeUi): def __init__(self, adb): super().__init__(adb) self.candidate = Candidate("CANDIDATE-NEW", f"{self.query if hasattr(self, 'query') else ''} NEW", None, None, score=120, bounds=(1, 1, 2, 2)) def scan(self, **kwargs): title = f"{kwargs['query']} NEW REPORT" self.candidate = Candidate("CANDIDATE-NEW", title, None, None, score=120, bounds=(1, 1, 2, 2)) return ScanOutcome("ENOUGH_CONFIRMED", (self.candidate,), 3, 1, 0) def open_detail(self, candidate): return Candidate(candidate.candidate_id, candidate.title, "NEW-INSTITUTION", "2026-07-29", ("NEW-ANALYST",), 1, candidate.score, candidate.bounds) with tempfile.TemporaryDirectory() as temp: root = Path(temp) first_raw = task_spec(root, mode="collect-one") first_raw["destination"] = str(root / "archive") first_spec = TaskSpec.from_mapping(first_raw) first_adb = FakeAdb() HiborWorkflow(first_spec).collect(adb=first_adb, ui=FakeUi(first_adb), execute=True, synthetic=True) quota_date = QuotaLedger(first_spec.quota_path).snapshot().quota_date ledger = QuotaLedger(first_spec.quota_path, quota_date=quota_date) observation = AppQuotaObservation.build( device_serial="emulator-5554", quota_date=quota_date, captured_at_utc="2026-07-29T03:00:00Z", visible_remaining=0, ui_snapshot_fingerprint="b" * 64, ) ledger.observe_app_remaining(task_id="OBS", requester_role="R", handoff_id="H", observation=observation) batch_raw = task_spec(root, mode="batch", quantity=2) batch_raw["task_id"] = "TASK-MIXED" batch_raw["handoff_id"] = "HANDOFF-MIXED" batch_raw["destination"] = str(root / "archive") batch_spec = TaskSpec.from_mapping(batch_raw) adb = FakeAdb() result = HiborWorkflow(batch_spec).collect(adb=adb, ui=NewCandidateUi(adb), execute=True, synthetic=True) self.assertEqual((result["succeeded"], result["triggered"]), (1, 0)) self.assertIn(result["status"], {"PARTIAL_QUOTA_STOP", "PARTIAL_SUCCESS"}) self.assertFalse(adb.triggered) if __name__ == "__main__": unittest.main()