"""Append-only local queue and reload state for generic authenticated jobs.""" from __future__ import annotations import hashlib import json import os import secrets from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Iterator from .constants import ( COMPLETION_CLOSURE_REQUIRED, EXTENSION_BUILD, ERROR_CODE_RE, HANDOFF_ID_RE, JOB_ID_RE, MESSAGE_ID_RE, MAX_CLAIM_ATTEMPTS, INTERNAL_ONLY_PAGE_PENDING_CODES, LEGACY_MISSING_DIAGNOSTIC_REPLAY_LINES, LEGACY_PAGE_METADATA_REPLAY_LINES, LEGACY_PAGE_METADATA_REPLAY_PREFIX_BYTES, LEGACY_PAGE_METADATA_REPLAY_PREFIX_SHA256, LEGACY_PAGE_METADATA_REPLAY_STATE_SUFFIX, PREPARELESS_REJECT_CODES, QUEUE_LEASE_SECONDS, QUEUE_SCHEMA_VERSION, SUCCESSOR_QUEUE_SCHEMA_VERSION, UPPER_SHA256_RE, RELOAD_BACKOFF_SECONDS, canonical_url, stable_job_id, stable_successor_job_id, validate_bvid, validate_creator_uid, validate_prepareless_terminal, validate_postprocess_terminal, validate_runtime_diagnostic, ) from .protocol import ( ProtocolError, encode_json, strict_json_loads, validate_job, validate_media_complete_identity, ) @dataclass(frozen=True) class AuthorizedSuccessor: creator_uid: str bvid: str predecessor_job_id: str retry_generation: int terminal_error_code: str @dataclass(frozen=True) class ReleaseApproval: """Immutable value emitted only by the producer's deployment trust gate.""" authorization_message_id: str authorization_handoff_id: str authorization_sha256: str repair_review_result_message_id: str repair_audit_id: str repair_audit_bytes: int repair_audit_sha256: str successors: tuple[AuthorizedSuccessor, ...] def _read_jsonl(path: Path, *, missing_ok: bool) -> list[dict[str, Any]]: if not path.exists(): if missing_ok: return [] raise ProtocolError("E_QUEUE") if not path.is_file() or path.is_symlink(): raise ProtocolError("E_QUEUE") payload = path.read_bytes() if len(payload) > 16 * 1024 * 1024: raise ProtocolError("E_QUEUE") if payload and not payload.endswith(b"\n"): raise ProtocolError("E_QUEUE_PARTIAL") result: list[dict[str, Any]] = [] for line in payload.splitlines(): if not line or len(line) > 4096: raise ProtocolError("E_QUEUE") result.append(strict_json_loads(line)) return result def _append_jsonl(path: Path, value: dict[str, Any]) -> None: payload = encode_json(value, 4096) + b"\n" with path.open("ab", buffering=0) as stream: if stream.write(payload) != len(payload): raise ProtocolError("E_QUEUE_WRITE") stream.flush() os.fsync(stream.fileno()) def _media_complete_durability_test_seam(_stage: str) -> None: """No-op production seam for crash-window durability counterexamples.""" return None def _postprocess_recovery_claim_test_seam(_stage: str) -> None: """No-op production seam for recovery-claim crash counterexamples.""" return None def _append_jsonl_batch(path: Path, values: list[dict[str, Any]]) -> None: if not values: return payload = b"".join(encode_json(value, 4096) + b"\n" for value in values) with path.open("ab", buffering=0) as stream: if stream.write(payload) != len(payload): raise ProtocolError("E_QUEUE_WRITE") stream.flush() os.fsync(stream.fileno()) def _append_bytes(path: Path, payload: bytes) -> None: if not payload: return with path.open("ab", buffering=0) as stream: if stream.write(payload) != len(payload): raise ProtocolError("E_QUEUE_WRITE") stream.flush() os.fsync(stream.fileno()) def _canonical_line(value: dict[str, Any]) -> bytes: return encode_json(value, 4096) + b"\n" @dataclass(frozen=True) class _QueueSnapshot: jobs: dict[str, dict[str, Any]] records: dict[str, dict[str, Any]] children: dict[str, str] blocks: tuple[tuple[str, str, str, tuple[str, ...], bytes], ...] committed_bytes: bytes pending_suffix: bytes _LINEAGE_KEYS = { "predecessor_job_id", "retry_generation", "predecessor_terminal_error_code", "authorization_message_id", "authorization_handoff_id", "authorization_sha256", "repair_review_result_message_id", "repair_audit_id", "repair_audit_bytes", "repair_audit_sha256", } _SUCCESSOR_RECORD_KEYS = { "schema", "record_type", "job_id", "bvid", "creator_uid", "expected_duration_ms", "discovered_at_unix_ms", "published_at", "title", "lineage", } _BEGIN_KEYS = { "schema", "record_type", "authorization_message_id", "authorization_handoff_id", "authorization_sha256", "successor_count", } _COMMIT_KEYS = _BEGIN_KEYS | {"block_sha256"} def _validate_authorization_identity(message_id: Any, handoff_id: Any, sha256: Any) -> None: if ( not isinstance(message_id, str) or not MESSAGE_ID_RE.fullmatch(message_id) or not isinstance(handoff_id, str) or not HANDOFF_ID_RE.fullmatch(handoff_id) or not isinstance(sha256, str) or not UPPER_SHA256_RE.fullmatch(sha256) ): raise ProtocolError("E_LINEAGE") def _successor_record(value: dict[str, Any], allowed_creators: frozenset[str]) -> dict[str, Any]: if set(value) != _SUCCESSOR_RECORD_KEYS or value.get("schema") != SUCCESSOR_QUEUE_SCHEMA_VERSION or value.get("record_type") != "SUCCESSOR_JOB": raise ProtocolError("E_LINEAGE") creator = value.get("creator_uid") if creator not in allowed_creators: raise ProtocolError("E_ALLOWLIST") discovered = value.get("discovered_at_unix_ms") if isinstance(discovered, bool) or not isinstance(discovered, int) or discovered <= 0: raise ProtocolError("E_LINEAGE") lineage = value.get("lineage") if not isinstance(lineage, dict) or set(lineage) != _LINEAGE_KEYS: raise ProtocolError("E_LINEAGE") runtime = { "job_id": value["job_id"], "bvid": value["bvid"], "creator_uid": creator, "canonical_url": canonical_url(value["bvid"]), "expected_duration_ms": value["expected_duration_ms"], "published_at": value["published_at"], "title": value["title"], "lineage": lineage, } try: validate_job(runtime) except (ProtocolError, ValueError) as exc: raise ProtocolError("E_LINEAGE") from exc return runtime def _queue_snapshot(payload: bytes, allowed_creators: frozenset[str]) -> _QueueSnapshot: """Parse committed schema-1 records and schema-2 blocks, retaining one tail suffix.""" if len(payload) > 16 * 1024 * 1024: raise ProtocolError("E_QUEUE") jobs: dict[str, dict[str, Any]] = {} records: dict[str, dict[str, Any]] = {} ingress: dict[str, dict[str, Any]] = {} children: dict[str, str] = {} blocks: list[tuple[str, str, str, tuple[str, ...], bytes]] = [] lines: list[tuple[int, int, bytes, dict[str, Any]]] = [] offset = 0 for raw_line in payload.splitlines(keepends=True): start = offset offset += len(raw_line) if not raw_line.endswith(b"\n"): break content = raw_line[:-1] if content.endswith(b"\r"): content = content[:-1] if not content or len(content) > 4096: raise ProtocolError("E_QUEUE") value = strict_json_loads(content) if not isinstance(value, dict): raise ProtocolError("E_QUEUE") lines.append((start, offset, raw_line, value)) index = 0 committed_end = 0 while index < len(lines): start, end, raw_line, value = lines[index] if value.get("schema") == QUEUE_SCHEMA_VERSION: if value.get("record_type") is not None: raise ProtocolError("E_QUEUE") job = _ingress_job(value, allowed_creators) job_id = job["job_id"] previous = ingress.get(job_id) if previous is not None and previous != value: raise ProtocolError("E_QUEUE_CONFLICT") ingress.setdefault(job_id, value) prior_job = jobs.get(job_id) if prior_job is not None and prior_job != job: raise ProtocolError("E_QUEUE_CONFLICT") jobs.setdefault(job_id, job) records.setdefault(job_id, value) committed_end = end index += 1 continue if value.get("schema") != SUCCESSOR_QUEUE_SCHEMA_VERSION or value.get("record_type") != "SUCCESSOR_BEGIN": raise ProtocolError("E_LINEAGE_RECOVERY") if set(value) != _BEGIN_KEYS or raw_line != _canonical_line(value): raise ProtocolError("E_LINEAGE") _validate_authorization_identity( value.get("authorization_message_id"), value.get("authorization_handoff_id"), value.get("authorization_sha256"), ) count = value.get("successor_count") if isinstance(count, bool) or not isinstance(count, int) or not 1 <= count <= 100: raise ProtocolError("E_LINEAGE") needed = count + 2 if index + needed > len(lines): break block_lines = lines[index:index + needed] job_ids: list[str] = [] block_records: list[tuple[dict[str, Any], dict[str, Any]]] = [] for _, _, job_line, raw_job in block_lines[1:-1]: if job_line != _canonical_line(raw_job): raise ProtocolError("E_LINEAGE") runtime = _successor_record(raw_job, allowed_creators) lineage = runtime["lineage"] if ( lineage["authorization_message_id"] != value["authorization_message_id"] or lineage["authorization_handoff_id"] != value["authorization_handoff_id"] or lineage["authorization_sha256"] != value["authorization_sha256"] ): raise ProtocolError("E_LINEAGE") job_ids.append(runtime["job_id"]) block_records.append((raw_job, runtime)) if job_ids != sorted(job_ids) or len(set(job_ids)) != count: raise ProtocolError("E_LINEAGE") _, block_end, commit_line, commit = block_lines[-1] if set(commit) != _COMMIT_KEYS or commit.get("schema") != SUCCESSOR_QUEUE_SCHEMA_VERSION or commit.get("record_type") != "SUCCESSOR_COMMIT" or commit_line != _canonical_line(commit): raise ProtocolError("E_LINEAGE") for key in _BEGIN_KEYS - {"record_type"}: if commit.get(key) != value.get(key): raise ProtocolError("E_LINEAGE") block_prefix = b"".join(item[2] for item in block_lines[:-1]) block_hash = hashlib.sha256(block_prefix).hexdigest().upper() if commit.get("block_sha256") != block_hash: raise ProtocolError("E_LINEAGE") full_block = block_prefix + commit_line for raw_job, runtime in block_records: job_id = runtime["job_id"] parent = runtime["lineage"]["predecessor_job_id"] if job_id in jobs or parent in children: raise ProtocolError("E_LINEAGE_CONFLICT") jobs[job_id] = runtime records[job_id] = raw_job children[parent] = job_id blocks.append(( value["authorization_message_id"], value["authorization_handoff_id"], value["authorization_sha256"], tuple(job_ids), full_block, )) committed_end = block_end index += needed return _QueueSnapshot( jobs=jobs, records=records, children=children, blocks=tuple(blocks), committed_bytes=payload[:committed_end], pending_suffix=payload[committed_end:], ) def _ingress_job(value: dict[str, Any], allowed_creators: frozenset[str]) -> dict[str, Any]: expected = {"schema", "bvid", "creator_uid", "expected_duration_ms", "discovered_at_unix_ms", "published_at", "title"} if set(value) != expected or value.get("schema") != QUEUE_SCHEMA_VERSION: raise ProtocolError("E_QUEUE") try: bvid = validate_bvid(value["bvid"]) creator = validate_creator_uid(value["creator_uid"]) except ValueError as exc: raise ProtocolError("E_JOB") from exc duration = value["expected_duration_ms"] discovered = value["discovered_at_unix_ms"] if ( creator not in allowed_creators or isinstance(duration, bool) or not isinstance(duration, int) or not 1_000 <= duration <= 86_400_000 or isinstance(discovered, bool) or not isinstance(discovered, int) or discovered <= 0 ): raise ProtocolError("E_ALLOWLIST" if creator not in allowed_creators else "E_JOB") job = { "job_id": stable_job_id(creator, bvid), "bvid": bvid, "creator_uid": creator, "canonical_url": canonical_url(bvid), "expected_duration_ms": duration, "published_at": value["published_at"], "title": value["title"], } try: return validate_job(job) except ProtocolError as exc: raise ProtocolError("E_JOB") from exc def validate_ingress_record( value: dict[str, Any], allowed_creators: frozenset[str] ) -> dict[str, Any]: """Public producer/consumer boundary for one exact schema-1 record.""" return _ingress_job(value, allowed_creators) _EVENTS = { "CLAIMED", "STARTED", "MEDIA_COMPLETE", "POSTPROCESS_CLAIMED", "COMPLETE", "FAILED", "POSTPROCESS_FAILED", } _TERMINAL_EVENTS = {"COMPLETE", "FAILED", "POSTPROCESS_FAILED"} _POSTPROCESS_RECOVERY_KEYS = { "media_complete_event_sha256", "media_complete_lease_id", "media", } def _media_complete_event_sha256(value: dict[str, Any]) -> str: if value.get("event") != "MEDIA_COMPLETE": raise ProtocolError("E_QUEUE_STATE") return hashlib.sha256(_canonical_line(value)).hexdigest().upper() def _postprocess_recovery_binding(value: dict[str, Any]) -> dict[str, Any]: if value.get("event") != "MEDIA_COMPLETE" or "media" not in value: raise ProtocolError("E_QUEUE_STATE") return { "media_complete_event_sha256": _media_complete_event_sha256(value), "media_complete_lease_id": value["lease_id"], "media": value["media"], } def _validate_postprocess_recovery_binding( value: object, job: dict[str, Any], ) -> dict[str, Any]: if not isinstance(value, dict) or set(value) != _POSTPROCESS_RECOVERY_KEYS: raise ProtocolError("E_QUEUE_STATE") digest = value["media_complete_event_sha256"] prior_lease = value["media_complete_lease_id"] if not isinstance(digest, str) or UPPER_SHA256_RE.fullmatch(digest) is None: raise ProtocolError("E_QUEUE_STATE") if ( not isinstance(prior_lease, str) or len(prior_lease) != 32 or any(ch not in "0123456789abcdef" for ch in prior_lease) ): raise ProtocolError("E_QUEUE_STATE") try: media = validate_media_complete_identity(value["media"], job) except ProtocolError as exc: raise ProtocolError("E_QUEUE_STATE") from exc return { "media_complete_event_sha256": digest, "media_complete_lease_id": prior_lease, "media": media, } def _event( value: dict[str, Any], jobs: dict[str, dict[str, Any]], *, allow_exact_legacy_replay: bool = False, allow_exact_missing_diagnostic_replay: bool = False, ) -> dict[str, Any]: expected = { "schema", "event", "job_id", "bvid", "creator_uid", "lease_id", "at_unix_ms", "lease_expires_unix_ms", "error_code", } allowed_shapes = ( expected, expected | {"diagnostic"}, expected | {"media"}, expected | {"recovery"}, ) if set(value) not in allowed_shapes or value.get("schema") != QUEUE_SCHEMA_VERSION or value.get("event") not in _EVENTS: raise ProtocolError("E_QUEUE_STATE") if not isinstance(value["job_id"], str) or not JOB_ID_RE.fullmatch(value["job_id"]): raise ProtocolError("E_QUEUE_STATE") job = jobs.get(value["job_id"]) if job is None or value["creator_uid"] != job["creator_uid"] or value["bvid"] != job["bvid"]: raise ProtocolError("E_QUEUE_STATE") lease = value["lease_id"] if not isinstance(lease, str) or len(lease) != 32 or any(ch not in "0123456789abcdef" for ch in lease): raise ProtocolError("E_QUEUE_STATE") for name in ("at_unix_ms", "lease_expires_unix_ms"): if isinstance(value[name], bool) or not isinstance(value[name], int) or value[name] <= 0: raise ProtocolError("E_QUEUE_STATE") error = value["error_code"] if error is not None and (not isinstance(error, str) or not error.startswith("E_")): raise ProtocolError("E_QUEUE_STATE") if value["event"] in {"FAILED", "POSTPROCESS_FAILED"} and error is None: raise ProtocolError("E_QUEUE_STATE") if value["event"] not in {"FAILED", "POSTPROCESS_FAILED"} and error is not None: raise ProtocolError("E_QUEUE_STATE") if value["event"] == "MEDIA_COMPLETE": if "media" not in value: raise ProtocolError("E_QUEUE_STATE") try: validate_media_complete_identity(value["media"], job) except ProtocolError as exc: raise ProtocolError("E_QUEUE_STATE") from exc elif "media" in value: raise ProtocolError("E_QUEUE_STATE") if value["event"] == "POSTPROCESS_CLAIMED": if "recovery" not in value: raise ProtocolError("E_QUEUE_STATE") _validate_postprocess_recovery_binding(value["recovery"], job) elif "recovery" in value: raise ProtocolError("E_QUEUE_STATE") if error in INTERNAL_ONLY_PAGE_PENDING_CODES and not allow_exact_legacy_replay: raise ProtocolError("E_QUEUE_STATE") if "diagnostic" in value: if value["event"] not in {"FAILED", "POSTPROCESS_FAILED"}: raise ProtocolError("E_QUEUE_STATE") try: if value["event"] == "POSTPROCESS_FAILED": validate_postprocess_terminal(error, value["diagnostic"]) elif error in PREPARELESS_REJECT_CODES: validate_prepareless_terminal(error, value["diagnostic"]) else: validate_runtime_diagnostic(value["diagnostic"]) except ValueError as exc: raise ProtocolError("E_QUEUE_STATE") from exc elif value["event"] == "POSTPROCESS_FAILED": try: validate_postprocess_terminal(error, None) except ValueError as exc: raise ProtocolError("E_QUEUE_STATE") from exc elif error in PREPARELESS_REJECT_CODES and not allow_exact_missing_diagnostic_replay: raise ProtocolError("E_QUEUE_STATE") return value def _terminal_request( job: dict[str, Any], lease_id: str, now_ms: int, *, complete: bool, error_code: str | None, diagnostic: dict[str, object] | None, ) -> dict[str, Any]: """Validate one live terminal request before any idempotent decision.""" validated_job = validate_job(job) if not isinstance(complete, bool): raise ProtocolError("E_QUEUE_STATE") if error_code == COMPLETION_CLOSURE_REQUIRED: raise ProtocolError("E_QUEUE_STATE") payload: dict[str, Any] = { "schema": QUEUE_SCHEMA_VERSION, "event": "COMPLETE" if complete else "FAILED", "job_id": validated_job["job_id"], "bvid": validated_job["bvid"], "creator_uid": validated_job["creator_uid"], "lease_id": lease_id, "at_unix_ms": now_ms, "lease_expires_unix_ms": now_ms + QUEUE_LEASE_SECONDS * 1_000, "error_code": error_code, } if diagnostic is not None: payload["diagnostic"] = diagnostic return _event(payload, {validated_job["job_id"]: validated_job}) def _is_governed_legacy_state_path(path: Path) -> bool: if not path.is_absolute(): return False suffix = tuple(part.casefold() for part in LEGACY_PAGE_METADATA_REPLAY_STATE_SUFFIX) parts = tuple(part.casefold() for part in path.parts) return len(parts) >= len(suffix) and parts[-len(suffix):] == suffix def _replay_state_events(path: Path, jobs: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: if not path.exists(): return [] if not path.is_file() or path.is_symlink(): raise ProtocolError("E_QUEUE") payload = path.read_bytes() if len(payload) > 16 * 1024 * 1024: raise ProtocolError("E_QUEUE") if payload and not payload.endswith(b"\n"): raise ProtocolError("E_QUEUE_PARTIAL") governed = ( _is_governed_legacy_state_path(path) and len(payload) >= LEGACY_PAGE_METADATA_REPLAY_PREFIX_BYTES ) if governed and hashlib.sha256( payload[:LEGACY_PAGE_METADATA_REPLAY_PREFIX_BYTES] ).hexdigest().upper() != LEGACY_PAGE_METADATA_REPLAY_PREFIX_SHA256: raise ProtocolError("E_QUEUE_STATE") expected = { line_number: (line_bytes, line_sha256) for line_number, line_bytes, line_sha256 in LEGACY_PAGE_METADATA_REPLAY_LINES } expected_missing = { line_number: (line_bytes, line_sha256, error_code) for line_number, line_bytes, line_sha256, error_code in LEGACY_MISSING_DIAGNOSTIC_REPLAY_LINES } observed: list[tuple[int, int, str]] = [] observed_missing: list[tuple[int, int, str, str]] = [] events: list[dict[str, Any]] = [] latest_by_job: dict[str, dict[str, Any]] = {} media_complete_by_job: dict[str, dict[str, Any]] = {} for line_number, line in enumerate(payload.splitlines(keepends=True), 1): if not line.endswith(b"\n") or len(line) <= 1 or len(line) - 1 > 4096: raise ProtocolError("E_QUEUE") value = strict_json_loads(line[:-1]) legacy = value.get("error_code") in INTERNAL_ONLY_PAGE_PENDING_CODES allow = False allow_missing = False if legacy: identity = (len(line), hashlib.sha256(line).hexdigest().upper()) allow = governed and expected.get(line_number) == identity if not allow: raise ProtocolError("E_QUEUE_STATE") observed.append((line_number, *identity)) elif value.get("error_code") in PREPARELESS_REJECT_CODES and "diagnostic" not in value: identity_with_error = ( len(line), hashlib.sha256(line).hexdigest().upper(), value["error_code"], ) allow_missing = governed and expected_missing.get(line_number) == identity_with_error if not allow_missing: raise ProtocolError("E_QUEUE_STATE") observed_missing.append((line_number, *identity_with_error)) parsed = _event( value, jobs, allow_exact_legacy_replay=allow, allow_exact_missing_diagnostic_replay=allow_missing, ) job_id = parsed["job_id"] latest = latest_by_job.get(job_id) if parsed["event"] == "MEDIA_COMPLETE": if job_id in media_complete_by_job: raise ProtocolError("E_QUEUE_STATE") if ( latest is None or latest["event"] != "STARTED" or latest["lease_id"] != parsed["lease_id"] ): raise ProtocolError("E_QUEUE_STATE") media_complete_by_job[job_id] = parsed elif parsed["event"] == "POSTPROCESS_CLAIMED": prior_media = media_complete_by_job.get(job_id) if ( prior_media is None or latest is None or latest["event"] not in {"MEDIA_COMPLETE", "POSTPROCESS_CLAIMED"} or parsed["recovery"] != _postprocess_recovery_binding(prior_media) ): raise ProtocolError("E_QUEUE_STATE") elif latest is not None and latest["event"] == "MEDIA_COMPLETE": if ( parsed["event"] not in {"COMPLETE", "POSTPROCESS_FAILED"} or parsed["lease_id"] != latest["lease_id"] ): raise ProtocolError("E_QUEUE_STATE") elif latest is not None and latest["event"] == "POSTPROCESS_CLAIMED": if ( parsed["event"] not in {"COMPLETE", "POSTPROCESS_FAILED"} or parsed["lease_id"] != latest["lease_id"] ): raise ProtocolError("E_QUEUE_STATE") events.append(parsed) latest_by_job[job_id] = parsed if governed and tuple(observed) != LEGACY_PAGE_METADATA_REPLAY_LINES: raise ProtocolError("E_QUEUE_STATE") if governed and tuple(observed_missing) != LEGACY_MISSING_DIAGNOSTIC_REPLAY_LINES: raise ProtocolError("E_QUEUE_STATE") return events class QueueStore: def __init__( self, queue_path: Path, state_path: Path, lock_path: Path, allowed_creators: frozenset[str], ) -> None: self.queue_path = queue_path self.state_path = state_path self.lock_path = lock_path self.allowed_creators = allowed_creators @contextmanager def _locked(self) -> Iterator[None]: import msvcrt self.lock_path.parent.mkdir(parents=True, exist_ok=True) with self.lock_path.open("a+b") as stream: if stream.seek(0, os.SEEK_END) == 0: stream.write(b"\0") stream.flush() os.fsync(stream.fileno()) stream.seek(0) try: msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) except OSError as exc: raise ProtocolError("E_QUEUE_BUSY") from exc try: yield finally: stream.seek(0) msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1) def _jobs(self) -> list[dict[str, Any]]: payload = self.queue_path.read_bytes() if self.queue_path.exists() else b"" snapshot = _queue_snapshot(payload, self.allowed_creators) if snapshot.pending_suffix: raise ProtocolError("E_QUEUE_PARTIAL") return list(snapshot.jobs.values()) def append_ingress_jobs(self, records: list[dict[str, Any]]) -> dict[str, int]: """Append a prevalidated producer batch under the consumer's exact lock. The full incoming batch and the existing queue are checked before the first append. Replaying byte-equivalent schema-1 records is idempotent; any field drift for a stable job id fails closed. """ if not isinstance(records, list) or not 1 <= len(records) <= 10_000: raise ProtocolError("E_QUEUE") incoming: dict[str, tuple[dict[str, Any], dict[str, Any]]] = {} order: list[str] = [] for raw in records: if not isinstance(raw, dict): raise ProtocolError("E_QUEUE") job = _ingress_job(raw, self.allowed_creators) job_id = job["job_id"] previous = incoming.get(job_id) if previous is not None and previous[0] != raw: raise ProtocolError("E_QUEUE_CONFLICT") if previous is None: incoming[job_id] = (raw, job) order.append(job_id) with self._locked(): payload = self.queue_path.read_bytes() if self.queue_path.exists() else b"" snapshot = _queue_snapshot(payload, self.allowed_creators) if snapshot.pending_suffix: raise ProtocolError("E_QUEUE_PARTIAL") existing = snapshot.records to_append: list[dict[str, Any]] = [] unchanged = 0 for job_id in order: raw = incoming[job_id][0] previous = existing.get(job_id) if previous is None: to_append.append(raw) elif previous == raw: unchanged += 1 else: raise ProtocolError("E_QUEUE_CONFLICT") _append_jsonl_batch(self.queue_path, to_append) return {"appended": len(to_append), "unchanged": unchanged} @staticmethod def _validate_release(approval: ReleaseApproval) -> None: if not isinstance(approval, ReleaseApproval): raise ProtocolError("E_AUTH_TRUST") _validate_authorization_identity( approval.authorization_message_id, approval.authorization_handoff_id, approval.authorization_sha256, ) if ( not MESSAGE_ID_RE.fullmatch(approval.repair_review_result_message_id) or not isinstance(approval.repair_audit_id, str) or not approval.repair_audit_id.startswith("DEV-AUDIT-") or isinstance(approval.repair_audit_bytes, bool) or not isinstance(approval.repair_audit_bytes, int) or approval.repair_audit_bytes <= 0 or not UPPER_SHA256_RE.fullmatch(approval.repair_audit_sha256) or not isinstance(approval.successors, tuple) or not 1 <= len(approval.successors) <= 100 ): raise ProtocolError("E_AUTH_TRUST") identities: list[tuple[str, str, str]] = [] for item in approval.successors: if not isinstance(item, AuthorizedSuccessor): raise ProtocolError("E_AUTH_TRUST") try: validate_creator_uid(item.creator_uid) validate_bvid(item.bvid) except ValueError as exc: raise ProtocolError("E_AUTH_TRUST") from exc if ( not JOB_ID_RE.fullmatch(item.predecessor_job_id) or isinstance(item.retry_generation, bool) or not isinstance(item.retry_generation, int) or not 1 <= item.retry_generation <= 1_000_000 or not ERROR_CODE_RE.fullmatch(item.terminal_error_code) ): raise ProtocolError("E_AUTH_TRUST") identities.append((item.creator_uid, item.bvid, item.predecessor_job_id)) if identities != sorted(identities) or len(set(identities)) != len(identities): raise ProtocolError("E_AUTH_TRUST") @staticmethod def _build_successor_block( approval: ReleaseApproval, records: list[dict[str, Any]] ) -> tuple[bytes, tuple[str, ...]]: ordered = sorted(records, key=lambda value: value["job_id"]) begin = { "schema": SUCCESSOR_QUEUE_SCHEMA_VERSION, "record_type": "SUCCESSOR_BEGIN", "authorization_message_id": approval.authorization_message_id, "authorization_handoff_id": approval.authorization_handoff_id, "authorization_sha256": approval.authorization_sha256, "successor_count": len(ordered), } prefix = _canonical_line(begin) + b"".join(_canonical_line(value) for value in ordered) commit = { **begin, "record_type": "SUCCESSOR_COMMIT", "block_sha256": hashlib.sha256(prefix).hexdigest().upper(), } block = prefix + _canonical_line(commit) if len(block) > 1024 * 1024: raise ProtocolError("E_LINEAGE") return block, tuple(value["job_id"] for value in ordered) def append_authorized_successors( self, approval_loader: Callable[[], ReleaseApproval], catalog_records: list[dict[str, Any]], ) -> dict[str, int]: """Append or recover one deterministic authorized schema-2 block.""" catalog: dict[tuple[str, str], dict[str, Any]] = {} if not isinstance(catalog_records, list) or not catalog_records: raise ProtocolError("E_CATALOG") for raw in catalog_records: if not isinstance(raw, dict): raise ProtocolError("E_CATALOG") job = _ingress_job(raw, self.allowed_creators) key = (job["creator_uid"], job["bvid"]) previous = catalog.get(key) if previous is not None and previous != raw: raise ProtocolError("E_CATALOG_CONFLICT") catalog.setdefault(key, raw) with self._locked(): approval = approval_loader() self._validate_release(approval) payload = self.queue_path.read_bytes() if self.queue_path.exists() else b"" snapshot = _queue_snapshot(payload, self.allowed_creators) events = self._events(snapshot.jobs) latest: dict[str, dict[str, Any]] = {} for event in events: latest[event["job_id"]] = event successor_records: list[dict[str, Any]] = [] for item in approval.successors: predecessor = snapshot.jobs.get(item.predecessor_job_id) source = snapshot.records.get(item.predecessor_job_id) current_catalog = catalog.get((item.creator_uid, item.bvid)) if predecessor is None or source is None or current_catalog is None: raise ProtocolError("E_LINEAGE_TERMINAL") if predecessor["creator_uid"] != item.creator_uid or predecessor["bvid"] != item.bvid: raise ProtocolError("E_LINEAGE_CONFLICT") state = latest.get(item.predecessor_job_id) failed_terminal = ( state is not None and state["event"] in {"FAILED", "POSTPROCESS_FAILED"} and item.terminal_error_code != COMPLETION_CLOSURE_REQUIRED and state["error_code"] == item.terminal_error_code ) completion_closure = ( state is not None and state["event"] == "COMPLETE" and state["error_code"] is None and item.terminal_error_code == COMPLETION_CLOSURE_REQUIRED ) if not (failed_terminal or completion_closure): raise ProtocolError("E_LINEAGE_TERMINAL") predecessor_lineage = predecessor.get("lineage") expected_generation = 1 if predecessor_lineage is None else predecessor_lineage["retry_generation"] + 1 if item.retry_generation != expected_generation: raise ProtocolError("E_LINEAGE_GENERATION") for key in ( "bvid", "creator_uid", "expected_duration_ms", "discovered_at_unix_ms", "published_at", "title", ): if source.get(key) != current_catalog.get(key): raise ProtocolError("E_CATALOG_CONFLICT") lineage = { "predecessor_job_id": item.predecessor_job_id, "retry_generation": item.retry_generation, "predecessor_terminal_error_code": item.terminal_error_code, "authorization_message_id": approval.authorization_message_id, "authorization_handoff_id": approval.authorization_handoff_id, "authorization_sha256": approval.authorization_sha256, "repair_review_result_message_id": approval.repair_review_result_message_id, "repair_audit_id": approval.repair_audit_id, "repair_audit_bytes": approval.repair_audit_bytes, "repair_audit_sha256": approval.repair_audit_sha256, } try: job_id = stable_successor_job_id( item.creator_uid, item.bvid, item.predecessor_job_id, item.retry_generation, item.terminal_error_code, approval.authorization_message_id, approval.authorization_handoff_id, approval.authorization_sha256, approval.repair_review_result_message_id, approval.repair_audit_id, approval.repair_audit_bytes, approval.repair_audit_sha256, ) except ValueError as exc: raise ProtocolError("E_LINEAGE") from exc successor_records.append({ "schema": SUCCESSOR_QUEUE_SCHEMA_VERSION, "record_type": "SUCCESSOR_JOB", "job_id": job_id, "bvid": item.bvid, "creator_uid": item.creator_uid, "expected_duration_ms": source["expected_duration_ms"], "discovered_at_unix_ms": source["discovered_at_unix_ms"], "published_at": source["published_at"], "title": source["title"], "lineage": lineage, }) block, target_ids = self._build_successor_block(approval, successor_records) matching_blocks = [ existing for existing in snapshot.blocks if existing[:3] == ( approval.authorization_message_id, approval.authorization_handoff_id, approval.authorization_sha256, ) ] if matching_blocks: if len(matching_blocks) != 1 or matching_blocks[0][3] != target_ids or matching_blocks[0][4] != block or snapshot.pending_suffix: raise ProtocolError("E_LINEAGE_CONFLICT") return {"appended": 0, "unchanged": len(target_ids), "recovered": 0} for item, target_id in zip(approval.successors, ( value["job_id"] for value in successor_records )): existing_child = snapshot.children.get(item.predecessor_job_id) if existing_child is not None and existing_child != target_id: raise ProtocolError("E_LINEAGE_CONFLICT") suffix = snapshot.pending_suffix if suffix and not block.startswith(suffix): raise ProtocolError("E_LINEAGE_RECOVERY") remaining = block[len(suffix):] _append_bytes(self.queue_path, remaining) final_payload = self.queue_path.read_bytes() if not final_payload.startswith(payload) or final_payload != snapshot.committed_bytes + block: raise ProtocolError("E_LINEAGE_RECOVERY") final_snapshot = _queue_snapshot(final_payload, self.allowed_creators) if final_snapshot.pending_suffix or not any(existing[3] == target_ids and existing[4] == block for existing in final_snapshot.blocks): raise ProtocolError("E_LINEAGE_RECOVERY") return { "appended": len(target_ids), "unchanged": 0, "recovered": len(target_ids) if suffix else 0, } def _events(self, jobs: dict[str, dict[str, Any]] | None = None) -> list[dict[str, Any]]: if jobs is None: payload = self.queue_path.read_bytes() if self.queue_path.exists() else b"" snapshot = _queue_snapshot(payload, self.allowed_creators) if snapshot.pending_suffix: raise ProtocolError("E_QUEUE_PARTIAL") jobs = snapshot.jobs return _replay_state_events(self.state_path, jobs) def claim_next(self, now_ms: int) -> tuple[dict[str, Any], str] | None: with self._locked(): payload = self.queue_path.read_bytes() if self.queue_path.exists() else b"" snapshot = _queue_snapshot(payload, self.allowed_creators) if snapshot.pending_suffix: raise ProtocolError("E_QUEUE_PARTIAL") jobs = list(snapshot.jobs.values()) events = self._events(snapshot.jobs) latest: dict[str, dict[str, Any]] = {} for item in events: latest[item["job_id"]] = item for job in jobs: current = latest.get(job["job_id"]) if current and current["event"] in _TERMINAL_EVENTS: continue if current and current["event"] in {"MEDIA_COMPLETE", "POSTPROCESS_CLAIMED"}: if ( current["event"] == "POSTPROCESS_CLAIMED" and current["lease_expires_unix_ms"] >= now_ms ): continue prior = [ item for item in events if item["job_id"] == job["job_id"] and item["event"] == "MEDIA_COMPLETE" ] if len(prior) != 1: raise ProtocolError("E_QUEUE_STATE") lease = secrets.token_hex(16) recovery = _postprocess_recovery_binding(prior[0]) _postprocess_recovery_claim_test_seam("BEFORE_APPEND") self._append_event( job, lease, "POSTPROCESS_CLAIMED", now_ms, None, recovery=recovery, ) _postprocess_recovery_claim_test_seam("AFTER_APPEND_BEFORE_READBACK") reread = [ item for item in self._events(snapshot.jobs) if item["job_id"] == job["job_id"] ] if ( not reread or reread[-1]["event"] != "POSTPROCESS_CLAIMED" or reread[-1]["lease_id"] != lease or reread[-1].get("recovery") != recovery ): raise ProtocolError("E_QUEUE_WRITE") _postprocess_recovery_claim_test_seam("AFTER_READBACK") return job, lease if current and current["event"] == "STARTED": if current["lease_expires_unix_ms"] < now_ms: self._append_event(job, current["lease_id"], "FAILED", now_ms, "E_ORPHANED") continue if current and current["event"] == "CLAIMED": if current["lease_expires_unix_ms"] >= now_ms: continue attempts = sum( item["event"] == "CLAIMED" and item["job_id"] == job["job_id"] for item in events ) if attempts >= MAX_CLAIM_ATTEMPTS: self._append_event(job, current["lease_id"], "FAILED", now_ms, "E_CLAIM_EXPIRED") continue lease = secrets.token_hex(16) self._append_event(job, lease, "CLAIMED", now_ms, None) return job, lease return None def _append_event( self, job: dict[str, Any], lease_id: str, event: str, now_ms: int, error_code: str | None, diagnostic: dict[str, object] | None = None, media: dict[str, Any] | None = None, recovery: dict[str, Any] | None = None, ) -> None: validate_job(job) if error_code == COMPLETION_CLOSURE_REQUIRED: raise ProtocolError("E_QUEUE_STATE") if error_code in INTERNAL_ONLY_PAGE_PENDING_CODES: raise ProtocolError("E_QUEUE_STATE") if error_code in PREPARELESS_REJECT_CODES: try: diagnostic = validate_prepareless_terminal(error_code, diagnostic) except ValueError as exc: raise ProtocolError("E_QUEUE_STATE") from exc if event != "FAILED": raise ProtocolError("E_QUEUE_STATE") elif event == "POSTPROCESS_FAILED": try: diagnostic = validate_postprocess_terminal(error_code, diagnostic) except ValueError as exc: raise ProtocolError("E_QUEUE_STATE") from exc elif diagnostic is not None: try: validate_runtime_diagnostic(diagnostic) except ValueError as exc: raise ProtocolError("E_QUEUE_STATE") from exc if event not in {"FAILED", "POSTPROCESS_FAILED"}: raise ProtocolError("E_QUEUE_STATE") payload = { "schema": QUEUE_SCHEMA_VERSION, "event": event, "job_id": job["job_id"], "bvid": job["bvid"], "creator_uid": job["creator_uid"], "lease_id": lease_id, "at_unix_ms": now_ms, "lease_expires_unix_ms": now_ms + QUEUE_LEASE_SECONDS * 1_000, "error_code": error_code, } if diagnostic is not None: payload["diagnostic"] = diagnostic if event == "MEDIA_COMPLETE": try: payload["media"] = validate_media_complete_identity(media, job) except ProtocolError as exc: raise ProtocolError("E_QUEUE_STATE") from exc elif media is not None: raise ProtocolError("E_QUEUE_STATE") if event == "POSTPROCESS_CLAIMED": payload["recovery"] = _validate_postprocess_recovery_binding(recovery, job) elif recovery is not None: raise ProtocolError("E_QUEUE_STATE") _append_jsonl( self.state_path, payload, ) def assert_claim(self, job: dict[str, Any], lease_id: str, now_ms: int) -> None: validate_job(job) with self._locked(): latest = None for item in self._events(): if item["job_id"] == job["job_id"]: latest = item if ( latest is None or latest["event"] != "CLAIMED" or latest["lease_id"] != lease_id or latest["lease_expires_unix_ms"] < now_ms ): raise ProtocolError("E_LEASE") def mark_started(self, job: dict[str, Any], lease_id: str, now_ms: int) -> None: with self._locked(): latest = None for item in self._events(): if item["job_id"] == job["job_id"]: latest = item if latest is None or latest["event"] != "CLAIMED" or latest["lease_id"] != lease_id: raise ProtocolError("E_LEASE") self._append_event(job, lease_id, "STARTED", now_ms, None) def postprocess_recovery_claim( self, job: dict[str, Any], lease_id: str, ) -> dict[str, Any] | None: """Return the exact durable recovery binding, or None for an ordinary claim.""" validated_job = validate_job(job) with self._locked(): matching = [ item for item in self._events() if item["job_id"] == validated_job["job_id"] ] latest = matching[-1] if matching else None if latest is None or latest["lease_id"] != lease_id: raise ProtocolError("E_LEASE") if latest["event"] == "CLAIMED": return None if latest["event"] != "POSTPROCESS_CLAIMED": raise ProtocolError("E_LEASE") prior = [item for item in matching if item["event"] == "MEDIA_COMPLETE"] if ( len(prior) != 1 or latest.get("recovery") != _postprocess_recovery_binding(prior[0]) ): raise ProtocolError("E_QUEUE_STATE") return dict(latest["recovery"]) def mark_media_complete( self, job: dict[str, Any], lease_id: str, now_ms: int, media: dict[str, Any], ) -> dict[str, Any]: """Persist verified user media before any reentrant postprocess outcome.""" validated_job = validate_job(job) validated_media = validate_media_complete_identity(media, validated_job) with self._locked(): matching = [ item for item in self._events() if item["job_id"] == job["job_id"] ] latest = matching[-1] if matching else None prior = [item for item in matching if item["event"] == "MEDIA_COMPLETE"] if prior: if ( len(prior) != 1 or prior[0].get("media") != validated_media ): raise ProtocolError("E_LEASE") if latest is None or latest["lease_id"] != lease_id: raise ProtocolError("E_LEASE") if prior[0]["lease_id"] == lease_id: if latest["event"] not in { "MEDIA_COMPLETE", "COMPLETE", "POSTPROCESS_FAILED", }: raise ProtocolError("E_LEASE") else: recovery = [ item for item in matching if item["event"] == "POSTPROCESS_CLAIMED" and item["lease_id"] == lease_id ] if ( len(recovery) != 1 or recovery[0].get("recovery") != _postprocess_recovery_binding(prior[0]) or latest["event"] not in { "POSTPROCESS_CLAIMED", "COMPLETE", "POSTPROCESS_FAILED", } ): raise ProtocolError("E_LEASE") return prior[0] if latest is None or latest["event"] != "STARTED" or latest["lease_id"] != lease_id: raise ProtocolError("E_LEASE") _media_complete_durability_test_seam("BEFORE_APPEND") self._append_event( validated_job, lease_id, "MEDIA_COMPLETE", now_ms, None, media=validated_media, ) _media_complete_durability_test_seam("AFTER_APPEND_BEFORE_READBACK") matching = [ item for item in self._events() if item["job_id"] == validated_job["job_id"] ] if not matching or matching[-1].get("media") != validated_media: raise ProtocolError("E_QUEUE_WRITE") _media_complete_durability_test_seam("AFTER_READBACK") return matching[-1] def mark_postprocess_failed( self, job: dict[str, Any], lease_id: str, now_ms: int, *, error_code: str, diagnostic: dict[str, object] | None = None, ) -> None: """Terminalize postprocess without erasing the prior MEDIA_COMPLETE fact.""" validated_job = validate_job(job) try: diagnostic = validate_postprocess_terminal(error_code, diagnostic) except ValueError as exc: raise ProtocolError("E_QUEUE_STATE") from exc if ( not isinstance(error_code, str) or not error_code.startswith("E_") ): raise ProtocolError("E_QUEUE_STATE") requested = { "schema": QUEUE_SCHEMA_VERSION, "event": "POSTPROCESS_FAILED", "job_id": validated_job["job_id"], "bvid": validated_job["bvid"], "creator_uid": validated_job["creator_uid"], "lease_id": lease_id, "at_unix_ms": now_ms, "lease_expires_unix_ms": now_ms + QUEUE_LEASE_SECONDS * 1_000, "error_code": error_code, } if diagnostic is not None: requested["diagnostic"] = diagnostic requested = _event(requested, {validated_job["job_id"]: validated_job}) with self._locked(): matching = [ item for item in self._events() if item["job_id"] == job["job_id"] ] latest = matching[-1] if matching else None if latest and latest["event"] in _TERMINAL_EVENTS: identity_fields = ( "event", "job_id", "bvid", "creator_uid", "lease_id", "error_code", ) if ( any(latest[name] != requested[name] for name in identity_fields) or ("diagnostic" in latest) != ("diagnostic" in requested) or latest.get("diagnostic") != requested.get("diagnostic") ): raise ProtocolError("E_LEASE") return if ( latest is None or latest["event"] not in {"MEDIA_COMPLETE", "POSTPROCESS_CLAIMED"} or latest["lease_id"] != lease_id ): raise ProtocolError("E_LEASE") self._append_event( job, lease_id, "POSTPROCESS_FAILED", now_ms, error_code, requested.get("diagnostic"), ) def reject_claim( self, job: dict[str, Any], lease_id: str, now_ms: int, error_code: str, diagnostic: dict[str, object] | None = None, ) -> None: validate_job(job) try: validated_diagnostic = validate_prepareless_terminal(error_code, diagnostic) except ValueError as exc: raise ProtocolError("E_REJECT") with self._locked(): latest = None for item in self._events(): if item["job_id"] == job["job_id"]: latest = item if latest and latest["event"] == "FAILED" and latest["lease_id"] == lease_id: if latest["error_code"] != error_code or latest.get("diagnostic") != validated_diagnostic: raise ProtocolError("E_LEASE") return if latest is None or latest["event"] != "CLAIMED" or latest["lease_id"] != lease_id: raise ProtocolError("E_LEASE") self._append_event(job, lease_id, "FAILED", now_ms, error_code, validated_diagnostic) def mark_terminal( self, job: dict[str, Any], lease_id: str, now_ms: int, *, complete: bool, error_code: str | None, diagnostic: dict[str, object] | None = None, ) -> None: requested = _terminal_request( job, lease_id, now_ms, complete=complete, error_code=error_code, diagnostic=diagnostic, ) with self._locked(): latest = None for item in self._events(): if item["job_id"] == job["job_id"]: latest = item if latest and latest["event"] in _TERMINAL_EVENTS: identity_fields = ( "event", "job_id", "bvid", "creator_uid", "lease_id", "error_code", ) if ( any(latest[name] != requested[name] for name in identity_fields) or ("diagnostic" in latest) != ("diagnostic" in requested) or latest.get("diagnostic") != requested.get("diagnostic") ): raise ProtocolError("E_LEASE") return if latest is None or latest["lease_id"] != lease_id: raise ProtocolError("E_LEASE") if latest["event"] == "MEDIA_COMPLETE" and requested["event"] != "COMPLETE": raise ProtocolError("E_QUEUE_STATE") if latest["event"] == "POSTPROCESS_CLAIMED" and requested["event"] != "COMPLETE": raise ProtocolError("E_QUEUE_STATE") self._append_event( job, lease_id, requested["event"], now_ms, requested["error_code"], requested.get("diagnostic"), ) class ReloadStore: def __init__(self, path: Path, generation: str) -> None: self.path = path self.generation = generation def _records(self) -> list[dict[str, Any]]: records = _read_jsonl(self.path, missing_ok=True) for value in records: if set(value) != {"schema", "generation", "event", "token", "from_build", "to_build", "at_unix_ms"}: raise ProtocolError("E_RELOAD_STATE") if value["schema"] != 1 or value["event"] not in {"OFFERED", "BEGIN", "APPLIED"}: raise ProtocolError("E_RELOAD_STATE") if not all(isinstance(value[name], str) for name in ("generation", "token", "from_build", "to_build")): raise ProtocolError("E_RELOAD_STATE") if len(value["token"]) != 32 or any(ch not in "0123456789abcdef" for ch in value["token"]): raise ProtocolError("E_RELOAD_STATE") if isinstance(value["at_unix_ms"], bool) or not isinstance(value["at_unix_ms"], int): raise ProtocolError("E_RELOAD_STATE") return records def status(self, current_build: str, now_ms: int) -> dict[str, Any]: records = [item for item in self._records() if item["generation"] == self.generation] if current_build == EXTENSION_BUILD: if not records or records[-1]["event"] != "APPLIED": token = records[-1]["token"] if records else secrets.token_hex(16) _append_jsonl(self.path, {"schema": 1, "generation": self.generation, "event": "APPLIED", "token": token, "from_build": current_build, "to_build": EXTENSION_BUILD, "at_unix_ms": now_ms}) return {"required_extension_build": EXTENSION_BUILD, "reload_required": False, "reload_token": None, "retry_after_unix_ms": 0} begun = next((item for item in reversed(records) if item["event"] == "BEGIN"), None) if begun is not None: return {"required_extension_build": EXTENSION_BUILD, "reload_required": False, "reload_token": None, "retry_after_unix_ms": begun["at_unix_ms"] + RELOAD_BACKOFF_SECONDS * 1_000} offered = records[-1] if records and records[-1]["event"] == "OFFERED" else None if offered is None: offered = {"schema": 1, "generation": self.generation, "event": "OFFERED", "token": secrets.token_hex(16), "from_build": current_build, "to_build": EXTENSION_BUILD, "at_unix_ms": now_ms} _append_jsonl(self.path, offered) return {"required_extension_build": EXTENSION_BUILD, "reload_required": True, "reload_token": offered["token"], "retry_after_unix_ms": 0} def begin(self, current_build: str, token: str, now_ms: int) -> None: records = [item for item in self._records() if item["generation"] == self.generation] if current_build == EXTENSION_BUILD or not records: raise ProtocolError("E_RELOAD") latest = records[-1] if latest["event"] != "OFFERED" or latest["token"] != token or latest["from_build"] != current_build: raise ProtocolError("E_RELOAD") _append_jsonl(self.path, {**latest, "event": "BEGIN", "at_unix_ms": now_ms})