from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
from enum import Enum
|
from pathlib import Path
|
from typing import Any, Mapping
|
import re
|
|
|
class ContractError(ValueError):
|
def __init__(self, code: "ErrorCode", field_name: str | None, detail: str):
|
super().__init__(f"{code.value}:{field_name or '-'}:{detail}")
|
self.code = code
|
self.field_name = field_name
|
self.detail = detail
|
|
|
class ErrorCode(str, Enum):
|
NONE = "NONE"
|
TASK_SPEC_INVALID = "TASK_SPEC_INVALID"
|
CLOCK_INVALID = "CLOCK_INVALID"
|
DEADLINE_EXPIRED = "DEADLINE_EXPIRED"
|
PROCESS_START_FAILED = "PROCESS_START_FAILED"
|
PROCESS_TIMEOUT = "PROCESS_TIMEOUT"
|
PROCESS_OUTPUT_LIMIT = "PROCESS_OUTPUT_LIMIT"
|
PROCESS_LIVENESS_UNKNOWN = "PROCESS_LIVENESS_UNKNOWN"
|
DEVICE_NOT_UNIQUE = "DEVICE_NOT_UNIQUE"
|
DEVICE_NOT_ONLINE = "DEVICE_NOT_ONLINE"
|
PACKAGE_MISSING = "PACKAGE_MISSING"
|
CACHE_UNREADABLE = "CACHE_UNREADABLE"
|
UI_ANCHOR_DRIFT = "UI_ANCHOR_DRIFT"
|
ACCESS_CONTROL_PRESENT = "ACCESS_CONTROL_PRESENT"
|
UI_CURSOR_RESTORE_FAILED = "UI_CURSOR_RESTORE_FAILED"
|
DETAIL_RESULT_MISMATCH = "DETAIL_RESULT_MISMATCH"
|
CANDIDATE_REJECTED = "CANDIDATE_REJECTED"
|
TRIGGER_NOT_OCCURRED = "TRIGGER_NOT_OCCURRED"
|
TRIGGER_UNCERTAIN = "TRIGGER_UNCERTAIN"
|
CACHE_TIMEOUT = "CACHE_TIMEOUT"
|
CACHE_AMBIGUOUS = "CACHE_AMBIGUOUS"
|
REMOTE_FILE_CHANGED = "REMOTE_FILE_CHANGED"
|
LOCK_TIMEOUT = "LOCK_TIMEOUT"
|
QUOTA_LEDGER_INVALID = "QUOTA_LEDGER_INVALID"
|
QUOTA_REPLAY_CONFLICT = "QUOTA_REPLAY_CONFLICT"
|
QUOTA_EXHAUSTED = "QUOTA_EXHAUSTED"
|
QUOTA_DATE_UNCERTAIN = "QUOTA_DATE_UNCERTAIN"
|
PULL_FAILED = "PULL_FAILED"
|
PDF_MAGIC_INVALID = "PDF_MAGIC_INVALID"
|
BYTE_COUNT_MISMATCH = "BYTE_COUNT_MISMATCH"
|
HASH_MISMATCH = "HASH_MISMATCH"
|
PDF_NOT_OPENABLE = "PDF_NOT_OPENABLE"
|
PAGE_COUNT_MISMATCH = "PAGE_COUNT_MISMATCH"
|
PDF_ENCRYPTED = "PDF_ENCRYPTED"
|
FINAL_PATH_CONFLICT = "FINAL_PATH_CONFLICT"
|
PUBLISH_FAILED = "PUBLISH_FAILED"
|
RECOVERY_UNKNOWN = "RECOVERY_UNKNOWN"
|
MANIFEST_INVALID = "MANIFEST_INVALID"
|
PERSIST_LATE = "PERSIST_LATE"
|
DUPLICATE_EXISTING_ARTIFACT = "DUPLICATE_EXISTING_ARTIFACT"
|
UNEXPECTED_EXCEPTION = "UNEXPECTED_EXCEPTION"
|
|
|
class TerminalStatus(str, Enum):
|
STATE_UNCERTAIN = "STATE_UNCERTAIN"
|
BLOCKED_ACCESS_CONTROL = "BLOCKED_ACCESS_CONTROL"
|
TIME_BUDGET_STOP = "TIME_BUDGET_STOP"
|
PARTIAL_QUOTA_STOP = "PARTIAL_QUOTA_STOP"
|
BLOCKED_INPUT = "BLOCKED_INPUT"
|
BLOCKED_ENVIRONMENT = "BLOCKED_ENVIRONMENT"
|
BLOCKED_AMBIGUOUS_MAPPING = "BLOCKED_AMBIGUOUS_MAPPING"
|
VALIDATION_FAILED = "VALIDATION_FAILED"
|
INTERNAL_ERROR = "INTERNAL_ERROR"
|
PARTIAL_SUCCESS = "PARTIAL_SUCCESS"
|
SUCCESS = "SUCCESS"
|
|
|
TERMINAL_MAPPING: dict[TerminalStatus, tuple[str, int]] = {
|
TerminalStatus.STATE_UNCERTAIN: ("STATE_UNCERTAIN", 27),
|
TerminalStatus.BLOCKED_ACCESS_CONTROL: ("BLOCKED_ACCESS_CONTROL", 13),
|
TerminalStatus.TIME_BUDGET_STOP: ("TIME_BUDGET_STOP", 10),
|
TerminalStatus.PARTIAL_QUOTA_STOP: ("PARTIAL_QUOTA_STOP", 11),
|
TerminalStatus.BLOCKED_INPUT: ("BLOCKED_INPUT", 12),
|
TerminalStatus.BLOCKED_ENVIRONMENT: ("BLOCKED_ENVIRONMENT", 14),
|
TerminalStatus.BLOCKED_AMBIGUOUS_MAPPING: ("BLOCKED_AMBIGUOUS_MAPPING", 15),
|
TerminalStatus.VALIDATION_FAILED: ("VALIDATION_FAILED", 16),
|
TerminalStatus.INTERNAL_ERROR: ("INTERNAL_ERROR", 20),
|
TerminalStatus.PARTIAL_SUCCESS: ("ACCEPTED", 2),
|
TerminalStatus.SUCCESS: ("ACCEPTED", 0),
|
}
|
TERMINAL_PRECEDENCE = tuple(TERMINAL_MAPPING)
|
|
|
class EvidenceState(str, Enum):
|
N = "N"
|
V = "V"
|
I = "I"
|
U = "U"
|
|
|
class PackageState(str, Enum):
|
P00_INIT = "P00_INIT"
|
P01_RESERVED = "P01_RESERVED"
|
P02_TRIGGER_UNKNOWN = "P02_TRIGGER_UNKNOWN"
|
P03_CACHE_MATCHED = "P03_CACHE_MATCHED"
|
P04_STAGING_PARTIAL = "P04_STAGING_PARTIAL"
|
P05_STAGING_VALID = "P05_STAGING_VALID"
|
P06_PUBLISHED = "P06_PUBLISHED"
|
P07_MANIFESTED = "P07_MANIFESTED"
|
P08_DELIVERY = "P08_DELIVERY"
|
P09_TIMING = "P09_TIMING"
|
P10_CLOSED = "P10_CLOSED"
|
|
|
class Lane(str, Enum):
|
NEW = "NEW_TRIGGER"
|
PURE = "PURE_REUSE"
|
|
|
class ItemStatus(str, Enum):
|
SUCCESS = "SUCCESS"
|
DUPLICATE = "DUPLICATE"
|
FAILED = "FAILED"
|
STOPPED = "STOPPED"
|
|
|
class QuotaState(str, Enum):
|
NONE = "NONE"
|
RESERVED = "RESERVED"
|
CONFIRMED = "CONFIRMED"
|
UNCERTAIN = "UNCERTAIN"
|
RELEASED = "RELEASED"
|
|
|
HASH_RE = re.compile(r"^[0-9a-f]{64}$")
|
ID_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,200}$")
|
|
|
TASK_KEYS_V003 = (
|
"schema_version", "contract_version", "project_id", "handoff_id", "task_id",
|
"source_role_instance_id", "source_thread_id", "target_role_instance_id",
|
"target_thread_id", "reply_thread_id", "requester", "review_owner", "mode",
|
"query", "quantity", "aliases", "analysts", "institutions", "report_types",
|
"date_range", "minimum_pages", "exclude", "source_scope", "destination",
|
"priority", "naming_requirement", "output_root", "quota_ledger", "adb_executable",
|
"pdfinfo_executable", "package_name", "cache_root", "device_serial",
|
"observed_at_utc", "total_budget_ms", "close_reserve_ms",
|
"batch_increment_budget_ms", "min_screens", "normal_max_screens",
|
"hard_max_screens", "hard_max_candidates", "performance_slot_id",
|
"performance_plan_id",
|
)
|
|
TASK_KEYS = TASK_KEYS_V003 + ("expected_reports",)
|
|
EXPECTED_REPORT_KEYS = (
|
"report_identity", "title", "institution", "report_date", "page_count",
|
)
|
|
|
@dataclass(frozen=True)
|
class TaskSpec:
|
raw: Mapping[str, Any]
|
|
@classmethod
|
def from_mapping(cls, value: Mapping[str, Any], *, exact: bool = True) -> "TaskSpec":
|
keys = tuple(value.keys())
|
schema = value.get("schema_version")
|
expected_keys = TASK_KEYS if schema == "HIBOR_FAST_TASK_SPEC_V004" else TASK_KEYS_V003
|
if schema not in {"HIBOR_FAST_TASK_SPEC_V003", "HIBOR_FAST_TASK_SPEC_V004"}:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "schema_version", "V003 or V004 required")
|
if exact and keys != expected_keys:
|
missing = [k for k in expected_keys if k not in value]
|
extra = [k for k in value if k not in expected_keys]
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "keys", f"order/missing/extra:{missing}:{extra}")
|
required = {"project_id", "task_id", "handoff_id", "mode", "query", "quantity",
|
"output_root", "quota_ledger", "adb_executable", "pdfinfo_executable",
|
"package_name", "cache_root", "total_budget_ms", "close_reserve_ms"}
|
if missing := required.difference(value):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "keys", f"missing:{sorted(missing)}")
|
if value["project_id"] != "project-info":
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "project_id", "must be project-info")
|
if value["package_name"] != "cn.com.hibor":
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "package_name", "must be cn.com.hibor")
|
if value["cache_root"] != "/sdcard/Android/data/cn.com.hibor/files/myfile/":
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "cache_root", "unexpected cache")
|
if value["mode"] not in {"dry-run", "discover", "collect-one", "batch", "resume-postprocess"}:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "mode", "unsupported")
|
quantity = value["quantity"]
|
if type(quantity) is not int or quantity < 1 or quantity > 10:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "quantity", "must be 1..10")
|
total = value["total_budget_ms"]
|
expected = 600_000 if quantity == 1 else 600_000 + 240_000 * (quantity - 1)
|
if type(total) is not int or total != expected:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "total_budget_ms", f"expected {expected}")
|
if value["close_reserve_ms"] < 15_000 or value["close_reserve_ms"] >= total:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "close_reserve_ms", "outside safe range")
|
if type(value.get("batch_increment_budget_ms")) is not int or value["batch_increment_budget_ms"] != 240_000:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "batch_increment_budget_ms", "must be 240000")
|
if not isinstance(value.get("observed_at_utc"), str) or not value["observed_at_utc"].endswith("Z"):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "observed_at_utc", "UTC Z timestamp required")
|
if not isinstance(value.get("source_scope"), Mapping):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "source_scope", "mapping required")
|
for path_key in ("output_root", "quota_ledger", "adb_executable", "pdfinfo_executable"):
|
if not isinstance(value[path_key], str) or not value[path_key]:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, path_key, "non-empty string required")
|
expected_reports = value.get("expected_reports", [])
|
if not isinstance(expected_reports, list):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "expected_reports", "array required")
|
identities: set[str] = set()
|
for index, report in enumerate(expected_reports):
|
if not isinstance(report, Mapping) or tuple(report.keys()) != EXPECTED_REPORT_KEYS:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID,
|
f"expected_reports[{index}]", "exact ordered report keys required")
|
identity = report["report_identity"]
|
if not isinstance(identity, str) or not HASH_RE.fullmatch(identity) or identity in identities:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID,
|
f"expected_reports[{index}].report_identity", "unique SHA-256 required")
|
identities.add(identity)
|
for field_name in ("title", "institution", "report_date"):
|
if not isinstance(report[field_name], str) or not report[field_name].strip():
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID,
|
f"expected_reports[{index}].{field_name}", "non-empty string required")
|
if type(report["page_count"]) is not int or report["page_count"] < 1:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID,
|
f"expected_reports[{index}].page_count", "positive INT required")
|
if value["mode"] in {"collect-one", "batch"}:
|
if schema != "HIBOR_FAST_TASK_SPEC_V004" or len(expected_reports) != quantity:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "expected_reports",
|
"V004 collection requires one exact report per item")
|
elif expected_reports:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "expected_reports",
|
"only collect-one/batch may bind reports")
|
return cls(dict(value))
|
|
def __getattr__(self, name: str) -> Any:
|
try:
|
return self.raw[name]
|
except KeyError as exc:
|
raise AttributeError(name) from exc
|
|
@property
|
def output_path(self) -> Path:
|
return Path(self.raw["output_root"])
|
|
@property
|
def quota_path(self) -> Path:
|
return Path(self.raw["quota_ledger"])
|
|
|
@dataclass(frozen=True)
|
class PackageEvidence:
|
lane: Lane
|
state: PackageState
|
staging: EvidenceState
|
final: EvidenceState
|
quota_terminal: EvidenceState
|
artifact_terminal: EvidenceState
|
manifest: EvidenceState
|
delivery: EvidenceState
|
timing: EvidenceState
|
terminal: EvidenceState
|
|
def any_unknown(self) -> bool:
|
return EvidenceState.U in (
|
self.staging, self.final, self.quota_terminal, self.artifact_terminal,
|
self.manifest, self.delivery, self.timing, self.terminal,
|
)
|
|
|
@dataclass(frozen=True)
|
class TerminalReceipt:
|
target_path: Path
|
terminal_path: Path | None
|
persist_attempted: bool
|
persisted: bool | None
|
presence: EvidenceState
|
operation_completed: bool | None
|
liveness_unknown: bool
|
stdout_emitted: bool
|
bytes: int | None
|
sha256: str | None
|
error_code: ErrorCode | None
|
source_state: PackageState
|
|
def validate(self) -> None:
|
if self.presence is EvidenceState.V:
|
valid = (self.persist_attempted and self.persisted is True and
|
self.operation_completed is True and not self.liveness_unknown and
|
self.terminal_path == self.target_path and self.bytes is not None and
|
self.bytes >= 2 and self.sha256 is not None and HASH_RE.fullmatch(self.sha256) and
|
self.error_code is None)
|
elif self.presence is EvidenceState.U:
|
pair = ((self.operation_completed is True and not self.liveness_unknown and
|
self.error_code is ErrorCode.RECOVERY_UNKNOWN) or
|
(self.operation_completed is None and self.liveness_unknown and
|
self.error_code is ErrorCode.PROCESS_LIVENESS_UNKNOWN))
|
valid = (self.persist_attempted and self.persisted is None and pair and
|
self.terminal_path is None and self.bytes is None and self.sha256 is None)
|
elif self.presence is EvidenceState.I:
|
known_pair = (self.bytes is None and self.sha256 is None) or (
|
self.bytes is not None and self.sha256 is not None and HASH_RE.fullmatch(self.sha256))
|
valid = (self.persist_attempted and self.persisted is False and
|
self.operation_completed is True and not self.liveness_unknown and
|
self.terminal_path is None and self.error_code is ErrorCode.PERSIST_LATE and known_pair)
|
else:
|
if self.persist_attempted:
|
valid = (self.persisted is False and self.operation_completed is True and
|
not self.liveness_unknown and self.terminal_path is None and
|
self.bytes is None and self.sha256 is None and
|
self.error_code is ErrorCode.PERSIST_LATE)
|
else:
|
valid = (self.persisted is False and self.operation_completed is False and
|
not self.liveness_unknown and self.terminal_path is None and
|
self.bytes is None and self.sha256 is None)
|
if not valid:
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "TerminalReceipt", "cross-field mismatch")
|
|
|
@dataclass(frozen=True)
|
class Candidate:
|
candidate_id: str
|
title: str
|
institution: str | None
|
report_date: str | None
|
analysts: tuple[str, ...] = field(default_factory=tuple)
|
page_count: int | None = None
|
score: int = 0
|
bounds: tuple[int, int, int, int] | None = None
|
|
def __post_init__(self) -> None:
|
if not ID_RE.fullmatch(self.candidate_id):
|
raise ContractError(ErrorCode.TASK_SPEC_INVALID, "candidate_id", "invalid")
|
