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