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