from __future__ import annotations from contextlib import contextmanager from dataclasses import dataclass from datetime import date, datetime, timezone import csv import hashlib import os import time from pathlib import Path from typing import Callable, Iterator, Mapping from .models import ContractError, ErrorCode QUOTA_COLUMNS = ( "quota_date", "timezone", "platform_limit", "automation_target", "automation_hard_stop", "reserved_buffer", "task_id", "requester_role", "handoff_id", "event_at", "report_identity", "event_type", "confirmed_consumed", "uncertain_consumed", "active_reservation_delta", "success_unique_pdf_delta", "duplicate_or_failed_delta", "cumulative_consumed", "safe_available_after", "note", "event_family", "schema_version", "event_id", "idempotency_key", "event_seq", "run_id", "slot_id", "reservation_id", "ref_event_id", "evidence_ref", "external_baseline_floor", "confirmed_delta", "uncertain_delta", "app_total_consumed", ) EVENT_FAMILY = { "BASELINE_ESTIMATE": "BASELINE", "CORRECTION_RAISE": "BASELINE", "APP_RECONCILE": "APP_OBSERVATION", "RESERVE": "RESERVATION", "CONSUME_CONFIRMED": "QUOTA_TERMINAL", "CONSUME_UNCERTAIN": "QUOTA_TERMINAL", "RELEASE": "QUOTA_TERMINAL", "ARTIFACT_SUCCESS": "ARTIFACT_TERMINAL", "ARTIFACT_DUPLICATE_OR_FAILED": "ARTIFACT_TERMINAL", } @dataclass(frozen=True) class QuotaSnapshot: quota_date: str confirmed: int uncertain: int active: int external_floor: int app_total: int effective_consumed: int safe_available: int row_count: int @dataclass(frozen=True) class Reservation: reservation_id: str | None allowed: bool replayed: bool snapshot: QuotaSnapshot stop_code: ErrorCode | None terminal_event_id: str | None = None terminal_event_type: str | None = None @dataclass(frozen=True) class AppQuotaObservation: observation_id: str device_serial: str captured_at_utc: str visible_remaining: int app_total_consumed: int ui_snapshot_fingerprint: str @classmethod def build(cls, *, device_serial: str, quota_date: str, captured_at_utc: str, visible_remaining: int, ui_snapshot_fingerprint: str) -> "AppQuotaObservation": if not device_serial or not captured_at_utc or not ui_snapshot_fingerprint: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "app_observation", "missing evidence") if type(visible_remaining) is not int or not 0 <= visible_remaining <= 30: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "visible_remaining", "outside 0..30") preimage = f"{device_serial}|{quota_date}|{captured_at_utc}|{visible_remaining}|{ui_snapshot_fingerprint}" observation_id = _hash(preimage) return cls(observation_id, device_serial, captured_at_utc, visible_remaining, 30 - visible_remaining, ui_snapshot_fingerprint) def _hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() class QuotaLedger: platform_limit = 30 automation_target = 25 automation_hard_stop = 27 reserved_buffer = 3 def __init__(self, path: Path, *, quota_date: str | None = None, timeout_provider: Callable[[str, int], int] | None = None): self.path = Path(path) self.quota_date = quota_date or date.today().isoformat() self.timeout_provider = timeout_provider def initialize(self, *, task_id: str, requester_role: str, handoff_id: str, known_floor: int = 3, evidence_ref: str = "KNOWN-2026-07-29-CHINA-DUTYFREE-3") -> QuotaSnapshot: if known_floor < 0 or known_floor > self.platform_limit: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "known_floor", "outside 0..30") self.path.parent.mkdir(parents=True, exist_ok=True) try: fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_BINARY", 0), 0o600) except FileExistsError: return self.snapshot() try: with os.fdopen(fd, "w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=QUOTA_COLUMNS, lineterminator="\r\n") writer.writeheader() event = self._event( event_type="BASELINE_ESTIMATE", task_id=task_id, requester_role=requester_role, handoff_id=handoff_id, report_identity="__BASELINE__", run_id="BASELINE", slot_id="", reservation_id="", ref_event_id="", evidence_ref=evidence_ref, external_baseline_floor=known_floor, confirmed_delta=0, uncertain_delta=0, active_delta=0, success_delta=0, duplicate_delta=0, event_seq=1, cumulative=known_floor, safe=self._safe(known_floor, 0), app_total=0, ) writer.writerow(event) stream.flush() os.fsync(stream.fileno()) except BaseException: raise return self.snapshot() def snapshot(self) -> QuotaSnapshot: with self._locked("r") as stream: rows = self._read_rows(stream) return self._fold(rows) def observe_app_remaining(self, *, task_id: str, requester_role: str, handoff_id: str, observation: AppQuotaObservation) -> Mapping[str, str]: rebuilt = AppQuotaObservation.build( device_serial=observation.device_serial, quota_date=self.quota_date, captured_at_utc=observation.captured_at_utc, visible_remaining=observation.visible_remaining, ui_snapshot_fingerprint=observation.ui_snapshot_fingerprint, ) if rebuilt != observation: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "app_observation", "identity mismatch") key = _hash(f"{self.quota_date}|APP_OBSERVATION|{observation.observation_id}") with self._locked("r+") as stream: rows = self._read_rows(stream) existing = self._find(rows, "APP_OBSERVATION", key) if existing: if (existing["app_total_consumed"] != str(observation.app_total_consumed) or existing["evidence_ref"] != f"APP-OBS:{observation.observation_id}"): raise ContractError(ErrorCode.QUOTA_REPLAY_CONFLICT, "app_observation", "row drift") return existing snap = self._fold(rows) app_total = max(snap.app_total, observation.app_total_consumed) effective = max(snap.external_floor + snap.confirmed + snap.uncertain, app_total) event = self._event( event_type="APP_RECONCILE", task_id=task_id, requester_role=requester_role, handoff_id=handoff_id, report_identity="__APP_TOTAL__", run_id="APP-OBSERVATION", slot_id="", reservation_id="", ref_event_id="", evidence_ref=f"APP-OBS:{observation.observation_id}", external_baseline_floor=0, confirmed_delta=0, uncertain_delta=0, active_delta=0, success_delta=0, duplicate_delta=0, event_seq=len(rows)+1, cumulative=effective, safe=self._safe(effective, snap.active), app_total=observation.app_total_consumed, idempotency_key=key, note=f"VISIBLE_REMAINING={observation.visible_remaining}", ) self._append(stream, event) self._verify_appended(stream, event) return event def reserve(self, *, task_id: str, requester_role: str, handoff_id: str, run_id: str, slot_id: str, report_identity: str, observation_event_id: str = "") -> Reservation: preimage = f"{self.quota_date}|RESERVATION|{task_id}|{handoff_id}|{slot_id}|{report_identity}" key = _hash(preimage) with self._locked("r+") as stream: rows = self._read_rows(stream) existing = self._find(rows, "RESERVATION", key) snap = self._fold(rows) if existing: terminals = [row for row in rows if row["event_family"] == "QUOTA_TERMINAL" and row["ref_event_id"] == existing["event_id"]] if len(terminals) > 1: raise ContractError(ErrorCode.QUOTA_REPLAY_CONFLICT, "reservation", "multiple terminals") terminal = terminals[0] if terminals else None return Reservation(existing["event_id"], False, True, snap, ErrorCode.QUOTA_REPLAY_CONFLICT, terminal["event_id"] if terminal else None, terminal["event_type"] if terminal else None) if observation_event_id: observations = [row for row in rows if row["event_family"] == "APP_OBSERVATION" and row["event_id"] == observation_event_id and row["quota_date"] == self.quota_date] if len(observations) != 1: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "observation_event_id", "not committed") if snap.safe_available <= 0: return Reservation(None, False, False, snap, ErrorCode.QUOTA_EXHAUSTED) event = self._event( event_type="RESERVE", task_id=task_id, requester_role=requester_role, handoff_id=handoff_id, report_identity=report_identity, run_id=run_id, slot_id=slot_id, reservation_id="", ref_event_id=observation_event_id, evidence_ref="", external_baseline_floor=0, confirmed_delta=0, uncertain_delta=0, active_delta=1, success_delta=0, duplicate_delta=0, event_seq=len(rows)+1, cumulative=snap.effective_consumed, safe=self._safe(snap.effective_consumed, snap.active+1), app_total=snap.app_total, idempotency_key=key, ) event["reservation_id"] = event["event_id"] self._append(stream, event) self._verify_appended(stream, event) after = self._fold(rows + [event]) return Reservation(event["event_id"], True, False, after, None) def terminal(self, reservation: Reservation, *, event_type: str, task_id: str, requester_role: str, handoff_id: str, run_id: str, slot_id: str, report_identity: str, note: str) -> Mapping[str, str]: if not reservation.allowed or not reservation.reservation_id: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "reservation", "not allowed") if event_type not in {"CONSUME_CONFIRMED", "CONSUME_UNCERTAIN", "RELEASE"}: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "event_type", "invalid terminal") key = _hash(f"{self.quota_date}|QUOTA_TERMINAL|{reservation.reservation_id}") with self._locked("r+") as stream: rows = self._read_rows(stream) existing = self._find(rows, "QUOTA_TERMINAL", key) if existing: if existing["event_type"] != event_type: raise ContractError(ErrorCode.QUOTA_REPLAY_CONFLICT, "event_type", "terminal race") return existing snap = self._fold(rows) confirmed = 1 if event_type == "CONSUME_CONFIRMED" else 0 uncertain = 1 if event_type == "CONSUME_UNCERTAIN" else 0 cumulative = max(snap.external_floor + snap.confirmed + snap.uncertain + confirmed + uncertain, snap.app_total) event = self._event( event_type=event_type, task_id=task_id, requester_role=requester_role, handoff_id=handoff_id, report_identity=report_identity, run_id=run_id, slot_id=slot_id, reservation_id=reservation.reservation_id, ref_event_id=reservation.reservation_id, evidence_ref="", external_baseline_floor=0, confirmed_delta=confirmed, uncertain_delta=uncertain, active_delta=-1, success_delta=0, duplicate_delta=0, event_seq=len(rows)+1, cumulative=cumulative, safe=self._safe(cumulative, max(0, snap.active-1)), app_total=snap.app_total, idempotency_key=key, note=note, ) self._append(stream, event) self._verify_appended(stream, event) return event def artifact(self, *, task_id: str, requester_role: str, handoff_id: str, run_id: str, slot_id: str, report_identity: str, quota_terminal_event_id: str, success: bool, note: str) -> Mapping[str, str]: key = _hash(f"{self.quota_date}|ARTIFACT_TERMINAL|{task_id}|{handoff_id}|{run_id}|{slot_id}|{report_identity}") event_type = "ARTIFACT_SUCCESS" if success else "ARTIFACT_DUPLICATE_OR_FAILED" with self._locked("r+") as stream: rows = self._read_rows(stream) existing = self._find(rows, "ARTIFACT_TERMINAL", key) if existing: if existing["event_type"] != event_type: raise ContractError(ErrorCode.QUOTA_REPLAY_CONFLICT, "artifact", "outcome conflict") return existing snap = self._fold(rows) event = self._event( event_type=event_type, task_id=task_id, requester_role=requester_role, handoff_id=handoff_id, report_identity=report_identity, run_id=run_id, slot_id=slot_id, reservation_id="", ref_event_id=quota_terminal_event_id, evidence_ref="", external_baseline_floor=0, confirmed_delta=0, uncertain_delta=0, active_delta=0, success_delta=1 if success else 0, duplicate_delta=0 if success else 1, event_seq=len(rows)+1, cumulative=snap.effective_consumed, safe=snap.safe_available, app_total=snap.app_total, idempotency_key=key, note=note, ) self._append(stream, event) self._verify_appended(stream, event) return event def _event(self, *, event_type: str, task_id: str, requester_role: str, handoff_id: str, report_identity: str, run_id: str, slot_id: str, reservation_id: str, ref_event_id: str, evidence_ref: str, external_baseline_floor: int, confirmed_delta: int, uncertain_delta: int, active_delta: int, success_delta: int, duplicate_delta: int, event_seq: int, cumulative: int, safe: int, app_total: int, idempotency_key: str | None = None, note: str = "") -> dict[str, str]: family = EVENT_FAMILY[event_type] key = idempotency_key or _hash( f"{self.quota_date}|BASELINE|{event_type}|{ref_event_id or 'NONE'}|{evidence_ref}|{external_baseline_floor}") event_id = _hash(f"HIBOR-QUOTA-EVENT-V004|{key}|{event_type}") now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") data = {name: "" for name in QUOTA_COLUMNS} data.update({ "quota_date": self.quota_date, "timezone": "Asia/Shanghai", "platform_limit": str(self.platform_limit), "automation_target": str(self.automation_target), "automation_hard_stop": str(self.automation_hard_stop), "reserved_buffer": str(self.reserved_buffer), "task_id": task_id, "requester_role": requester_role, "handoff_id": handoff_id, "event_at": now, "report_identity": report_identity, "event_type": event_type, "confirmed_consumed": str(max(0, confirmed_delta)), "uncertain_consumed": str(max(0, uncertain_delta)), "active_reservation_delta": str(active_delta), "success_unique_pdf_delta": str(success_delta), "duplicate_or_failed_delta": str(duplicate_delta), "cumulative_consumed": str(cumulative), "safe_available_after": str(safe), "note": note, "event_family": family, "schema_version": "HIBOR_QUOTA_EVENT_V004", "event_id": event_id, "idempotency_key": key, "event_seq": str(event_seq), "run_id": run_id, "slot_id": slot_id, "reservation_id": reservation_id, "ref_event_id": ref_event_id, "evidence_ref": evidence_ref, "external_baseline_floor": str(external_baseline_floor), "confirmed_delta": str(confirmed_delta), "uncertain_delta": str(uncertain_delta), "app_total_consumed": str(app_total), }) return data @contextmanager def _locked(self, mode: str) -> Iterator[object]: if not self.path.exists(): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "path", "ledger absent") stream = self.path.open(mode, encoding="utf-8", newline="") acquired = False try: timeout_ms = (self.timeout_provider("quota_lock", 5_000) if self.timeout_provider is not None else 5_000) deadline = time.monotonic() + timeout_ms / 1000 if os.name == "nt": import msvcrt while True: try: stream.seek(0) msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) acquired = True break except OSError as exc: if time.monotonic() >= deadline: raise ContractError(ErrorCode.LOCK_TIMEOUT, "quota", "lock deadline") from exc time.sleep(min(0.01, max(0.0, deadline - time.monotonic()))) else: import fcntl while True: try: fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) acquired = True break except BlockingIOError as exc: if time.monotonic() >= deadline: raise ContractError(ErrorCode.LOCK_TIMEOUT, "quota", "lock deadline") from exc time.sleep(min(0.01, max(0.0, deadline - time.monotonic()))) yield stream finally: try: if acquired and os.name == "nt": import msvcrt stream.seek(0) msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1) elif acquired: import fcntl fcntl.flock(stream.fileno(), fcntl.LOCK_UN) finally: stream.close() @staticmethod def _read_rows(stream: object) -> list[dict[str, str]]: stream.seek(0) reader = csv.DictReader(stream) if tuple(reader.fieldnames or ()) != QUOTA_COLUMNS: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "header", "schema mismatch") rows = list(reader) if any(None in row for row in rows): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "row", "column overflow") return rows @staticmethod def _find(rows: list[dict[str, str]], family: str, key: str) -> dict[str, str] | None: found = [r for r in rows if r["event_family"] == family and r["idempotency_key"] == key] if len(found) > 1: raise ContractError(ErrorCode.QUOTA_REPLAY_CONFLICT, "idempotency_key", "duplicate rows") return found[0] if found else None def _fold(self, rows: list[dict[str, str]]) -> QuotaSnapshot: confirmed = uncertain = active = external = app_total = 0 terminal_reservations: set[str] = set() active_reservations: set[str] = set() for index, row in enumerate(rows, 1): if row["quota_date"] != self.quota_date or row["event_seq"] != str(index): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "event_seq", "date/order mismatch") external = max(external, int(row["external_baseline_floor"] or 0)) app_total = max(app_total, int(row["app_total_consumed"] or 0)) confirmed += int(row["confirmed_delta"] or 0) uncertain += int(row["uncertain_delta"] or 0) if row["event_family"] == "RESERVATION": active_reservations.add(row["event_id"]) elif row["event_family"] == "QUOTA_TERMINAL": terminal_reservations.add(row["ref_event_id"]) active = len(active_reservations - terminal_reservations) local_total = external + confirmed + uncertain effective = max(local_total, app_total) return QuotaSnapshot(self.quota_date, confirmed, uncertain, active, external, app_total, effective, self._safe(effective, active), len(rows)) def _safe(self, effective: int, active: int) -> int: return max(0, min(self.automation_target, self.automation_hard_stop) - effective - active) @staticmethod def _append(stream: object, event: Mapping[str, str]) -> None: stream.seek(0, os.SEEK_END) writer = csv.DictWriter(stream, fieldnames=QUOTA_COLUMNS, lineterminator="\r\n") writer.writerow(event) stream.flush() os.fsync(stream.fileno()) def _verify_appended(self, stream: object, event: Mapping[str, str]) -> None: stream.flush() stream.seek(0) rows = self._read_rows(stream) matches = [row for row in rows if row["event_id"] == event["event_id"]] if len(matches) != 1 or matches[0] != dict(event): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "append", "reopen mismatch") stream.seek(0, os.SEEK_END)