from __future__ import annotations from dataclasses import dataclass, field, replace from datetime import datetime, timezone import csv import hashlib import json from pathlib import Path import re import time from typing import Any, Callable, Mapping from zoneinfo import ZoneInfo from .archive import publish_no_replace, sha256_file, validate_pdf from .budget import Budget from .cache import CacheWatcher from .items import terminal_item from .manifests import (MANIFEST_COLUMNS, utc_now, write_bytes_create_new, write_json_create_new, write_manifest_create_new) from .models import ( Candidate, ContractError, ErrorCode, EvidenceState, ItemStatus, PackageState, QuotaState, TaskSpec, TerminalReceipt, TerminalStatus, ) from .quota import AppQuotaObservation, QuotaLedger, QuotaSnapshot, Reservation from .selection import score_candidate from .terminal import TerminalWriter, build_terminal SOURCE_SITE = "慧博 APP（安卓模拟器本地缓存）" @dataclass class StageTimings: rows: list[dict[str, Any]] = field(default_factory=list) def record(self, stage: str, start_ns: int, end_ns: int, status: str) -> None: self.rows.append({ "stage": stage, "started_monotonic_ns": start_ns, "ended_monotonic_ns": end_ns, "elapsed_ms": (end_ns - start_ns) // 1_000_000, "status": status, }) def _identity(candidate: Candidate) -> str: text = "|".join((candidate.title, candidate.institution or "", candidate.report_date or "")) return hashlib.sha256(text.encode("utf-8")).hexdigest() def _safe_part(value: str, maximum: int = 60) -> str: value = re.sub(r"[<>:\"/\\|?*\x00-\x1f]", "_", value).strip(" ._") return (value or "UNKNOWN")[:maximum] def _manifest_row(**values: Any) -> dict[str, Any]: return {key: values.get(key) for key in MANIFEST_COLUMNS} class HiborWorkflow: def __init__(self, spec: TaskSpec, *, budget: Budget | None = None, monotonic_ns: Callable[[], int] = time.monotonic_ns, utc_now_provider: Callable[[], datetime] | None = None): self.spec = spec self.budget = budget or Budget( spec.total_budget_ms, spec.close_reserve_ms, observed_at_utc=spec.observed_at_utc, batch_increment_ms=spec.batch_increment_budget_ms, ) self._monotonic_ns = monotonic_ns self._utc_now = utc_now_provider or (lambda: datetime.now(timezone.utc)) self.timings = StageTimings() self.item_windows: list[dict[str, Any]] = [] self._open_item_windows: dict[int, dict[str, Any]] = {} self.started_at = self._utc_now().isoformat().replace("+00:00", "Z") def dry_run(self) -> dict[str, Any]: """Read-only environment check. It never starts ADB or creates a quota ledger.""" start = self._monotonic_ns() tool_checks = { "adb_executable_exists": Path(self.spec.adb_executable).is_file(), "pdfinfo_executable_exists": Path(self.spec.pdfinfo_executable).is_file(), "output_parent_exists": self.spec.output_path.parent.is_dir(), "quota_ledger_exists": self.spec.quota_path.is_file(), } status = "PASS" if all(v for k, v in tool_checks.items() if k != "quota_ledger_exists") else "BLOCKED" self.timings.record("dry_run", start, self._monotonic_ns(), status) quota = self._quota_snapshot_readonly() target = self.spec.output_path / "terminal" / f"{self.spec.task_id}.json" receipt = TerminalReceipt(target, None, False, False, EvidenceState.N, False, False, True, None, None, None, PackageState.P00_INIT) base = self._base_terminal(quota, run_id=f"DRY-{self.spec.task_id}") if status == "PASS": return build_terminal(base, receipt, status=TerminalStatus.SUCCESS, stop_code=None) return build_terminal(base, receipt, status=TerminalStatus.BLOCKED_ENVIRONMENT, stop_code=ErrorCode.PROCESS_START_FAILED, blocker="LOCAL_TOOL_PREFLIGHT") def discover(self, *, adb: Any, ui: Any, execute: bool = False) -> dict[str, Any]: """Discover and verify exact report identities without quota or download actions. This entry may navigate public APP result/detail screens, but it never calls the download trigger, creates a quota ledger, pulls cache files, or publishes artifacts. Its canonical stdout is the input to the separately frozen performance plan. """ if execute is not True: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execute", "explicit True required") if self.spec.mode != "discover": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "mode", "discover required") if hasattr(adb, "set_timeout_provider"): adb.set_timeout_provider( lambda operation, maximum: self.budget.require("discovery", operation, maximum) ) started_ns = self._monotonic_ns() self.budget.checkpoint("discovery", "adb_preflight") adb.preflight() self.budget.checkpoint("discovery", "search") ui.search(self.spec.query) scan = ui.scan( query=self.spec.query, quantity=self.spec.quantity, aliases=tuple(self.spec.aliases), institutions=tuple(self.spec.institutions), minimum_pages=self.spec.minimum_pages, min_screens=self.spec.min_screens, normal_max_screens=self.spec.normal_max_screens, hard_max_screens=self.spec.hard_max_screens, hard_max_candidates=self.spec.hard_max_candidates, ) candidates: list[dict[str, Any]] = [] seen: set[str] = set() for candidate in scan.candidates: if len(candidates) >= self.spec.quantity: break self.budget.checkpoint("discovery", "detail") ui.search(self.spec.query) detailed = ui.open_detail(candidate) detailed = replace(detailed, score=score_candidate( detailed, query=self.spec.query, aliases=tuple(self.spec.aliases), institutions=tuple(self.spec.institutions), minimum_pages=self.spec.minimum_pages, )) self._enforce_hard_filters(detailed) identity = _identity(detailed) if identity in seen: continue seen.add(identity) candidates.append({ "report_identity": identity, "title": detailed.title, "institution": detailed.institution, "report_date": detailed.report_date, "page_count": detailed.page_count, }) if len(candidates) < self.spec.quantity: raise ContractError(ErrorCode.CANDIDATE_REJECTED, "discovery", "insufficient exact candidates") ended_ns = self._monotonic_ns() return { "schema_version": "HIBOR_FAST_DISCOVERY_V001", "task_id": self.spec.task_id, "handoff_id": self.spec.handoff_id, "performance_plan_id": self.spec.performance_plan_id, "performance_slot_id": self.spec.performance_slot_id, "query": self.spec.query, "observed_at_utc": self.spec.observed_at_utc, "completed_at_utc": self._utc_now().isoformat().replace("+00:00", "Z"), "started_monotonic_ns": started_ns, "ended_monotonic_ns": ended_ns, "screens_scanned": scan.screens_scanned, "candidate_count": len(candidates), "candidates": candidates, "quota_ledger_touched": False, "trigger_attempted": False, } def collect(self, *, adb: Any, ui: Any, execute: bool = False, synthetic: bool = False, watcher_factory: Callable[[Callable[[], Any]], CacheWatcher] = CacheWatcher) -> dict[str, Any]: """Execute one bounded collection run using injected ADB/UI adapters. ``execute`` is deliberately mandatory so importing this module cannot trigger APP work. ``synthetic`` skips the external ``pdfinfo`` process and is accepted only by local tests. """ if execute is not True: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execute", "explicit True required") if self.spec.mode not in {"collect-one", "batch", "resume-postprocess"}: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "mode", "collect-one, batch or resume-postprocess required") run_id = self._run_id() run_root = self.spec.output_path / run_id screenshot_root = run_root / "screenshots" staging_root = run_root / "staging" manifest_path = run_root / "manifest.csv" delivery_path = run_root / "delivery.md" timing_path = run_root / "timing.json" terminal_path = run_root / "report_collection_terminal.json" items: list[dict[str, Any]] = [] manifest_rows: list[dict[str, Any]] = [] triggered = 0 stop_code: ErrorCode | None = None blocker: str | None = None terminal_status = TerminalStatus.SUCCESS quota: QuotaLedger | None = None local_date = datetime.now(ZoneInfo("Asia/Shanghai")).date().isoformat() known_floor = 3 if local_date == "2026-07-29" else 0 quota_snapshot = QuotaSnapshot(local_date, 0, 0, 0, known_floor, 0, known_floor, 25 - known_floor, 0) package_state = PackageState.P00_INIT manifest_verified = False delivery_verified = False timing_verified = False try: # Bind the first SLA window before replay/reuse discovery so even a # zero-trigger result must close within observed_at+600s. self._begin_item(1) # A fully persisted request is an immutable replay. Returning it before # ADB, APP or quota access is the primary duplicate-trigger guard. self.budget.checkpoint("replay", "terminal_read") replay = TerminalWriter().read_valid( terminal_path, task_id=self.spec.task_id, handoff_id=self.spec.handoff_id, run_id=run_id, ) if replay is not None: return replay quota_snapshot = self._quota_snapshot_readonly() if hasattr(adb, "set_timeout_provider"): adb.set_timeout_provider( lambda operation, maximum: self.budget.require("external", operation, maximum) ) reuse_rows = self._discover_project_reuse(synthetic=synthetic) if len(reuse_rows) >= self.spec.quantity: self._close_reuse_item_windows(self.spec.quantity) return self._close_project_reuse( run_id=run_id, run_root=run_root, manifest_path=manifest_path, delivery_path=delivery_path, timing_path=timing_path, terminal_path=terminal_path, rows=reuse_rows[:self.spec.quantity], ) if reuse_rows: reuse_manifest, reuse_items = self._build_reuse_payload(run_id, reuse_rows) manifest_rows.extend(reuse_manifest) items.extend(reuse_items) self._close_reuse_item_windows(len(reuse_items)) if self.spec.mode == "resume-postprocess": cache_rows = self._materialize_cache_reuse( adb=adb, run_root=run_root, synthetic=synthetic, ) if len(cache_rows) < self.spec.quantity: raise ContractError(ErrorCode.CANDIDATE_REJECTED, "resume_items", "insufficient verified cache evidence") return self._close_project_reuse( run_id=run_id, run_root=run_root, manifest_path=manifest_path, delivery_path=delivery_path, timing_path=timing_path, terminal_path=terminal_path, rows=cache_rows[:self.spec.quantity], ) except ContractError as exc: return self._preflight_terminal(run_id, terminal_path, quota_snapshot, exc) try: stage = self._monotonic_ns() if self.budget.work_deadline_reached: raise ContractError(ErrorCode.DEADLINE_EXPIRED, "preflight", "work deadline") self.budget.checkpoint("preflight", "adb_preflight") adb.preflight() self.budget.checkpoint("preflight", "adb_preflight_complete") if not synthetic: if not Path(self.spec.pdfinfo_executable).is_file(): raise ContractError(ErrorCode.PROCESS_START_FAILED, "pdfinfo", "executable absent") quota_date = datetime.now(ZoneInfo("Asia/Shanghai")).date().isoformat() quota = QuotaLedger( self.spec.quota_path, quota_date=quota_date, timeout_provider=lambda operation, maximum: self.budget.require( "quota", operation, maximum, close=self.budget.work_deadline_reached), ) known_floor = 3 if quota_date == "2026-07-29" else 0 self.budget.checkpoint("quota", "initialize") quota_snapshot = quota.initialize( task_id=self.spec.task_id, requester_role=self.spec.requester, handoff_id=self.spec.handoff_id, known_floor=known_floor, evidence_ref="KNOWN-2026-07-29-CHINA-DUTYFREE-3" if known_floor else "NO_KNOWN_EXTERNAL_BASELINE", ) adb.screenshot(screenshot_root / "01-start.png") self.timings.record("preflight", stage, self._monotonic_ns(), "PASS") stage = self._monotonic_ns() ui.search(self.spec.query) app_observation_event_id = "" if hasattr(ui, "observe_app_remaining"): observation = ui.observe_app_remaining(quota_date) if observation is not None: if not isinstance(observation, AppQuotaObservation): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "app_observation", "type") self.budget.checkpoint("quota", "app_reconcile") app_event = quota.observe_app_remaining( task_id=self.spec.task_id, requester_role=self.spec.requester, handoff_id=self.spec.handoff_id, observation=observation, ) app_observation_event_id = app_event["event_id"] quota_snapshot = quota.snapshot() needed = self.spec.quantity - len(items) scan = ui.scan( query=self.spec.query, quantity=needed, aliases=tuple(self.spec.aliases), institutions=tuple(self.spec.institutions), minimum_pages=self.spec.minimum_pages, min_screens=self.spec.min_screens, normal_max_screens=self.spec.normal_max_screens, hard_max_screens=self.spec.hard_max_screens, hard_max_candidates=self.spec.hard_max_candidates, ) adb.screenshot(screenshot_root / "02-results.png") expected_slice = self.spec.expected_reports[len(items):] candidates = [] for expected_report in expected_slice: matches = [candidate for candidate in scan.candidates if candidate.title == expected_report["title"]] if len(matches) != 1: if candidates: break raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "expected_report.title", "not uniquely present in discovery scan") candidates.append(matches[0]) if not candidates: raise ContractError(ErrorCode.CANDIDATE_REJECTED, "scan", "no candidate") self.timings.record("ui_scan", stage, self._monotonic_ns(), scan.status) reuse_count = len(items) for new_offset, candidate in enumerate(candidates, 1): index = reuse_count + new_offset self._begin_item(index) if self.budget.work_deadline_reached: stop_code = ErrorCode.DEADLINE_EXPIRED terminal_status = TerminalStatus.TIME_BUDGET_STOP blocker = "WORK_DEADLINE_REACHED" break slot_id = f"SLOT-{index:03d}" item_id = f"ITEM-{index:03d}" reservation: Reservation | None = None quota_terminal: dict[str, str] | None = None action_attempted = False trigger_confirmed = False item_terminal_recorded = False detailed = candidate report_identity: str | None = None source_state = PackageState.P00_INIT match = None remote_sha: str | None = None published = None artifact_event: dict[str, str] | None = None try: stage = self._monotonic_ns() expected_report = self.spec.expected_reports[index - 1] if new_offset > 1: ui.search(self.spec.query) scan = ui.scan( query=self.spec.query, quantity=new_offset, aliases=tuple(self.spec.aliases), institutions=tuple(self.spec.institutions), minimum_pages=self.spec.minimum_pages, min_screens=self.spec.min_screens, normal_max_screens=self.spec.normal_max_screens, hard_max_screens=self.spec.hard_max_screens, hard_max_candidates=self.spec.hard_max_candidates, ) matches = [row for row in scan.candidates if row.title == expected_report["title"]] if len(matches) != 1: raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "expected_report.title", "not uniquely relocatable") candidate = matches[0] # The scanner leaves the viewport at its final page. Restart the query so # the UI adapter can deterministically relocate the exact title before tap. ui.search(self.spec.query) detailed = ui.open_detail(candidate) detailed = replace(detailed, score=score_candidate( detailed, query=self.spec.query, aliases=tuple(self.spec.aliases), institutions=tuple(self.spec.institutions), minimum_pages=self.spec.minimum_pages, )) report_identity = _identity(detailed) self._enforce_hard_filters(detailed) actual_report = { "report_identity": report_identity, "title": detailed.title, "institution": detailed.institution, "report_date": detailed.report_date, "page_count": detailed.page_count, } if actual_report != expected_report: raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "expected_report", "detail identity/fields drift") if any(item.get("report_identity") == report_identity for item in items): self.timings.record(f"item_{index}_detail", stage, self._monotonic_ns(), "SKIP_ALREADY_REUSED") item_terminal_recorded = True if not self._complete_item(index): raise ContractError(ErrorCode.DEADLINE_EXPIRED, f"item_{index}", "item_close") continue adb.screenshot(screenshot_root / f"{index + 2:02d}-detail.png") self.timings.record(f"item_{index}_detail", stage, self._monotonic_ns(), "PASS") self.budget.checkpoint(f"item_{index}", "quota_reserve") reservation = quota.reserve( task_id=self.spec.task_id, requester_role=self.spec.requester, handoff_id=self.spec.handoff_id, run_id=run_id, slot_id=slot_id, report_identity=report_identity, observation_event_id=app_observation_event_id, ) quota_snapshot = reservation.snapshot if not reservation.allowed: replay_block = reservation.replayed stop_code = ErrorCode.QUOTA_EXHAUSTED if replay_block: stop_code = ErrorCode.QUOTA_REPLAY_CONFLICT terminal_status = (self._status_for_error(stop_code, bool(items)) if replay_block else TerminalStatus.PARTIAL_QUOTA_STOP) blocker = "TERMINALLED_OR_ACTIVE_RESERVATION_REPLAY" if replay_block else "DAILY_SAFE_QUOTA_EXHAUSTED" item_terminal_recorded = True if not self._complete_item(index): raise ContractError(ErrorCode.DEADLINE_EXPIRED, f"item_{index}", "item_close") row = self._failure_manifest_row( run_id=run_id, item_id=item_id, slot_id=slot_id, candidate=detailed, report_identity=report_identity, code=stop_code, note=blocker, reservation=None, quota_terminal=None, artifact_event_id=None, action_attempted=False, ) manifest_rows.append(row) items.append(terminal_item({ "item_id": item_id, "slot_id": slot_id, "candidate_id": detailed.candidate_id, "report_identity": report_identity, "state": PackageState.P00_INIT.value, "status": ItemStatus.STOPPED.value, "stop_code": stop_code.value, "trigger_attempted": False, "triggered": False, "quota_state": QuotaState.NONE.value, "quota_reservation_id": None, "quota_terminal_event_id": None, "quota_artifact_event_id": None, "title": detailed.title, "institution": detailed.institution, "report_date": detailed.report_date, "analysts": list(detailed.analysts), "page_count": detailed.page_count, "bytes": None, "sha256": None, "source_cache_path": None, "source_file_name": None, "final_path": None, "manifest_row_id": row["row_id"], "reused_without_new_trigger": False, "error_or_note": blocker, })) break source_state = PackageState.P01_RESERVED watcher = watcher_factory(adb.list_cache) self.budget.checkpoint(f"item_{index}", "cache_baseline") baseline = watcher.baseline() stage = self._monotonic_ns() action_attempted = True ui.trigger_current_detail() source_state = PackageState.P02_TRIGGER_UNKNOWN remaining = self.budget.require(f"item_{index}", "cache_wait", 90_000) match = watcher.wait_for_unique_stable(baseline, timeout_ms=remaining) if match.status == "AMBIGUOUS": raise ContractError(ErrorCode.CACHE_AMBIGUOUS, "cache", ",".join(match.ambiguous_paths)) if match.status != "UNIQUE_STABLE" or match.remote is None: raise ContractError(ErrorCode.CACHE_TIMEOUT, "cache", "no stable new file") trigger_confirmed = True triggered += 1 source_state = PackageState.P03_CACHE_MATCHED self.budget.checkpoint(f"item_{index}", "quota_confirm") quota_terminal = dict(quota.terminal( reservation, event_type="CONSUME_CONFIRMED", task_id=self.spec.task_id, requester_role=self.spec.requester, handoff_id=self.spec.handoff_id, run_id=run_id, slot_id=slot_id, report_identity=report_identity, note="UNIQUE_STABLE_CACHE_FILE", )) self.timings.record(f"item_{index}_cache", stage, self._monotonic_ns(), "PASS") stage = self._monotonic_ns() remote_sha = adb.remote_sha256(match.remote.path) staging_path = staging_root / f"{item_id}.pdf" adb.pull(match.remote.path, staging_path, timeout_ms=self.budget.require(f"item_{index}", "adb_pull", 60_000)) source_state = PackageState.P04_STAGING_PARTIAL if staging_path.stat().st_size != match.remote.bytes: raise ContractError(ErrorCode.BYTE_COUNT_MISMATCH, "staging", "remote/local bytes") if sha256_file(staging_path, checkpoint=lambda: self.budget.checkpoint( f"item_{index}", "staging_sha256")) != remote_sha: raise ContractError(ErrorCode.HASH_MISMATCH, "staging", "remote/local sha256") self.timings.record(f"item_{index}_copy", stage, self._monotonic_ns(), "PASS") stage = self._monotonic_ns() pdfinfo = None if synthetic else Path(self.spec.pdfinfo_executable) evidence = validate_pdf( staging_path, pdfinfo_executable=pdfinfo, timeout_ms=self.budget.require(f"item_{index}", "pdfinfo_staging", 20_000), checkpoint=lambda: self.budget.checkpoint(f"item_{index}", "validate_staging"), ) source_state = PackageState.P05_STAGING_VALID if evidence.page_count is not None and detailed.page_count != evidence.page_count: raise ContractError(ErrorCode.PAGE_COUNT_MISMATCH, "page_count", "detail/pdfinfo mismatch") name = "-".join(( _safe_part(detailed.report_date or "UNKNOWN", 10), _safe_part(detailed.institution or "UNKNOWN", 40), _safe_part(detailed.title, 80), )) + ".pdf" destination = Path(self.spec.destination) if not destination.is_absolute(): destination = Path.cwd() / destination published = publish_no_replace( staging_path, destination / name, expected_sha256=remote_sha, pdfinfo_executable=pdfinfo, timeout_ms=self.budget.require(f"item_{index}", "publish_pdfinfo", 20_000), checkpoint=lambda: self.budget.checkpoint(f"item_{index}", "publish"), ) source_state = PackageState.P06_PUBLISHED self.budget.checkpoint(f"item_{index}", "quota_artifact") artifact_event = dict(quota.artifact( task_id=self.spec.task_id, requester_role=self.spec.requester, handoff_id=self.spec.handoff_id, run_id=run_id, slot_id=slot_id, report_identity=report_identity, quota_terminal_event_id=quota_terminal["event_id"], success=not published.duplicate, note="PDF_VALID_PUBLISHED" if not published.duplicate else "IDENTICAL_EXISTING_PDF", )) item_status = ItemStatus.DUPLICATE if published.duplicate else ItemStatus.SUCCESS item_stop = ErrorCode.DUPLICATE_EXISTING_ARTIFACT if published.duplicate else None row_id = hashlib.sha256( f"{run_id}|{item_id}|{report_identity}|{published.evidence.sha256}".encode("utf-8") ).hexdigest() external_hash = hashlib.sha256( f"{match.remote.path}|{match.remote.bytes}|{remote_sha}|{published.evidence.bytes}|{published.evidence.sha256}".encode("utf-8") ).hexdigest() item_terminal_recorded = True if not self._complete_item(index): raise ContractError(ErrorCode.DEADLINE_EXPIRED, f"item_{index}", "item_close") manifest_rows.append(_manifest_row( task_id=self.spec.task_id, requested_by=self.spec.requester, review_owner=self.spec.review_owner, source_url="NOT_APPLICABLE_APP_CACHE", source_site=SOURCE_SITE, title=detailed.title, publisher=detailed.institution, report_date=detailed.report_date, downloaded_at=utc_now(), http_status="NOT_APPLICABLE_APP_CACHE", content_type="application/pdf", file_name=published.final_path.name, relative_path=str(published.final_path), bytes=published.evidence.bytes, sha256=published.evidence.sha256, download_status=item_status.value, error_or_note="IDENTICAL_EXISTING_PDF" if published.duplicate else "NONE", source_cache_path=match.remote.path, source_file_name=match.remote.name, android_package=self.spec.package_name, extension_added="false", pdf_magic_valid="true", remote_sha256=remote_sha, local_sha256=published.evidence.sha256, openability="true", page_count=detailed.page_count, encryption_status="UNKNOWN" if published.evidence.encrypted is None else str(published.evidence.encrypted).lower(), schema_version="HIBOR_REPORT_MANIFEST_V004", row_id=row_id, handoff_id=self.spec.handoff_id, run_id=run_id, item_id=item_id, slot_id=slot_id, query=self.spec.query, candidate_id=detailed.candidate_id, report_identity=report_identity, analysts="|".join(detailed.analysts), selection_reason=f"FAST_SCORE={detailed.score}", remote_bytes=match.remote.bytes, local_bytes=published.evidence.bytes, quota_reservation_id=reservation.reservation_id, quota_terminal_event_id=quota_terminal["event_id"], quota_artifact_event_id=artifact_event["event_id"], status=item_status.value, stop_code=item_stop.value if item_stop else "", reused_without_new_trigger="false", external_evidence_hash=external_hash, manifested_at_utc=utc_now(), )) items.append(terminal_item({ "item_id": item_id, "slot_id": slot_id, "candidate_id": detailed.candidate_id, "report_identity": report_identity, "state": PackageState.P06_PUBLISHED.value, "status": item_status.value, "stop_code": item_stop.value if item_stop else None, "trigger_attempted": True, "triggered": True, "quota_state": QuotaState.CONFIRMED.value, "quota_reservation_id": reservation.reservation_id, "quota_terminal_event_id": quota_terminal["event_id"], "quota_artifact_event_id": artifact_event["event_id"], "title": detailed.title, "institution": detailed.institution, "report_date": detailed.report_date, "analysts": list(detailed.analysts), "page_count": detailed.page_count, "bytes": published.evidence.bytes, "sha256": published.evidence.sha256, "source_cache_path": match.remote.path, "source_file_name": match.remote.name, "final_path": str(published.final_path), "manifest_row_id": row_id, "reused_without_new_trigger": False, "error_or_note": "IDENTICAL_EXISTING_PDF" if published.duplicate else "NONE", })) package_state = PackageState.P06_PUBLISHED self.timings.record(f"item_{index}_validation_publish", stage, self._monotonic_ns(), item_status.value) except ContractError as exc: artifact_event_id = artifact_event["event_id"] if artifact_event else None try: if reservation and reservation.allowed and quota_terminal is None: self.budget.checkpoint(f"item_{index}", "quota_failure_terminal", close=True) event_type = "CONSUME_UNCERTAIN" if action_attempted else "RELEASE" quota_terminal = dict(quota.terminal( reservation, event_type=event_type, task_id=self.spec.task_id, requester_role=self.spec.requester, handoff_id=self.spec.handoff_id, run_id=run_id, slot_id=slot_id, report_identity=report_identity or f"UNKNOWN-{item_id}", note=exc.code.value, )) if (quota_terminal and quota_terminal["event_type"] != "RELEASE" and artifact_event is None): self.budget.checkpoint(f"item_{index}", "quota_failure_artifact", close=True) artifact_event = dict(quota.artifact( task_id=self.spec.task_id, requester_role=self.spec.requester, handoff_id=self.spec.handoff_id, run_id=run_id, slot_id=slot_id, report_identity=report_identity or f"UNKNOWN-{item_id}", quota_terminal_event_id=quota_terminal["event_id"], success=bool(published is not None and not published.duplicate), note=("PDF_VALID_PUBLISHED_DURING_FAILURE_CLOSE" if published is not None and not published.duplicate else exc.code.value), )) artifact_event_id = artifact_event["event_id"] except ContractError as closure_exc: exc = closure_exc if not item_terminal_recorded: item_terminal_recorded = True if not self._complete_item(index): exc = ContractError(ErrorCode.DEADLINE_EXPIRED, f"item_{index}", "item_close") stop_code = exc.code blocker = f"{exc.field_name or '-'}:{exc.detail}" quota_state = self._quota_state_from_terminal(quota_terminal) failure_row = self._failure_manifest_row( run_id=run_id, item_id=item_id, slot_id=slot_id, candidate=detailed, report_identity=report_identity, code=exc.code, note=blocker, reservation=reservation, quota_terminal=quota_terminal, artifact_event_id=artifact_event_id, action_attempted=action_attempted, match=match, remote_sha=remote_sha, published=published, ) manifest_rows.append(failure_row) items.append(terminal_item({ "item_id": item_id, "slot_id": slot_id, "candidate_id": detailed.candidate_id if detailed else None, "report_identity": report_identity, "state": source_state.value, "status": ItemStatus.STOPPED.value, "stop_code": exc.code.value, "trigger_attempted": action_attempted, "triggered": trigger_confirmed, "quota_state": quota_state.value, "quota_reservation_id": reservation.reservation_id if reservation else None, "quota_terminal_event_id": quota_terminal["event_id"] if quota_terminal else None, "quota_artifact_event_id": artifact_event_id, "title": detailed.title if detailed else None, "institution": detailed.institution if detailed else None, "report_date": detailed.report_date if detailed else None, "analysts": list(detailed.analysts) if detailed and detailed.analysts else None, "page_count": (published.evidence.page_count if published and published.evidence.page_count is not None else detailed.page_count if detailed else None), "bytes": published.evidence.bytes if published else None, "sha256": published.evidence.sha256 if published else None, "source_cache_path": match.remote.path if match and match.remote else None, "source_file_name": match.remote.name if match and match.remote else None, "final_path": str(published.final_path) if published else None, "manifest_row_id": failure_row["row_id"], "reused_without_new_trigger": False, "error_or_note": blocker, })) terminal_status = ( TerminalStatus.STATE_UNCERTAIN if quota_state is QuotaState.UNCERTAIN else self._status_for_error(exc.code, any( row["status"] in {ItemStatus.SUCCESS.value, ItemStatus.DUPLICATE.value} for row in items )) ) break if terminal_status is TerminalStatus.SUCCESS and len(items) < self.spec.quantity: stop_code = ErrorCode.CANDIDATE_REJECTED blocker = "INSUFFICIENT_QUALIFIED_CANDIDATES" terminal_status = TerminalStatus.PARTIAL_SUCCESS if items else TerminalStatus.BLOCKED_INPUT try: adb.screenshot(screenshot_root / "99-end.png") except ContractError: if terminal_status is TerminalStatus.SUCCESS: terminal_status = TerminalStatus.PARTIAL_SUCCESS stop_code = ErrorCode.UI_CURSOR_RESTORE_FAILED blocker = "END_SCREENSHOT_FAILED" self.budget.checkpoint("close", "quota_snapshot", batch_close=True) quota_snapshot = quota.snapshot() stage = self._monotonic_ns() if manifest_rows: write_manifest_create_new( manifest_path, manifest_rows, checkpoint=lambda: self.budget.checkpoint("close", "manifest", batch_close=True), ) manifest_verified = True package_state = PackageState.P07_MANIFESTED else: manifest_verified = False self.timings.record("manifest", stage, self._monotonic_ns(), "PASS" if manifest_verified else "ABSENT") stage = self._monotonic_ns() delivery_text = self._delivery_text(run_id, items, quota_snapshot, terminal_status, stop_code) write_bytes_create_new( delivery_path, delivery_text.encode("utf-8"), checkpoint=lambda: self.budget.checkpoint("close", "delivery", batch_close=True), ) delivery_verified = delivery_path.read_bytes() == delivery_text.encode("utf-8") if not delivery_verified: raise ContractError(ErrorCode.PERSIST_LATE, "delivery", "readback mismatch") package_state = PackageState.P08_DELIVERY self.timings.record("delivery", stage, self._monotonic_ns(), "PASS") stage = self._monotonic_ns() timing_value = self._timing_value(run_id) write_json_create_new( timing_path, timing_value, checkpoint=lambda: self.budget.checkpoint("close", "timing", batch_close=True), ) timing_verified = True package_state = PackageState.P09_TIMING self.timings.record("timing", stage, self._monotonic_ns(), "PASS") except ContractError as exc: stop_code = exc.code blocker = f"{exc.field_name or '-'}:{exc.detail}" terminal_status = self._status_for_error(exc.code, bool(items)) except Exception as exc: # fail closed at the public boundary stop_code = ErrorCode.UNEXPECTED_EXCEPTION blocker = f"{type(exc).__name__}:{str(exc)[:240]}" terminal_status = TerminalStatus.INTERNAL_ERROR if quota is not None: try: quota_snapshot = quota.snapshot() except ContractError: terminal_status = TerminalStatus.STATE_UNCERTAIN stop_code = ErrorCode.QUOTA_LEDGER_INVALID blocker = "QUOTA_FINAL_SNAPSHOT_UNCERTAIN" base = self._base_terminal( quota_snapshot, run_id=run_id, items=items, triggered=triggered, manifest_path=manifest_path if manifest_verified else None, delivery_path=delivery_path if delivery_verified else None, timing_path=timing_path if timing_verified else None, ) intended_items = [dict(row, state=PackageState.P10_CLOSED.value) for row in items] intended_base = dict(base) intended_base["items"] = intended_items intended = TerminalReceipt( terminal_path, terminal_path, True, True, EvidenceState.V, True, False, True, 2, "0" * 64, None, PackageState.P10_CLOSED, ) record = build_terminal(intended_base, intended, status=terminal_status, stop_code=stop_code, blocker=blocker) receipt = TerminalWriter().persist( terminal_path, record, source_state=package_state, checkpoint=lambda: self.budget.checkpoint("close", "terminal", batch_close=True), ) if receipt.presence is EvidenceState.V: return build_terminal(intended_base, receipt, status=terminal_status, stop_code=stop_code, blocker=blocker) if (receipt.presence is EvidenceState.N and not receipt.persist_attempted and self.budget.batch_close_deadline_reached): return build_terminal(base, receipt, status=TerminalStatus.TIME_BUDGET_STOP, stop_code=ErrorCode.DEADLINE_EXPIRED, blocker="BATCH_CLOSE_DEADLINE_EXHAUSTED") terminal_status = (TerminalStatus.STATE_UNCERTAIN if receipt.presence is EvidenceState.U else TerminalStatus.INTERNAL_ERROR) return build_terminal(base, receipt, status=terminal_status, stop_code=receipt.error_code or ErrorCode.PERSIST_LATE, blocker="TERMINAL_PERSIST_NOT_VALID") def _quota_snapshot_readonly(self) -> QuotaSnapshot: if self.spec.quota_path.is_file(): local_date = datetime.now(ZoneInfo("Asia/Shanghai")).date().isoformat() return QuotaLedger( self.spec.quota_path, quota_date=local_date, timeout_provider=lambda operation, maximum: self.budget.require( "quota", operation, maximum, close=self.budget.work_deadline_reached), ).snapshot() local_date = datetime.now(ZoneInfo("Asia/Shanghai")).date().isoformat() known = 3 if local_date == "2026-07-29" else 0 return QuotaSnapshot(local_date, 0, 0, 0, known, 0, known, 25 - known, 0) def _begin_item(self, index: int) -> None: self.budget.begin_item(index) if index in self._open_item_windows: return close_deadline = self.budget.active_item_deadline work_deadline = self.budget.active_item_work_deadline if close_deadline is None or work_deadline is None: raise ContractError(ErrorCode.CLOCK_INVALID, f"item_{index}", "deadline unavailable") self._open_item_windows[index] = { "item_index": index, "started_monotonic_ns": self._monotonic_ns(), "work_deadline_monotonic_ns": int(work_deadline * 1_000_000_000), "close_deadline_monotonic_ns": int(close_deadline * 1_000_000_000), } def _complete_item(self, index: int) -> bool: window = self._open_item_windows.pop(index, None) if window is None: raise ContractError(ErrorCode.CLOCK_INVALID, f"item_{index}", "start evidence absent") terminal_ns = self._monotonic_ns() within_deadline = self.budget.complete_item(index) window.update({ "terminal_monotonic_ns": terminal_ns, "elapsed_ms": max(0, (terminal_ns - window["started_monotonic_ns"]) // 1_000_000), "terminal_delta_from_previous_ms": ( None if not self.item_windows else max(0, (terminal_ns - self.item_windows[-1]["terminal_monotonic_ns"]) // 1_000_000) ), "deadline_met": within_deadline, }) self.item_windows.append(window) return within_deadline def _timing_value(self, run_id: str) -> dict[str, Any]: return { "schema_version": "HIBOR_FAST_TIMING_V002", "run_id": run_id, "observed_at_utc": self.spec.observed_at_utc, "budget_ms": self.spec.total_budget_ms, "close_reserve_ms": self.spec.close_reserve_ms, "batch_increment_budget_ms": self.spec.batch_increment_budget_ms, "item_windows": self.item_windows, "rows": self.timings.rows, } def _preflight_terminal(self, run_id: str, terminal_path: Path, quota: QuotaSnapshot, exc: ContractError) -> dict[str, Any]: status = self._status_for_error(exc.code, False) base = self._base_terminal(quota, run_id=run_id, items=[], triggered=0) intended = TerminalReceipt( terminal_path, terminal_path, True, True, EvidenceState.V, True, False, True, 2, "0" * 64, None, PackageState.P10_CLOSED, ) record = build_terminal(base, intended, status=status, stop_code=exc.code, blocker=f"{exc.field_name or '-'}:{exc.detail}") receipt = TerminalWriter().persist( terminal_path, record, source_state=PackageState.P00_INIT, checkpoint=lambda: self.budget.checkpoint("close", "preflight_terminal", batch_close=True), ) if receipt.presence is EvidenceState.V: return build_terminal(base, receipt, status=status, stop_code=exc.code, blocker=f"{exc.field_name or '-'}:{exc.detail}") if (exc.code is ErrorCode.DEADLINE_EXPIRED and receipt.presence is EvidenceState.N and not receipt.persist_attempted): return build_terminal(base, receipt, status=TerminalStatus.TIME_BUDGET_STOP, stop_code=ErrorCode.DEADLINE_EXPIRED, blocker=f"{exc.field_name or '-'}:{exc.detail}") return build_terminal( base, receipt, status=TerminalStatus.STATE_UNCERTAIN if receipt.presence is EvidenceState.U else TerminalStatus.INTERNAL_ERROR, stop_code=receipt.error_code or ErrorCode.PERSIST_LATE, blocker="PREFLIGHT_TERMINAL_PERSIST_NOT_VALID", ) def _discover_project_reuse(self, *, synthetic: bool) -> list[dict[str, Any]]: """Verify immutable project PDF+manifest relationships without opening the APP.""" configured = self.spec.source_scope.get("reuse_manifest_paths", []) if configured is None: configured = [] if not isinstance(configured, list) or any(not isinstance(p, str) for p in configured): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "reuse_manifest_paths", "string array") paths = {Path(p) for p in configured} if self.spec.output_path.exists(): paths.update(self.spec.output_path.rglob("manifest.csv")) results: list[dict[str, Any]] = [] seen: set[str] = set() pdfinfo = None if synthetic else Path(self.spec.pdfinfo_executable) for manifest in sorted(paths, key=lambda p: str(p)): self.budget.checkpoint("reuse", "manifest_read") if not manifest.is_file() or manifest.is_symlink(): continue try: with manifest.open("r", encoding="utf-8", newline="") as stream: reader = csv.DictReader(stream) if tuple(reader.fieldnames or ()) != MANIFEST_COLUMNS: continue source_rows = list(reader) except (OSError, UnicodeError, csv.Error): continue for source in source_rows: if source.get("status") not in {"SUCCESS", "DUPLICATE"}: continue identity = source.get("report_identity", "") if not re.fullmatch(r"[0-9a-f]{64}", identity) or identity in seen: continue try: candidate = Candidate( source.get("candidate_id") or identity[:24], source["title"], source.get("publisher") or None, source.get("report_date") or None, tuple(x for x in source.get("analysts", "").split("|") if x), int(source["page_count"]) if source.get("page_count") else None, ) if _identity(candidate) != identity: continue self._enforce_hard_filters(candidate) final_path = Path(source["relative_path"]) if not final_path.is_absolute(): final_path = Path.cwd() / final_path evidence = validate_pdf( final_path, pdfinfo_executable=pdfinfo, timeout_ms=self.budget.require("reuse", "pdfinfo", 20_000), checkpoint=lambda: self.budget.checkpoint("reuse", "pdf_verify"), ) if (str(evidence.bytes) != source.get("bytes") or evidence.sha256 != source.get("sha256") or (evidence.page_count is not None and str(evidence.page_count) != source.get("page_count"))): continue except ContractError as exc: if exc.code is ErrorCode.DEADLINE_EXPIRED: raise continue except (OSError, ValueError, KeyError): continue seen.add(identity) results.append({"source": source, "candidate": candidate, "evidence": evidence, "final_path": final_path}) ranked = sorted(results, key=lambda row: ( -score_candidate(row["candidate"], query=self.spec.query, aliases=tuple(self.spec.aliases), institutions=tuple(self.spec.institutions), minimum_pages=self.spec.minimum_pages), row["candidate"].candidate_id, )) expected = [row["report_identity"] for row in self.spec.expected_reports] if expected: by_identity = {row["source"]["report_identity"]: row for row in ranked} return [by_identity[identity] for identity in expected if identity in by_identity] return ranked def _close_project_reuse(self, *, run_id: str, run_root: Path, manifest_path: Path, delivery_path: Path, timing_path: Path, terminal_path: Path, rows: list[dict[str, Any]]) -> dict[str, Any]: quota_before = self.spec.quota_path.read_bytes() if self.spec.quota_path.is_file() else None quota = self._quota_snapshot_readonly() manifest_rows, items = self._build_reuse_payload(run_id, rows) write_manifest_create_new( manifest_path, manifest_rows, checkpoint=lambda: self.budget.checkpoint("close", "reuse_manifest", batch_close=True), ) delivery = self._delivery_text(run_id, items, quota, TerminalStatus.SUCCESS, None) write_bytes_create_new( delivery_path, delivery.encode("utf-8"), checkpoint=lambda: self.budget.checkpoint("close", "reuse_delivery", batch_close=True), ) timing_value = self._timing_value(run_id) write_json_create_new( timing_path, timing_value, checkpoint=lambda: self.budget.checkpoint("close", "reuse_timing", batch_close=True), ) quota_after = self.spec.quota_path.read_bytes() if self.spec.quota_path.is_file() else None if quota_before != quota_after: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "pure_reuse", "ledger mutated") closed = [dict(item, state=PackageState.P10_CLOSED.value) for item in items] base = self._base_terminal(quota, run_id=run_id, items=closed, triggered=0, manifest_path=manifest_path, delivery_path=delivery_path, timing_path=timing_path) intended = TerminalReceipt(terminal_path, terminal_path, True, True, EvidenceState.V, True, False, True, 2, "0" * 64, None, PackageState.P10_CLOSED) record = build_terminal(base, intended, status=TerminalStatus.SUCCESS, stop_code=None, blocker=None) receipt = TerminalWriter().persist( terminal_path, record, source_state=PackageState.P09_TIMING, checkpoint=lambda: self.budget.checkpoint("close", "reuse_terminal", batch_close=True), ) if receipt.presence is not EvidenceState.V: return build_terminal(dict(base, items=items), receipt, status=TerminalStatus.STATE_UNCERTAIN if receipt.presence is EvidenceState.U else TerminalStatus.INTERNAL_ERROR, stop_code=receipt.error_code or ErrorCode.PERSIST_LATE, blocker="REUSE_TERMINAL_PERSIST_NOT_VALID") return build_terminal(base, receipt, status=TerminalStatus.SUCCESS, stop_code=None, blocker=None) def _build_reuse_payload(self, run_id: str, rows: list[dict[str, Any]], *, start_index: int = 1) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: manifest_rows: list[dict[str, Any]] = [] items: list[dict[str, Any]] = [] for index, entry in enumerate(rows, start_index): source = entry["source"] candidate: Candidate = entry["candidate"] evidence = entry["evidence"] final_path: Path = entry["final_path"] item_id, slot_id = f"ITEM-{index:03d}", f"SLOT-{index:03d}" row_id = hashlib.sha256( f"{run_id}|{item_id}|{source['report_identity']}|{evidence.sha256}|PURE".encode("utf-8") ).hexdigest() external_hash = hashlib.sha256( f"{source.get('row_id','')}|{evidence.bytes}|{evidence.sha256}|PURE".encode("utf-8") ).hexdigest() row = _manifest_row(**{key: source.get(key) for key in MANIFEST_COLUMNS}) row.update({ "task_id": self.spec.task_id, "requested_by": self.spec.requester, "review_owner": self.spec.review_owner, "downloaded_at": self.started_at, "relative_path": str(final_path), "download_status": ItemStatus.SUCCESS.value, "error_or_note": "ZERO_QUOTA_REUSE", "schema_version": "HIBOR_REPORT_MANIFEST_V004", "row_id": row_id, "handoff_id": self.spec.handoff_id, "run_id": run_id, "item_id": item_id, "slot_id": slot_id, "query": self.spec.query, "quota_reservation_id": None, "quota_terminal_event_id": None, "quota_artifact_event_id": None, "status": ItemStatus.SUCCESS.value, "stop_code": None, "reused_without_new_trigger": "true", "external_evidence_hash": external_hash, "manifested_at_utc": self.started_at, }) manifest_rows.append(_manifest_row(**row)) items.append(terminal_item({ "item_id": item_id, "slot_id": slot_id, "candidate_id": candidate.candidate_id, "report_identity": source["report_identity"], "state": PackageState.P06_PUBLISHED.value, "status": ItemStatus.SUCCESS.value, "stop_code": None, "trigger_attempted": False, "triggered": False, "quota_state": QuotaState.NONE.value, "quota_reservation_id": None, "quota_terminal_event_id": None, "quota_artifact_event_id": None, "title": candidate.title, "institution": candidate.institution, "report_date": candidate.report_date, "analysts": list(candidate.analysts), "page_count": evidence.page_count or candidate.page_count, "bytes": evidence.bytes, "sha256": evidence.sha256, "source_cache_path": source.get("source_cache_path") or None, "source_file_name": source.get("source_file_name") or None, "final_path": str(final_path), "manifest_row_id": row_id, "reused_without_new_trigger": True, "error_or_note": "ZERO_QUOTA_REUSE", })) return manifest_rows, items def _materialize_cache_reuse(self, *, adb: Any, run_root: Path, synthetic: bool) -> list[dict[str, Any]]: raw_items = self.spec.source_scope.get("resume_items", []) if not isinstance(raw_items, list): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "resume_items", "array required") self._begin_item(1) self.budget.checkpoint("cache_reuse", "adb_preflight") adb.preflight() files = {item.path: item for item in adb.list_cache()} output: list[dict[str, Any]] = [] pdfinfo = None if synthetic else Path(self.spec.pdfinfo_executable) for index, raw in enumerate(raw_items, 1): if not isinstance(raw, Mapping): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "resume_items", "mapping required") required = {"candidate_id", "report_identity", "title", "institution", "report_date", "analysts", "page_count", "remote_path", "remote_bytes", "remote_sha256"} if set(raw) != required: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "resume_items", "exact keys required") self._begin_item(index) candidate = Candidate( str(raw["candidate_id"]), str(raw["title"]), str(raw["institution"]), str(raw["report_date"]), tuple(str(x) for x in raw["analysts"]), int(raw["page_count"]), ) if _identity(candidate) != raw["report_identity"]: raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "report_identity", "cache identity") self._enforce_hard_filters(candidate) remote = files.get(str(raw["remote_path"])) if remote is None or remote.bytes != int(raw["remote_bytes"]): raise ContractError(ErrorCode.CACHE_AMBIGUOUS, "remote_path", "recorded cache object absent/drift") remote_sha = adb.remote_sha256(remote.path) if remote_sha != raw["remote_sha256"]: raise ContractError(ErrorCode.HASH_MISMATCH, "remote_sha256", "recorded cache hash drift") staging = run_root / "staging" / f"ITEM-{index:03d}.pdf" adb.pull(remote.path, staging, timeout_ms=self.budget.require(f"item_{index}", "cache_reuse_pull", 60_000)) name = "-".join((_safe_part(candidate.report_date or "UNKNOWN", 10), _safe_part(candidate.institution or "UNKNOWN", 40), _safe_part(candidate.title, 80))) + ".pdf" destination = Path(self.spec.destination) if not destination.is_absolute(): destination = Path.cwd() / destination published = publish_no_replace( staging, destination / name, expected_sha256=remote_sha, pdfinfo_executable=pdfinfo, timeout_ms=self.budget.require(f"item_{index}", "cache_reuse_publish", 20_000), checkpoint=lambda: self.budget.checkpoint(f"item_{index}", "cache_reuse_publish"), ) if published.evidence.page_count is not None and published.evidence.page_count != candidate.page_count: raise ContractError(ErrorCode.PAGE_COUNT_MISMATCH, "page_count", "cache reuse") source = {key: "" for key in MANIFEST_COLUMNS} source.update({ "source_url": "NOT_APPLICABLE_APP_CACHE", "source_site": SOURCE_SITE, "title": candidate.title, "publisher": candidate.institution, "report_date": candidate.report_date, "http_status": "NOT_APPLICABLE_APP_CACHE", "content_type": "application/pdf", "file_name": published.final_path.name, "relative_path": str(published.final_path), "bytes": str(published.evidence.bytes), "sha256": published.evidence.sha256, "source_cache_path": remote.path, "source_file_name": remote.name, "android_package": self.spec.package_name, "extension_added": "false", "pdf_magic_valid": "true", "remote_sha256": remote_sha, "local_sha256": published.evidence.sha256, "openability": "true", "page_count": str(candidate.page_count), "encryption_status": "UNKNOWN" if published.evidence.encrypted is None else str(published.evidence.encrypted).lower(), "candidate_id": candidate.candidate_id, "report_identity": raw["report_identity"], "analysts": "|".join(candidate.analysts), "selection_reason": "CACHE_REUSE_VERIFIED", "remote_bytes": str(remote.bytes), "local_bytes": str(published.evidence.bytes), "row_id": hashlib.sha256(f"CACHE|{raw['report_identity']}|{remote_sha}".encode()).hexdigest(), }) output.append({"source": source, "candidate": candidate, "evidence": published.evidence, "final_path": published.final_path}) if not self._complete_item(index): raise ContractError(ErrorCode.DEADLINE_EXPIRED, f"item_{index}", "item_close") return output def _close_reuse_item_windows(self, count: int) -> None: for index in range(1, count + 1): self._begin_item(index) if not self._complete_item(index): raise ContractError(ErrorCode.DEADLINE_EXPIRED, f"item_{index}", "item_close") def _run_id(self) -> str: canonical = json.dumps(self.spec.raw, ensure_ascii=False, separators=(",", ":"), sort_keys=False).encode("utf-8") token = hashlib.sha256(b"HIBOR-RUN-V001\x00" + canonical).hexdigest()[:24] task = re.sub(r"[^A-Za-z0-9_.-]", "-", self.spec.task_id)[:80] return f"RUN-{task}-{token}" @staticmethod def _status_for_error(code: ErrorCode, has_success: bool) -> TerminalStatus: if code in {ErrorCode.PROCESS_LIVENESS_UNKNOWN, ErrorCode.RECOVERY_UNKNOWN, ErrorCode.TRIGGER_UNCERTAIN}: return TerminalStatus.STATE_UNCERTAIN if code is ErrorCode.ACCESS_CONTROL_PRESENT: return TerminalStatus.BLOCKED_ACCESS_CONTROL if code is ErrorCode.DEADLINE_EXPIRED: return TerminalStatus.TIME_BUDGET_STOP if code is ErrorCode.QUOTA_EXHAUSTED: return TerminalStatus.PARTIAL_QUOTA_STOP if has_success: return TerminalStatus.PARTIAL_SUCCESS if code in {ErrorCode.PDF_MAGIC_INVALID, ErrorCode.BYTE_COUNT_MISMATCH, ErrorCode.HASH_MISMATCH, ErrorCode.PDF_NOT_OPENABLE, ErrorCode.PAGE_COUNT_MISMATCH, ErrorCode.PDF_ENCRYPTED}: return TerminalStatus.VALIDATION_FAILED if code in {ErrorCode.CANDIDATE_REJECTED, ErrorCode.DETAIL_RESULT_MISMATCH}: return TerminalStatus.BLOCKED_INPUT return TerminalStatus.BLOCKED_ENVIRONMENT @staticmethod def _quota_state_from_terminal(quota_terminal: dict[str, str] | None) -> QuotaState: """Project the durable quota terminal without rewriting confirmed facts.""" if quota_terminal is None: return QuotaState.NONE event_type = quota_terminal.get("event_type") if event_type == "CONSUME_CONFIRMED": return QuotaState.CONFIRMED if event_type == "CONSUME_UNCERTAIN": return QuotaState.UNCERTAIN if event_type == "RELEASE": return QuotaState.RELEASED raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota_terminal", f"unexpected event_type {event_type!r}") def _failure_manifest_row(self, *, run_id: str, item_id: str, slot_id: str, candidate: Candidate | None, report_identity: str | None, code: ErrorCode, note: str, reservation: Reservation | None, quota_terminal: dict[str, str] | None, artifact_event_id: str | None, action_attempted: bool, match: Any | None = None, remote_sha: str | None = None, published: Any | None = None) -> dict[str, Any]: identity = report_identity or f"UNKNOWN-{item_id}" row_id = hashlib.sha256( f"{run_id}|{item_id}|{identity}|{code.value}".encode("utf-8") ).hexdigest() remote = match.remote if match is not None else None if published is not None: external_hash = hashlib.sha256( (f"{remote.path if remote else ''}|{remote.bytes if remote else ''}|" f"{remote_sha or ''}|{published.evidence.bytes}|" f"{published.evidence.sha256}").encode("utf-8") ).hexdigest() download_status = (ItemStatus.DUPLICATE.value if published.duplicate else ItemStatus.SUCCESS.value) else: external_hash = hashlib.sha256( f"{identity}|{code.value}|{note}".encode("utf-8") ).hexdigest() download_status = ItemStatus.STOPPED.value return _manifest_row( task_id=self.spec.task_id, requested_by=self.spec.requester, review_owner=self.spec.review_owner, source_url="NOT_APPLICABLE_APP_CACHE" if published else "UNKNOWN", source_site=SOURCE_SITE, title=candidate.title if candidate else "UNKNOWN", publisher=candidate.institution if candidate and candidate.institution else "UNKNOWN", report_date=candidate.report_date if candidate and candidate.report_date else "UNKNOWN", downloaded_at=utc_now(), http_status="NOT_APPLICABLE_APP_CACHE" if published else "UNKNOWN", content_type="application/pdf" if published else "UNKNOWN", file_name=published.final_path.name if published else None, relative_path=str(published.final_path) if published else None, bytes=published.evidence.bytes if published else None, sha256=published.evidence.sha256 if published else None, download_status=download_status, error_or_note=note, source_cache_path=remote.path if remote else None, source_file_name=remote.name if remote else None, android_package=self.spec.package_name, extension_added="false", pdf_magic_valid="true" if published else "UNKNOWN", remote_sha256=remote_sha, local_sha256=published.evidence.sha256 if published else None, openability="true" if published else "UNKNOWN", page_count=(published.evidence.page_count if published and published.evidence.page_count is not None else candidate.page_count if candidate else None), encryption_status=("UNKNOWN" if not published or published.evidence.encrypted is None else str(published.evidence.encrypted).lower()), schema_version="HIBOR_REPORT_MANIFEST_V004", row_id=row_id, handoff_id=self.spec.handoff_id, run_id=run_id, item_id=item_id, slot_id=slot_id, query=self.spec.query, candidate_id=candidate.candidate_id if candidate else None, report_identity=identity, analysts="|".join(candidate.analysts) if candidate and candidate.analysts else None, selection_reason="FAST_SELECTED" if candidate else "NOT_SELECTED", remote_bytes=remote.bytes if remote else None, local_bytes=published.evidence.bytes if published else None, quota_reservation_id=reservation.reservation_id if reservation else None, quota_terminal_event_id=quota_terminal["event_id"] if quota_terminal else None, quota_artifact_event_id=artifact_event_id, status=download_status, stop_code=code.value, reused_without_new_trigger="false", external_evidence_hash=external_hash, manifested_at_utc=utc_now(), ) def _enforce_hard_filters(self, candidate: Candidate) -> None: title = candidate.title.casefold() required_terms = [self.spec.query.casefold(), *(str(x).casefold() for x in self.spec.aliases)] if not any(term and term in title for term in required_terms): raise ContractError(ErrorCode.CANDIDATE_REJECTED, "title", "query/alias absent") if any(str(term).casefold() in title for term in self.spec.exclude if str(term)): raise ContractError(ErrorCode.CANDIDATE_REJECTED, "exclude", "excluded title term") if self.spec.institutions and candidate.institution not in set(self.spec.institutions): raise ContractError(ErrorCode.CANDIDATE_REJECTED, "institution", "not allowed") if self.spec.analysts and not set(candidate.analysts).intersection(self.spec.analysts): raise ContractError(ErrorCode.CANDIDATE_REJECTED, "analysts", "no required analyst") if candidate.page_count is None or candidate.page_count < self.spec.minimum_pages: raise ContractError(ErrorCode.CANDIDATE_REJECTED, "page_count", "below minimum") if self.spec.date_range is not None: value = self.spec.date_range if isinstance(value, dict): start, end = value.get("start"), value.get("end") elif isinstance(value, (list, tuple)) and len(value) == 2: start, end = value else: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "date_range", "start/end required") if not isinstance(start, str) or not isinstance(end, str) or start > end: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "date_range", "invalid bounds") if candidate.report_date is None or not (start <= candidate.report_date <= end): raise ContractError(ErrorCode.CANDIDATE_REJECTED, "report_date", "outside date range") def _base_terminal(self, quota: QuotaSnapshot, *, run_id: str, items: list[dict[str, Any]] | None = None, triggered: int = 0, manifest_path: Path | None = None, delivery_path: Path | None = None, timing_path: Path | None = None) -> dict[str, Any]: items = items or [] succeeded = sum(row["status"] == ItemStatus.SUCCESS.value for row in items) duplicates = sum(row["status"] == ItemStatus.DUPLICATE.value for row in items) failed = sum(row["status"] in {ItemStatus.FAILED.value, ItemStatus.STOPPED.value} for row in items) now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") return { "project_id": self.spec.project_id, "task_id": self.spec.task_id, "handoff_id": self.spec.handoff_id, "source_role_instance_id": self.spec.source_role_instance_id, "source_thread_id": self.spec.source_thread_id, "target_role_instance_id": self.spec.target_role_instance_id, "target_thread_id": self.spec.target_thread_id, "reply_thread_id": self.spec.reply_thread_id, "requester": self.spec.requester, "review_owner": self.spec.review_owner, "run_id": run_id, "mode": self.spec.mode, "observed_at_utc": self.spec.observed_at_utc, "started_at_utc": self.started_at, "ended_at_utc": now, "total_elapsed_ms": self.budget.elapsed_ms, "work_deadline_reached": self.budget.work_deadline_reached, "close_deadline_reached": self.budget.close_deadline_reached, "requested": self.spec.quantity, "triggered": triggered, "succeeded": succeeded, "failed": failed, "duplicates": duplicates, "gaps": max(0, self.spec.quantity - succeeded - duplicates), "quota_confirmed": quota.confirmed, "quota_uncertain": quota.uncertain, "quota_active": quota.active, "quota_cumulative_consumed": quota.effective_consumed, "quota_safe_available": quota.safe_available, "quota_ledger_path": str(self.spec.quota_path), "items": sorted(items, key=lambda row: row["item_id"]), "manifest_path": str(manifest_path) if manifest_path else None, "delivery_path": str(delivery_path) if delivery_path else None, "timing_path": str(timing_path) if timing_path else None, "prohibited_action_attestation": { "remote_original_modified": False, "credential_persisted": False, "access_control_bypassed": False, "body_parsed": False, "research_conclusion_generated": False, "external_message_sent": False, }, } def _delivery_text(self, run_id: str, items: list[dict[str, Any]], quota: QuotaSnapshot, status: TerminalStatus, stop_code: ErrorCode | None) -> str: rows = [ "# 慧博研报采集交付回执", "", f"- run_id: `{run_id}`", f"- task_id: `{self.spec.task_id}`", f"- status: `{status.value}`", f"- stop_code: `{stop_code.value if stop_code else 'NONE'}`", f"- requested/triggered: `{self.spec.quantity}/{sum(x['triggered'] for x in items)}`", f"- quota confirmed/uncertain/active: `{quota.confirmed}/{quota.uncertain}/{quota.active}`", "", "## 原件元数据", "", ] for item in items: rows.append( f"- {item['item_id']}: {item['status']} | {item['title'] or 'UNKNOWN'} | " f"{item['bytes'] if item['bytes'] is not None else 'UNKNOWN'} bytes | " f"{item['sha256'] or 'UNKNOWN'}" ) rows.extend(("", "正文未解析；PDF 与 SHA-256 为权威原件。", "")) return "\n".join(rows)