#!/usr/bin/env python3 """Offline durable refresh transaction for the Bilibili dynamic collector. The module deliberately has no browser or network client. A supported external Chrome controller performs one bounded refresh and writes the reviewed evidence schema. This module validates and commits only that local evidence. """ from __future__ import annotations import hashlib import hmac import json import math import os import re import uuid from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path, PurePosixPath from typing import Any, Iterable, Mapping, Sequence from urllib.parse import urlsplit import bili_dynamic_collector as core TASK_ID = "DEV-PROJECT-INFO-BILI-DYNAMIC-REFRESH-COLLECTOR-20260813-001" PENDING_SCHEMA = 3 LEGACY_PENDING_SCHEMA = 2 SLOT_SCHEMA = 2 EVIDENCE_SCHEMA = 3 OBSERVATION_CONTRACT = Path(__file__).with_name("bili_dynamic_page_observation_contract.json") EXTRACTOR_SOURCE = Path(__file__).with_name("bili_dynamic_page_extract.js") RUNTIME_CONTRACT = Path(__file__).with_name("bili_dynamic_browser_runtime_contract.json") CONTROLLER_SOURCE = Path(__file__).with_name("bili_dynamic_refresh_controller.py") FORMAL_LOCK_NAME = ".bili-dynamic-formal-manifest.lock" FORMAL_STATUS_ALLOWLIST = { "SAVED", "ARTICLE_TEXT_BLOCKED_LOGGED_IN_CHROME_BRIDGE_TIMEOUT", "CONTENT_ACCESS_PENDING_CHROME", "TRANSCRIPTION_PASS_COMPLETE", "VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT", "VIDEO_DOWNLOAD_BLOCKED_AUTH_REQUIRED", "VIDEO_DOWNLOAD_BLOCKED_AUTH_SESSION_SOURCE_REVIEW_HOLD2", "VIDEO_DOWNLOAD_BLOCKED_EXTENSION_IDENTITY_VISIBILITY", "VIDEO_DOWNLOAD_BLOCKED_EXTENSION_NOT_LOADED", "VIDEO_DOWNLOAD_BLOCKED_RUNTIME_STABILITY_GATE", "VIDEO_DOWNLOAD_BLOCKED_STANDARD_EXTENSION_LOADING_DEVELOPMENT_DISPATCHED", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_ARCHIVE_METADATA_CONTRACT_REPAIR_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_ARCHIVE_METADATA_SOURCE_REVIEW_CAPACITY", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_ARCHIVE_METADATA_SOURCE_REVIEW_RESUME_SCHEDULED", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_BUILD003_ARCHIVE_METADATA_TYPE_FALSE_REJECTION", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_BUILD004_NOT_INSTALLABLE_TREE_HASH_MISMATCH", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_BUILD004_STATIC_PASS_HASH_ONLY_REVIEW_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_BUILD005_EXACT_PASS_INSTALL_SCHEDULING_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_BUILD005_STATIC_PASS_HASH_ONLY_REVIEW_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_CANONICAL_TREE_HASH_SOURCE_PASS_BUILD005_AUTHORIZED_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_CONTROLLED_BUILD003_IN_PROGRESS", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_CONTROLLED_BUILD004_IN_PROGRESS", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_METADATA_REPAIR_SOURCE_REREVIEW_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_REPLACEMENT_BUILD_002_ARCHIVE_METADATA_MISSING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_REPLACEMENT_BUILD_REPAIR_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_TREE_HASH_CONTRACT_REPAIR_SOURCE_REVIEW_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_TYPE_CONTRACT_REPAIR_COMPLETE_SOURCE_REREVIEW_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_TYPE_CONTRACT_REPAIR_SOURCE_REVIEW_PENDING", "VIDEO_DOWNLOAD_BLOCKED_YTDLP_TYPE_CONTRACT_SOURCE_PASS_BUILD004_AUTHORIZED_PENDING", "VIDEO_DOWNLOAD_PENDING_EXTENSION", } RETRYABLE_CONTENT = { "ARTICLE_TEXT_BLOCKED_LOGGED_IN_CHROME_BRIDGE_TIMEOUT", "CONTENT_ACCESS_PENDING_CHROME", } UNPARSED_REASONS = { "IDENTITY_MISSING", "IDENTITY_CONFLICT", "PUBLISHED_AT_MISSING", "PUBLISHED_AT_INVALID", "CONTENT_TYPE_UNKNOWN", "SOURCE_URL_INVALID", "NODE_TRUNCATED", "PARSER_REJECTED", } LIMIT_CODES = { "NONE", "OBSERVATION_LIMIT", "UNIQUE_CARD_LIMIT", "CARD_PER_OBSERVATION_LIMIT", "PROOF_BYTE_LIMIT", "TIME_LIMIT", } ROOT_EVIDENCE_KEYS = { "schema_version", "run_id", "transport", "requested_url", "final_url", "refresh_action", "refresh_count", "refresh_started_at", "refresh_finished_at", "read_finished_at", "page_outcome", "page_title", "creator", "extractor", "page_observation", "items", "discovery_summary", "safe_diagnostics", "runtime_contract", "runtime_observation", "controller_attestation", } @dataclass(frozen=True) class FileIdentity: exists: bool bytes: int sha256: str def as_dict(self) -> dict[str, Any]: return {"exists": self.exists, "bytes": self.bytes, "sha256": self.sha256} def _identity(path: Path) -> FileIdentity: if not path.exists(): return FileIdentity(False, 0, hashlib.sha256(b"").hexdigest()) core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file(): raise core.CollectorError("E_PATH", "Expected a regular file.", safety=True) payload = path.read_bytes() return FileIdentity(True, len(payload), hashlib.sha256(payload).hexdigest()) def _runtime_contract_identity(config: core.CollectorConfig) -> dict[str, Any]: refresh = config.refresh assert refresh is not None payload = RUNTIME_CONTRACT.read_bytes() value = _strict_json_bytes(payload, "runtime contract") expected = { "contract_id": "bili-supported-chrome-visible-runtime-v2", "overall_deadline_seconds": refresh.overall_deadline_seconds, "refresh_action_timeout_seconds": refresh.refresh_action_timeout_seconds, "observation_timeout_seconds": refresh.observation_timeout_seconds, "page_internal_settle_timeout_seconds": refresh.page_internal_settle_timeout_seconds, "max_refresh_count": refresh.max_refresh_count, "max_observation_count": 1, "controller_id": "bili-supported-chrome-controller-v1", "binding_algorithm": "hmac-sha256-controller-envelope-v1", } for field, wanted in expected.items(): if value.get(field) != wanted: raise core.CollectorError("E_CONFIG", "Runtime contract/config identity drifted.", safety=True) controller_payload = CONTROLLER_SOURCE.read_bytes() return { "contract_id": expected["contract_id"], "contract_bytes": len(payload), "contract_sha256": hashlib.sha256(payload).hexdigest(), **{key: expected[key] for key in expected if key != "contract_id"}, "controller_bytes": len(controller_payload), "controller_sha256": hashlib.sha256(controller_payload).hexdigest(), } def _controller_attestation_payload(evidence: Mapping[str, Any]) -> bytes: signable = dict(evidence) attestation = dict(signable["controller_attestation"]) attestation["binding_sha256"] = None signable["controller_attestation"] = attestation return core.canonical_json_bytes(signable, newline=False) def _strict_json_bytes(payload: bytes, description: str) -> Any: if payload.startswith(b"\xef\xbb\xbf"): raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{description} must not contain a BOM.") try: text = payload.decode("utf-8") except UnicodeDecodeError as exc: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{description} must be strict UTF-8.") from exc def pairs(values: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in values: if key in result: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{description} contains a duplicate key.") result[key] = value return result try: return json.loads(text, object_pairs_hook=pairs) except json.JSONDecodeError as exc: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{description} is not valid JSON.") from exc def _load_json_file(path: Path, description: str, *, max_bytes: int = 524288) -> tuple[dict[str, Any], bytes]: core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file(): raise core.CollectorError("E_EVIDENCE_PATH", f"{description} is not a regular file.", safety=True) if path.stat().st_size > max_bytes: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{description} exceeds the byte limit.") payload = path.read_bytes() value = _strict_json_bytes(payload, description) if not isinstance(value, dict): raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{description} root must be an object.") core.reject_secret_keys(value) return value, payload def _create_new(path: Path, payload: bytes) -> None: core.ensure_directory(path.parent, create=True) core.lexical_lstat_chain(path, allow_missing_leaf=True) try: with path.open("xb") as stream: stream.write(payload) stream.flush() os.fsync(stream.fileno()) except FileExistsError as exc: raise core.CollectorError("E_ALREADY_EXISTS", "Owned output already exists.", safety=True) from exc def _replace_manifest(path: Path, payload: bytes) -> None: """Atomic replacement that also preserves an originally absent preimage.""" if payload: core.atomic_replace_bytes(path, payload) elif path.exists(): core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file(): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Manifest rollback target is not regular.", safety=True) path.unlink() def _pending_path(config: core.CollectorConfig) -> Path: return config.state_dir / "refresh" / "pending.json" def _runs_dir(config: core.CollectorConfig) -> Path: return config.state_dir / "refresh" / "runs" def _canonical_config_hash(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _slot_path(config: core.CollectorConfig, hour_epoch: int) -> Path: assert config.refresh is not None return _runs_dir(config) / f"slot-{hour_epoch % config.refresh.run_history_slots:03d}.json" def _validate_refresh_roots(config: core.CollectorConfig, *, create_state: bool) -> None: refresh = config.refresh assert refresh is not None for path, create in ( (config.state_dir, create_state), (refresh.archive_dir, False), (refresh.intake_dir, False), ): core.ensure_directory(path, create=create) core.lexical_lstat_chain(path, allow_missing_leaf=False) if refresh.formal_manifest.parent != refresh.archive_dir: raise core.CollectorError("E_CONFIG", "formal_manifest must be directly inside archive_dir.", safety=True) core.lexical_lstat_chain(refresh.formal_manifest, allow_missing_leaf=True) def _load_pending(config: core.CollectorConfig) -> dict[str, Any] | None: path = _pending_path(config) if not path.exists(): return None value, _ = _load_json_file(path, "refresh pending") required = { "schema_version", "run_id", "owner_nonce", "phase", "task_id", "creator_uid", "creator_dynamic_url", "started_at", "deadline_at", "window_start", "window_end", "config_sha256", "state_manifest_preimage", "formal_manifest_preimage", "evidence_path", "intake_root", "evidence_identity", "planned_terminal", "transaction_identity", "last_transition_at", "runtime_contract", "controller_key_commitment", "controller_binding_sha256", } schema = value.get("schema_version") if schema == LEGACY_PENDING_SCHEMA: legacy_required = required - {"runtime_contract", "controller_key_commitment", "controller_binding_sha256"} if set(value) != legacy_required: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Legacy refresh pending schema is invalid.", safety=True) elif schema != PENDING_SCHEMA or set(value) != required: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Refresh pending schema is invalid.", safety=True) if schema == PENDING_SCHEMA and value.get("runtime_contract") != _runtime_contract_identity(config): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Refresh pending runtime contract drifted.", safety=True) if value.get("task_id") != TASK_ID or value.get("creator_uid") != config.creator_uid: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Refresh pending identity is invalid.", safety=True) if value.get("phase") not in { "AWAITING_EVIDENCE", "EVIDENCE_BOUND", "TRANSACTION_INTENT", "BUSINESS_COMMITTED", "TERMINAL_RECORDED" }: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Refresh pending phase is invalid.", safety=True) return value def _write_pending(config: core.CollectorConfig, pending: Mapping[str, Any], *, create: bool = False) -> None: payload = core.canonical_json_bytes(pending, newline=False) if create: _create_new(_pending_path(config), payload) else: core.atomic_replace_bytes(_pending_path(config), payload) def _started_slot(config: core.CollectorConfig, pending: Mapping[str, Any]) -> dict[str, Any]: started = core.parse_datetime(pending["started_at"], "pending.started_at") hour_epoch = math.floor(started.timestamp() / 3600) return { "schema_version": SLOT_SCHEMA, "slot_index": hour_epoch % 168, "hour_epoch": hour_epoch, "run_id": pending["run_id"], "run_state": "RUN_STARTED", "task_id": TASK_ID, "creator_uid": pending["creator_uid"], "started_at": pending["started_at"], "deadline_at": pending["deadline_at"], "terminal_at": None, "status": None, "error_code": None, "exit_code": None, "refresh_action": None, "refresh_count": 0, "page_authoritative": False, "coverage_complete": False, "coverage_proof": None, "evidence_sha256": None, "input_item_count": 0, "new_item_count": 0, "saved_artifact_count": 0, "state_manifest": None, "formal_manifest": None, "artifact_tree_sha256": None, "transaction_receipt": None, "warnings": [], } def _ensure_started_slot(config: core.CollectorConfig, pending: Mapping[str, Any]) -> bool: """Create a missing STARTED slot for the same durable pending run only. Returns True only when this call repaired the pending->STARTED crash window. Existing third content is never replaced. """ started = _started_slot(config, pending) payload = core.canonical_json_bytes(started, newline=False) started_at = core.parse_datetime(pending["started_at"], "pending.started_at") path = _slot_path(config, math.floor(started_at.timestamp() / 3600)) if path.exists(): core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file() or path.read_bytes() != payload: raise core.CollectorError( "E_RECOVERY_AMBIGUOUS", "The pending run slot contains non-matching content.", safety=True, ) return False _create_new(path, payload) core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file() or path.read_bytes() != payload: raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "STARTED slot durable readback mismatch.", safety=True) return True def _begin_result(config: core.CollectorConfig, pending: Mapping[str, Any]) -> dict[str, Any]: started = core.parse_datetime(pending["started_at"], "pending.started_at") result = { "status": "BROWSER_REFRESH_REQUIRED", "error_code": None, "run_id": pending["run_id"], "creator_uid": config.creator_uid, "refresh_count": 0, "page_authoritative": False, "coverage_complete": False, "new_items": 0, "saved_artifacts": 0, "formal_manifest_changed": False, "run_evidence_path": str(_slot_path(config, math.floor(started.timestamp() / 3600))), "evidence_path": pending["evidence_path"], "intake_root": pending["intake_root"], "deadline_at": pending["deadline_at"], } if pending["schema_version"] == PENDING_SCHEMA: runtime = pending["runtime_contract"] result.update({ "runtime_contract_id": runtime["contract_id"], "runtime_contract_sha256": runtime["contract_sha256"], "overall_deadline_seconds": runtime["overall_deadline_seconds"], "refresh_action_timeout_seconds": runtime["refresh_action_timeout_seconds"], "observation_timeout_seconds": runtime["observation_timeout_seconds"], "page_internal_settle_timeout_seconds": runtime["page_internal_settle_timeout_seconds"], }) return result def _validate_slot_available(config: core.CollectorConfig, hour_epoch: int) -> None: path = _slot_path(config, hour_epoch) if not path.exists(): return value, _ = _load_json_file(path, "run slot") old_hour = value.get("hour_epoch") if not isinstance(old_hour, int) or old_hour > hour_epoch: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Run slot clock is invalid.", safety=True) if old_hour == hour_epoch: raise core.CollectorError("E_RUN_HOUR_OCCUPIED", "This UTC hour already has an accepted run.", safety=True) if old_hour > hour_epoch - 168 or value.get("run_state") != "TERMINAL": raise core.CollectorError("E_RUN_HOUR_OCCUPIED", "Run slot is not safely reusable.", safety=True) def refresh_begin( config: core.CollectorConfig, config_path: Path, now: datetime, *, _controller_key_commitment: str | None = None, ) -> dict[str, Any]: refresh = config.refresh assert refresh is not None _validate_refresh_roots(config, create_state=True) recovered = _recover_or_replay(config, config_path, now) if recovered is not None: return recovered if ( not isinstance(_controller_key_commitment, str) or core.LOWER_SHA256_PATTERN.fullmatch(_controller_key_commitment) is None ): raise core.CollectorError( "E_CONTROLLER_ENTRY_REQUIRED", "New runtime-v2 runs must be created by the source-controlled refresh-run entry.", safety=True, ) hour_epoch = math.floor(now.timestamp() / 3600) _validate_slot_available(config, hour_epoch) run_id = hashlib.sha256( f"{TASK_ID}\n{config.creator_uid}\n{core.canonical_datetime(now)}\n{uuid.uuid4().hex}".encode("utf-8") ).hexdigest()[:32] evidence = config.state_dir / "refresh" / "incoming" / f"{run_id}.json" intake = refresh.intake_dir / run_id runtime_identity = _runtime_contract_identity(config) deadline = now + timedelta(seconds=refresh.overall_deadline_seconds) pending = { "schema_version": PENDING_SCHEMA, "run_id": run_id, "owner_nonce": uuid.uuid4().hex, "phase": "AWAITING_EVIDENCE", "task_id": TASK_ID, "creator_uid": config.creator_uid, "creator_dynamic_url": config.creator_dynamic_url, "started_at": core.canonical_datetime(now), "deadline_at": core.canonical_datetime(deadline), "window_start": core.canonical_datetime(now - timedelta(hours=config.window_hours)), "window_end": core.canonical_datetime(now), "config_sha256": _canonical_config_hash(config_path), "state_manifest_preimage": _identity(config.manifest_path).as_dict(), "formal_manifest_preimage": _identity(refresh.formal_manifest).as_dict(), "evidence_path": str(evidence), "intake_root": str(intake), "evidence_identity": None, "planned_terminal": None, "transaction_identity": None, "last_transition_at": core.canonical_datetime(now), "runtime_contract": runtime_identity, "controller_key_commitment": _controller_key_commitment, "controller_binding_sha256": None, } _write_pending(config, pending, create=True) _ensure_started_slot(config, pending) if _load_pending(config) != pending: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Pending readback mismatch.", safety=True) return _begin_result(config, pending) def _expected_source_hashes() -> tuple[dict[str, Any], str, str]: contract, raw = _load_json_file(OBSERVATION_CONTRACT, "page observation contract") return contract, hashlib.sha256(raw).hexdigest(), core.sha256_file(EXTRACTOR_SOURCE) def _exact_keys(value: Any, expected: set[str], field: str) -> Mapping[str, Any]: if not isinstance(value, Mapping) or set(value) != expected: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{field} has unexpected or missing keys.") return value def _safe_time(value: Any, field: str, pending: Mapping[str, Any]) -> datetime: parsed = core.parse_datetime(value, field).astimezone(timezone.utc) started = core.parse_datetime(pending["started_at"], "pending.started_at").astimezone(timezone.utc) deadline = core.parse_datetime(pending["deadline_at"], "pending.deadline_at").astimezone(timezone.utc) if parsed < started or parsed > deadline: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{field} is outside the run wall-clock bounds.") return parsed def _validate_observations(value: Any, pending: Mapping[str, Any], contract: Mapping[str, Any]) -> dict[str, Any]: root = _exact_keys(value, {"schema_version", "limits", "observations", "terminal_marker"}, "page_observation") if root["schema_version"] != 1 or root["limits"] != contract["limits"]: raise core.CollectorError("E_EVIDENCE_SCHEMA", "page_observation contract limits drifted.") observations = root["observations"] if not isinstance(observations, list) or not 1 <= len(observations) <= 64: raise core.CollectorError("E_EVIDENCE_SCHEMA", "page_observation.observations must contain 1..64 rows.") canonical_size = len(core.canonical_json_bytes(root, newline=False)) if canonical_size >= 524288: raise core.CollectorError("E_EVIDENCE_SCHEMA", "page observation proof reached its byte bound.") previous_cursor = 0 previous_time: datetime | None = None component_first_seen: list[str] = [] unique_components: set[str] = set() card_bindings: dict[str, dict[str, Any]] = {} card_token_owners: dict[str, str] = {} unparsed_order: list[str] = [] unique_unparsed: set[str] = set() observation_hashes: list[str] = [] total_visible = total_complete = total_unparsed = 0 sequence_complete = True any_limit = False last_cards: list[Mapping[str, Any]] = [] for ordinal, raw in enumerate(observations): row = _exact_keys( raw, { "ordinal", "observed_at", "cursor_before", "cursor_after", "visible_node_count", "complete_card_count", "unparsed_node_count", "cards", "unparsed_nodes", "limit_hit", }, f"observations[{ordinal}]", ) if row["ordinal"] != ordinal or row["cursor_before"] != previous_cursor: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation ordinals/cursors are discontinuous.") for field in ("cursor_before", "cursor_after", "visible_node_count", "complete_card_count", "unparsed_node_count"): if not isinstance(row[field], int) or isinstance(row[field], bool) or row[field] < 0: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"Observation {field} is invalid.") if row["cursor_after"] < row["cursor_before"]: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation cursor regressed.") observed_at = _safe_time(row["observed_at"], f"observations[{ordinal}].observed_at", pending) if previous_time is not None and observed_at < previous_time: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation time regressed.") previous_time = observed_at cards = row["cards"] unparsed = row["unparsed_nodes"] if not isinstance(cards, list) or not isinstance(unparsed, list): raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation cards/unparsed_nodes must be lists.") if ( row["complete_card_count"] != len(cards) or row["unparsed_node_count"] != len(unparsed) or row["visible_node_count"] != len(cards) + len(unparsed) or row["visible_node_count"] > 50 ): raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation node counts are not mutually exclusive.") positions: set[int] = set() before_unique = len(unique_components) normalized_cards: list[Mapping[str, Any]] = [] for index, raw_card in enumerate(cards): card = _exact_keys( raw_card, {"position", "identifiers", "stable_keys", "published_at", "content_type", "source_url"}, f"observations[{ordinal}].cards[{index}]", ) if not isinstance(card["position"], int) or card["position"] in positions: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation positions overlap.") positions.add(card["position"]) ids = _exact_keys(card["identifiers"], {"dynamic_id", "opus_id", "bvid"}, "card.identifiers") normalized = core.normalize_item( { **ids, "content_type": card["content_type"], "published_at": card["published_at"], "title": "observation-card", "source_url": card["source_url"], }, _OBSERVATION_CONFIG, index, ) if card["stable_keys"] != normalized["dedupe_keys"]: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Card stable_keys do not match CLI derivation.") component = "\n".join(normalized["dedupe_keys"]) binding = { "dedupe_keys": normalized["dedupe_keys"], "dynamic_id": normalized["dynamic_id"], "opus_id": normalized["opus_id"], "bvid": normalized["bvid"], "content_type": normalized["content_type"], "published_at": normalized["published_at"], "source_url": normalized["source_url"], } previous_binding = card_bindings.get(component) if previous_binding is not None and previous_binding != binding: raise core.CollectorError( "E_EVIDENCE_ITEM_BINDING", "Repeated observation card identity/content/time/source binding drifted.", safety=True, ) card_bindings[component] = binding for token in normalized["dedupe_keys"]: token_owner = card_token_owners.setdefault(token, component) if token_owner != component: raise core.CollectorError( "E_EVIDENCE_ITEM_BINDING", "Observed cards contain overlapping but non-identical stable components.", safety=True, ) if component not in unique_components: unique_components.add(component) component_first_seen.append(hashlib.sha256(component.encode()).hexdigest()) normalized_cards.append(card) for index, raw_node in enumerate(unparsed): node = _exact_keys(raw_node, {"position", "node_fingerprint_sha256", "reason_code"}, "unparsed_node") if not isinstance(node["position"], int) or node["position"] in positions: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation positions overlap.") positions.add(node["position"]) fingerprint = node["node_fingerprint_sha256"] if not isinstance(fingerprint, str) or core.LOWER_SHA256_PATTERN.fullmatch(fingerprint) is None: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Unparsed node fingerprint is invalid.") if node["reason_code"] not in UNPARSED_REASONS or fingerprint in unique_unparsed: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Unparsed node reason/fingerprint is invalid.") unique_unparsed.add(fingerprint) unparsed_order.append(fingerprint) if positions != set(range(row["visible_node_count"])): raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation positions must form an exact interval.") if row["limit_hit"] not in LIMIT_CODES: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation limit_hit is invalid.") if row["limit_hit"] != "NONE" or row["visible_node_count"] == 50: any_limit = True if ordinal < len(observations) - 1 and len(unique_components) == before_unique and root["terminal_marker"] is None: sequence_complete = False previous_cursor = row["cursor_after"] total_visible += row["visible_node_count"] total_complete += len(cards) total_unparsed += len(unparsed) observation_hashes.append(hashlib.sha256(core.canonical_json_bytes(row, newline=False)).hexdigest()) last_cards = normalized_cards if len(unique_components) >= 200 or len(observations) == 64: any_limit = True marker = root["terminal_marker"] window_terminated = False if marker is not None: marker = _exact_keys(marker, {"observation_ordinal", "kind", "selector_id", "normalized_text", "marker_sha256"}, "terminal_marker") if marker["observation_ordinal"] != len(observations) - 1: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Terminal marker must bind the final observation.") if marker["kind"] == "EXACT_END_OF_FEED": allowed = {(row["selector_id"], row["normalized_text"]) for row in contract["end_markers"]} if (marker["selector_id"], marker["normalized_text"]) not in allowed: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Terminal end marker is not registered.") marker_input = f"{marker['selector_id']}\n{marker['normalized_text']}".encode("utf-8") window_terminated = True elif marker["kind"] == "WINDOW_START_CARD": if marker["selector_id"] != "CARD_PUBLISHED_AT": raise core.CollectorError("E_EVIDENCE_SCHEMA", "Window marker selector is invalid.") matches = [card for card in last_cards if card["stable_keys"] == marker["normalized_text"].split("\n")] if len(matches) != 1 or core.parse_datetime(matches[0]["published_at"], "marker published_at") > core.parse_datetime(pending["window_start"], "window_start"): raise core.CollectorError("E_EVIDENCE_SCHEMA", "Window marker has no exact old card.") marker_input = core.canonical_json_bytes(matches[0], newline=False) window_terminated = True else: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Terminal marker kind is invalid.") if marker["marker_sha256"] != hashlib.sha256(marker_input).hexdigest(): raise core.CollectorError("E_EVIDENCE_SCHEMA", "Terminal marker hash mismatch.") parse_complete = total_unparsed == 0 and len(unique_unparsed) == 0 coverage_complete = sequence_complete and parse_complete and not any_limit and window_terminated return { "coverage_complete": coverage_complete, # Internal validation material. Only ``proof`` is persisted in the # terminal evidence, but the complete normalized card bindings are # retained until items have been checked one-for-one. "card_bindings": card_bindings, "proof": { "derivation_version": 1, "counts": { "observation_count": len(observations), "total_visible_node_count": total_visible, "total_complete_card_count": total_complete, "total_unparsed_node_count": total_unparsed, "unique_card_count": len(unique_components), "unique_unparsed_node_count": len(unique_unparsed), }, "ordered_observation_hashes": observation_hashes, "ordered_complete_component_hashes": component_first_seen, "ordered_unparsed_fingerprint_hashes": unparsed_order, "terminal_marker": marker, "derived": { "sequence_complete": sequence_complete, "parse_complete": parse_complete, "not_truncated": not any_limit, "window_terminated": window_terminated, "coverage_complete": coverage_complete, }, }, } # Observation normalization uses the caller config, installed just around validation. _OBSERVATION_CONFIG: core.CollectorConfig def _validate_evidence( config: core.CollectorConfig, pending: Mapping[str, Any], path: Path, *, _controller_key: bytes | None = None, ) -> tuple[dict[str, Any], bytes, dict[str, Any] | None, bool]: global _OBSERVATION_CONFIG if str(path) != pending["evidence_path"]: raise core.CollectorError("E_EVIDENCE_PATH", "Evidence path does not match refresh-begin.", safety=True) value, payload = _load_json_file(path, "browser evidence") _exact_keys(value, ROOT_EVIDENCE_KEYS, "evidence") if value["schema_version"] != EVIDENCE_SCHEMA or value["run_id"] != pending["run_id"]: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Evidence run/schema identity mismatch.") if value["transport"] != "codex_chrome_visible_page" or value["requested_url"] != config.creator_dynamic_url: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Evidence transport/request identity mismatch.") if pending.get("schema_version") != PENDING_SCHEMA: raise core.CollectorError("E_EVIDENCE_SCHEMA", "runtime-v2 evidence requires pending schema 3.") attestation = _exact_keys( value["controller_attestation"], { "controller_id", "controller_sha256", "binding_algorithm", "action_dispatched", "monotonic_run_started_ms", "monotonic_action_started_ms", "monotonic_action_finished_ms", "monotonic_observation_started_ms", "monotonic_observation_finished_ms", "monotonic_evidence_write_started_ms", "binding_sha256", }, "controller_attestation", ) pending_runtime = pending["runtime_contract"] if ( attestation["controller_id"] != pending_runtime["controller_id"] or attestation["controller_sha256"] != pending_runtime["controller_sha256"] or attestation["binding_algorithm"] != pending_runtime["binding_algorithm"] ): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller source identity drifted.", safety=True) binding = attestation["binding_sha256"] if not isinstance(binding, str) or core.LOWER_SHA256_PATTERN.fullmatch(binding) is None: raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller binding is invalid.", safety=True) if _controller_key is not None: commitment = hashlib.sha256(_controller_key).hexdigest() if not hmac.compare_digest(commitment, pending["controller_key_commitment"]): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller authority commitment mismatch.", safety=True) expected_binding = hmac.new( _controller_key, _controller_attestation_payload(value), hashlib.sha256 ).hexdigest() if not hmac.compare_digest(binding, expected_binding): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller envelope binding mismatch.", safety=True) elif ( pending.get("phase") != "EVIDENCE_BOUND" or pending.get("controller_binding_sha256") != binding ): raise core.CollectorError( "E_CONTROLLER_REQUIRED", "Caller-authored runtime-v2 evidence cannot enter the commit path.", safety=True, ) runtime_contract = _exact_keys( value["runtime_contract"], {"contract_id", "contract_bytes", "contract_sha256"}, "runtime_contract" ) if runtime_contract != { "contract_id": pending_runtime["contract_id"], "contract_bytes": pending_runtime["contract_bytes"], "contract_sha256": pending_runtime["contract_sha256"], }: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Runtime contract identity drifted.") runtime = _exact_keys( value["runtime_observation"], { "refresh_action_outcome", "refresh_action_elapsed_ms", "refresh_count", "observation_outcome", "observation_elapsed_ms", "observation_count", }, "runtime_observation", ) action_outcome = runtime["refresh_action_outcome"] observation_outcome = runtime["observation_outcome"] if action_outcome not in {"CONFIRMED", "TIMEOUT", "PRE_DISPATCH_ERROR", "POST_DISPATCH_ERROR"}: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Runtime action outcome is invalid.") if observation_outcome not in {"READABLE", "TIMEOUT", "ERROR", "ACCESS_BLOCKED", "NOT_ATTEMPTED", "DEADLINE_EXHAUSTED"}: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Runtime observation outcome is invalid.") for field, upper in ( ("refresh_action_elapsed_ms", config.refresh.refresh_action_timeout_seconds * 1000), ("observation_elapsed_ms", config.refresh.observation_timeout_seconds * 1000), ): raw = runtime[field] if not isinstance(raw, int) or isinstance(raw, bool) or not 0 <= raw <= upper: raise core.CollectorError("E_EVIDENCE_SCHEMA", f"{field} is outside the frozen budget.") if runtime["refresh_count"] != value["refresh_count"]: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Runtime/evidence refresh count differs.") monotonic_fields = ( "monotonic_run_started_ms", "monotonic_evidence_write_started_ms", ) if any(not isinstance(attestation[field], int) or isinstance(attestation[field], bool) or attestation[field] < 0 for field in monotonic_fields): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Controller monotonic envelope is invalid.", safety=True) run_ms = attestation["monotonic_run_started_ms"] write_ms = attestation["monotonic_evidence_write_started_ms"] if write_ms < run_ms or write_ms - run_ms > config.refresh.overall_deadline_seconds * 1000: raise core.CollectorError("E_OVERALL_DEADLINE", "Controller evidence write exceeded the total deadline.", safety=True) action_times = (attestation["monotonic_action_started_ms"], attestation["monotonic_action_finished_ms"]) observation_times = (attestation["monotonic_observation_started_ms"], attestation["monotonic_observation_finished_ms"]) if action_outcome == "PRE_DISPATCH_ERROR": if attestation["action_dispatched"] or action_times != (None, None) or observation_times != (None, None): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Pre-dispatch envelope is inconsistent.", safety=True) else: if not attestation["action_dispatched"] or any(not isinstance(item, int) or isinstance(item, bool) for item in action_times): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Dispatched action envelope is invalid.", safety=True) action_start, action_finish = action_times if not run_ms <= action_start <= action_finish <= write_ms or action_finish - action_start != runtime["refresh_action_elapsed_ms"]: raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Action monotonic envelope differs from runtime evidence.", safety=True) if runtime["observation_count"] == 0: if observation_times != (None, None): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Unattempted observation has timestamps.", safety=True) else: if any(not isinstance(item, int) or isinstance(item, bool) for item in observation_times): raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Observation monotonic envelope is invalid.", safety=True) observation_start, observation_finish = observation_times if not action_finish <= observation_start <= observation_finish <= write_ms or observation_finish - observation_start != runtime["observation_elapsed_ms"]: raise core.CollectorError("E_CONTROLLER_ATTESTATION", "Observation monotonic envelope differs from runtime evidence.", safety=True) if action_outcome == "PRE_DISPATCH_ERROR": if ( value["refresh_action"] is not None or runtime["refresh_count"] != 0 or runtime["observation_count"] != 0 or observation_outcome != "NOT_ATTEMPTED" or runtime["refresh_action_elapsed_ms"] != 0 or runtime["observation_elapsed_ms"] != 0 ): raise core.CollectorError("E_EVIDENCE_SCHEMA", "Pre-dispatch error matrix is invalid.") else: if value["refresh_action"] not in {"navigate", "reload"} or runtime["refresh_count"] != 1: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Exactly one dispatched browser refresh is required.") if observation_outcome == "DEADLINE_EXHAUSTED": if runtime["observation_count"] != 0 or runtime["observation_elapsed_ms"] != 0: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Deadline-exhausted observation matrix is invalid.") elif runtime["observation_count"] != 1: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Exactly one browser observation is required.") started = _safe_time(value["refresh_started_at"], "refresh_started_at", pending) finished = _safe_time(value["refresh_finished_at"], "refresh_finished_at", pending) read_finished = _safe_time(value["read_finished_at"], "read_finished_at", pending) if not started <= finished <= read_finished: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Evidence times are not monotonic.") action_wall_ms = int((finished - started).total_seconds() * 1000) observation_wall_ms = int((read_finished - finished).total_seconds() * 1000) if action_wall_ms != runtime["refresh_action_elapsed_ms"] or observation_wall_ms != runtime["observation_elapsed_ms"]: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Runtime elapsed values do not match evidence walls.") if value["page_outcome"] not in {"READABLE", "UNREADABLE_TIMEOUT", "UNREADABLE_ERROR", "ACCESS_BLOCKED"}: raise core.CollectorError("E_EVIDENCE_SCHEMA", "page_outcome is invalid.") if not isinstance(value["page_title"], str) or core.CONTROL_CHARACTER_PATTERN.search(value["page_title"]): raise core.CollectorError("E_EVIDENCE_SCHEMA", "page_title is invalid.") creator = _exact_keys(value["creator"], {"uid", "name", "profile_url"}, "creator") identity_match = ( str(creator["uid"]) == config.creator_uid and creator["name"] == config.creator_name and creator["profile_url"] == f"https://space.bilibili.com/{config.creator_uid}" and value["final_url"] == config.creator_dynamic_url ) discovery = _exact_keys(value["discovery_summary"], {"status", "item_count"}, "discovery_summary") if discovery["status"] not in {"NOT_USED", "EMPTY", "BLOCKED_412", "PARTIAL"} or not isinstance(discovery["item_count"], int): raise core.CollectorError("E_EVIDENCE_SCHEMA", "discovery_summary is invalid.") diagnostics = _exact_keys(value["safe_diagnostics"], {"code", "overall_deadline_seconds", "refresh_action_timeout_seconds", "observation_timeout_seconds"}, "safe_diagnostics") if diagnostics["code"] not in {"NONE", "ACTION_TIMEOUT", "ACTION_PRE_DISPATCH", "ACTION_POST_DISPATCH", "OBSERVATION_TIMEOUT", "OBSERVATION_ERROR", "ACCESS_INTERSTITIAL", "DEADLINE_EXHAUSTED"}: raise core.CollectorError("E_EVIDENCE_SCHEMA", "safe_diagnostics.code is invalid.") if diagnostics != { "code": diagnostics["code"], "overall_deadline_seconds": config.refresh.overall_deadline_seconds, "refresh_action_timeout_seconds": config.refresh.refresh_action_timeout_seconds, "observation_timeout_seconds": config.refresh.observation_timeout_seconds, }: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Evidence timeout contract drifted.") expected_code = { "PRE_DISPATCH_ERROR": "ACTION_PRE_DISPATCH", "POST_DISPATCH_ERROR": "ACTION_POST_DISPATCH", "TIMEOUT": "ACTION_TIMEOUT", }.get(action_outcome) if expected_code is None: expected_code = { "READABLE": "NONE", "TIMEOUT": "OBSERVATION_TIMEOUT", "ERROR": "OBSERVATION_ERROR", "ACCESS_BLOCKED": "ACCESS_INTERSTITIAL", "DEADLINE_EXHAUSTED": "DEADLINE_EXHAUSTED", }.get(observation_outcome) if diagnostics["code"] != expected_code: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Safe diagnostics do not match the runtime state.") contract, contract_hash, parser_hash = _expected_source_hashes() extractor = _exact_keys(value["extractor"], {"contract_id", "contract_sha256", "parser_version", "parser_sha256"}, "extractor") if extractor != { "contract_id": contract["contract_id"], "contract_sha256": contract_hash, "parser_version": contract["parser_version"], "parser_sha256": parser_hash, }: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Evidence extractor identity drifted.") items = value["items"] if not isinstance(items, list) or len(items) > config.refresh.max_items: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Evidence items exceed the limit.") observation_result = None expected_page = { "READABLE": "READABLE", "TIMEOUT": "UNREADABLE_TIMEOUT", "ERROR": "UNREADABLE_ERROR", "ACCESS_BLOCKED": "ACCESS_BLOCKED", "NOT_ATTEMPTED": "UNREADABLE_ERROR", "DEADLINE_EXHAUSTED": "UNREADABLE_TIMEOUT", }[observation_outcome] if value["page_outcome"] != expected_page: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Observation/page outcome projection differs.") if value["page_outcome"] == "READABLE": _OBSERVATION_CONFIG = config observation_result = _validate_observations(value["page_observation"], pending, contract) item_bindings: dict[str, dict[str, Any]] = {} item_token_owners: dict[str, str] = {} for index, raw_item in enumerate(items): item = _normalize_evidence_item(raw_item, config, index) component = "\n".join(item["dedupe_keys"]) binding = { "dedupe_keys": item["dedupe_keys"], "dynamic_id": item["dynamic_id"], "opus_id": item["opus_id"], "bvid": item["bvid"], "content_type": item["content_type"], "published_at": item["published_at"], "source_url": item["source_url"], } if component in item_bindings: raise core.CollectorError( "E_EVIDENCE_ITEM_BINDING", "Evidence items contain a duplicate stable component.", safety=True, ) item_bindings[component] = binding for token in item["dedupe_keys"]: token_owner = item_token_owners.setdefault(token, component) if token_owner != component: raise core.CollectorError( "E_EVIDENCE_ITEM_BINDING", "Evidence items contain overlapping but non-identical stable components.", safety=True, ) if item_bindings != observation_result["card_bindings"]: raise core.CollectorError( "E_EVIDENCE_ITEM_BINDING", "Observed cards and evidence items are not an exact normalized binding.", safety=True, ) elif value["page_observation"] is not None: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Unreadable evidence must not claim page observations.") elif items: raise core.CollectorError("E_EVIDENCE_SCHEMA", "Unreadable evidence must not contain content items.") return value, payload, observation_result, identity_match def _formal_tokens(event: Mapping[str, Any], config: core.CollectorConfig) -> list[str]: stable = event.get("stable_id") tokens: list[str] = [] legacy_image: re.Match[str] | None = None if isinstance(stable, str) and stable: if re.fullmatch(r"BV[0-9A-Za-z]{10}", stable, re.IGNORECASE): tokens.append(f"bvid:{stable.lower()}") elif re.fullmatch(r"[0-9]{1,32}", stable): tokens.append(f"opus:{stable}") else: legacy_image = re.fullmatch(r"([0-9]{1,32}):(image|cover):([1-9][0-9]*)", stable) if legacy_image is not None and ( event.get("item_type") == legacy_image.group(2) and event.get("source_parent_stable_id") == legacy_image.group(1) ): # Exact legacy image rows are independently deduped by their # canonical image URL. Do not merge them into the parent opus # component because one parent can legitimately own many URLs. pass elif not tokens: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal stable_id is invalid.", safety=True) source = event.get("source_url") if isinstance(source, str): if legacy_image is not None: parsed = urlsplit(source) if ( parsed.scheme != "https" or parsed.hostname not in {"i0.hdslb.com", "i1.hdslb.com", "i2.hdslb.com"} or parsed.query or parsed.fragment or not parsed.path.startswith("/bfs/") ): raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal image URL identity conflicts.", safety=True) canonical = source else: canonical = core.validate_url(source, "formal.source_url", config.allowed_source_hosts) tokens.append(f"url:{canonical}") path_parts = PurePosixPath(canonical.split("?", 1)[0].split("#", 1)[0]).parts if len(path_parts) >= 3 and path_parts[-2] == "opus" and path_parts[-1].isdecimal(): url_token = f"opus:{path_parts[-1]}" if stable is not None and url_token not in tokens: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal URL/stable opus identity conflicts.", safety=True) tokens.append(url_token) elif len(path_parts) >= 3 and path_parts[-2] == "video" and re.fullmatch(r"BV[0-9A-Za-z]{10}", path_parts[-1], re.I): url_token = f"bvid:{path_parts[-1].lower()}" if stable is not None and url_token not in tokens: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal URL/stable BVID identity conflicts.", safety=True) tokens.append(url_token) return sorted(set(tokens)) def _catalog_artifact(config: core.CollectorConfig, event: Mapping[str, Any], prefix: str, *, required: bool) -> dict[str, Any] | None: fields = (f"{prefix}path", f"{prefix}bytes", f"{prefix}sha256") present = [event.get(field) is not None for field in fields] if not any(present): if required: raise core.CollectorError("E_CATALOG_ARTIFACT", "Required formal artifact triple is absent.", safety=True) return None if not all(present): raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact triple is partial.", safety=True) relative_text, expected_bytes, expected_hash = (event[field] for field in fields) if not isinstance(relative_text, str) or core.CONTROL_CHARACTER_PATTERN.search(relative_text): raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact path is invalid.", safety=True) relative = PurePosixPath(relative_text) if relative.is_absolute() or not relative.parts or ".." in relative.parts or relative.as_posix() != relative_text: raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact path is not archive-relative POSIX.", safety=True) if not isinstance(expected_bytes, int) or isinstance(expected_bytes, bool) or expected_bytes < 0: raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact byte count is invalid.", safety=True) if not isinstance(expected_hash, str) or re.fullmatch(r"[0-9A-Fa-f]{64}", expected_hash) is None: raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact SHA-256 is invalid.", safety=True) artifact = core.absolute_lexical(config.refresh.archive_dir.joinpath(*relative.parts)) try: core.lexical_lstat_chain(artifact, allow_missing_leaf=False) except core.CollectorError as exc: raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact path cannot be verified.", safety=True) from exc if not core.path_within(artifact, core.absolute_lexical(config.refresh.archive_dir)) or not artifact.is_file(): raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact is absent or outside archive.", safety=True) if artifact.stat().st_size != expected_bytes or core.sha256_file(artifact) != expected_hash.lower(): raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal artifact bytes/hash mismatch.", safety=True) return {"path": relative_text, "bytes": expected_bytes, "sha256": expected_hash.lower()} def _catalog_schema2_artifacts(config: core.CollectorConfig, event: Mapping[str, Any]) -> list[dict[str, Any]]: raw = event.get("artifacts") if not isinstance(raw, list) or not raw: raise core.CollectorError("E_CATALOG_ARTIFACT", "Schema2 saved event requires artifacts.", safety=True) result: list[dict[str, Any]] = [] identities: set[tuple[str, int]] = set() paths: set[str] = set() for index, value in enumerate(raw): row = _exact_keys(value, {"kind", "sequence", "path", "bytes", "sha256"}, f"formal.artifacts[{index}]") if row["kind"] not in {"text", "image", "cover"} or not isinstance(row["sequence"], int) or isinstance(row["sequence"], bool) or row["sequence"] < 1: raise core.CollectorError("E_CATALOG_ARTIFACT", "Schema2 artifact identity is invalid.", safety=True) identity = (row["kind"], row["sequence"]) if identity in identities or row["path"] in paths: raise core.CollectorError("E_CATALOG_ARTIFACT", "Schema2 artifact identity/path is duplicated.", safety=True) identities.add(identity) paths.add(row["path"]) synthetic = {"path": row["path"], "bytes": row["bytes"], "sha256": row["sha256"]} validated = _catalog_artifact(config, synthetic, "", required=True) assert validated is not None result.append({"kind": row["kind"], "sequence": row["sequence"], **validated}) return result def _validate_legacy_formal_scalars(value: Any, path: str = "$formal") -> None: """Legacy rows contain words such as auth/session in audited status prose. Reject credential-bearing fields and unsafe scalar shapes without falsely rejecting the frozen historical status vocabulary. """ if isinstance(value, Mapping): for key, child in value.items(): key_text = str(key) if re.search(r"(?:password|passwd|cookie|access_token|refresh_token|captcha)", key_text, re.I): raise core.CollectorError("E_SECRET_FIELD", "Credential fields are forbidden in formal history.", safety=True) _validate_legacy_formal_scalars(child, f"{path}.{key_text}") elif isinstance(value, list): for index, child in enumerate(value): _validate_legacy_formal_scalars(child, f"{path}[{index}]") elif isinstance(value, str): if len(value.encode("utf-8")) > 16384 or core.CONTROL_CHARACTER_PATTERN.search(value): raise core.CollectorError("E_CATALOG", "Formal history contains an unsafe scalar.", safety=True) elif value is not None and not isinstance(value, (bool, int, float)): raise core.CollectorError("E_CATALOG", "Formal history contains an unsupported scalar.", safety=True) def load_formal_catalog(config: core.CollectorConfig) -> tuple[list[dict[str, Any]], set[str], dict[str, int]]: refresh = config.refresh assert refresh is not None path = refresh.formal_manifest if not path.exists(): return [], set(), {"events": 0, "components": 0, "saved": 0, "video": 0, "retryable": 0} raw = path.read_bytes() if raw and not raw.endswith(b"\n"): raise core.CollectorError("E_CATALOG", "Formal manifest must end in LF.", safety=True) events: list[dict[str, Any]] = [] rows: list[dict[str, Any]] = [] for line_number, line in enumerate(raw.splitlines(), 1): try: event = _strict_json_bytes(line, f"formal line {line_number}") except core.CollectorError as exc: raise core.CollectorError("E_CATALOG", "Formal manifest JSON is invalid.", safety=True) from exc _validate_legacy_formal_scalars(event, f"$formal[{line_number}]") if not isinstance(event, dict) or event.get("creator") != config.creator_name: raise core.CollectorError("E_CATALOG", "Formal creator identity is invalid.", safety=True) uid = event.get("creator_uid") if uid is not None and str(uid) != config.creator_uid: raise core.CollectorError("E_CATALOG", "Formal creator UID conflicts.", safety=True) if event.get("schema_version") == 1: status = event.get("status") if status not in FORMAL_STATUS_ALLOWLIST: raise core.CollectorError("E_CATALOG_STATUS", "Formal status is not registered.", safety=True) if not isinstance(event.get("stable_id"), str) or not isinstance(event.get("source_url"), str) or not isinstance(event.get("published_at"), str): raise core.CollectorError("E_CATALOG", "Schema1 identity/time fields are invalid.", safety=True) core.parse_datetime(event["published_at"], "formal.published_at") row_tokens = _formal_tokens(event, config) if not row_tokens: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal row has no stable token.", safety=True) outcome = "CONTENT_SAVED" if status == "SAVED" else ("CONTENT_RETRYABLE" if status in RETRYABLE_CONTENT else "VIDEO_TRACKED") if status == "SAVED": _catalog_artifact(config, event, "", required=True) _catalog_artifact(config, event, "image_", required=False) _catalog_artifact(config, event, "cover_", required=False) for key, value in event.items(): if key.endswith("_sha256") and value is not None and (not isinstance(value, str) or re.fullmatch(r"[0-9A-Fa-f]{64}", value) is None): raise core.CollectorError("E_CATALOG_ARTIFACT", "Formal SHA-256 scalar is invalid.", safety=True) domain = "video" if any(token.startswith("bvid:") for token in row_tokens) else "content" rows.append({"tokens": row_tokens, "outcome": outcome, "domain": domain, "schema2_entity_id": None}) elif event.get("schema_version") == 2: expected_keys = { "schema_version", "event_type", "creator", "creator_uid", "entity_id", "dynamic_id", "opus_id", "bvid", "dedupe_keys", "content_type", "published_at", "title", "source_url", "collected_at", "artifacts", "duration_seconds", "page_run_id", "coverage_complete", "status", } _exact_keys(event, expected_keys, f"formal line {line_number}") if ( event["event_type"] != "DYNAMIC_CONTENT_SAVED" or event["status"] != "SAVED" or not isinstance(event["creator_uid"], str) or event["creator_uid"] != config.creator_uid ): raise core.CollectorError("E_CATALOG_STATUS", "Schema2 event/status/UID is invalid.", safety=True) normalized = core.normalize_item(event, config, line_number) raw_keys = event["dedupe_keys"] if raw_keys != normalized["dedupe_keys"]: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Schema2 dedupe_keys are invalid.", safety=True) if not isinstance(event["entity_id"], str) or re.fullmatch(r"[0-9a-f]{24}", event["entity_id"]) is None: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Schema2 entity_id shape is invalid.", safety=True) core.parse_datetime(event["collected_at"], "formal.collected_at") if not isinstance(event["page_run_id"], str) or re.fullmatch(r"[0-9a-f]{32}", event["page_run_id"]) is None or not isinstance(event["coverage_complete"], bool): raise core.CollectorError("E_CATALOG", "Schema2 run/coverage identity is invalid.", safety=True) _catalog_schema2_artifacts(config, event) rows.append({ "tokens": raw_keys, "outcome": "CONTENT_SAVED", "domain": "video" if normalized["content_type"] == "video" else "content", "schema2_entity_id": event["entity_id"], }) else: raise core.CollectorError("E_CATALOG", "Formal schema/event is unsupported.", safety=True) events.append(event) parent = list(range(len(rows))) def find(index: int) -> int: while parent[index] != index: parent[index] = parent[parent[index]] index = parent[index] return index def union(left: int, right: int) -> None: a, b = find(left), find(right) if a != b: parent[max(a, b)] = min(a, b) owner_by_token: dict[str, int] = {} for index, row in enumerate(rows): for token in row["tokens"]: if not isinstance(token, str) or ":" not in token or core.CONTROL_CHARACTER_PATTERN.search(token): raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal token is invalid.", safety=True) prior = owner_by_token.setdefault(token, index) union(index, prior) components: dict[int, list[int]] = {} for index in range(len(rows)): components.setdefault(find(index), []).append(index) all_tokens: set[str] = set() final_outcomes: list[str] = [] for indexes in components.values(): component_tokens = sorted({token for index in indexes for token in rows[index]["tokens"]}) namespaces: dict[str, set[str]] = {} for token in component_tokens: namespace, value = token.split(":", 1) if namespace not in {"dynamic", "opus", "bvid", "url"}: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal token namespace is unsupported.", safety=True) namespaces.setdefault(namespace, set()).add(value) if any(len(values) > 1 for values in namespaces.values()): raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal component contains conflicting namespace identities.", safety=True) domains = {rows[index]["domain"] for index in indexes} if len(domains) != 1: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal component mixes content and video identities.", safety=True) expected_entity = core.entity_id_for_keys(component_tokens) for index in indexes: identity = rows[index]["schema2_entity_id"] if identity is not None and identity != expected_entity: raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Schema2 entity_id does not bind the global component.", safety=True) outcomes = {rows[index]["outcome"] for index in indexes} if "CONTENT_SAVED" in outcomes: final_outcomes.append("CONTENT_SAVED") elif domains == {"video"} and outcomes == {"VIDEO_TRACKED"}: final_outcomes.append("VIDEO_TRACKED") elif outcomes == {"CONTENT_RETRYABLE"}: final_outcomes.append("CONTENT_RETRYABLE") else: raise core.CollectorError("E_CATALOG_STATUS", "Formal component outcome combination is invalid.", safety=True) all_tokens.update(component_tokens) counts = { "events": len(events), "components": len(final_outcomes), "saved": final_outcomes.count("CONTENT_SAVED"), "video": final_outcomes.count("VIDEO_TRACKED"), "retryable": final_outcomes.count("CONTENT_RETRYABLE"), } return events, all_tokens, counts def _normalize_evidence_item(raw: Any, config: core.CollectorConfig, index: int) -> dict[str, Any]: expected = { "dynamic_id", "opus_id", "bvid", "content_type", "published_at", "title", "source_url", "body_text", "body_complete", "duration_seconds", "artifacts", } value = _exact_keys(raw, expected, f"items[{index}]") normalized = core.normalize_item(value, config, index) normalized.update({ "body_text": value["body_text"], "body_complete": value["body_complete"], "duration_seconds": value["duration_seconds"], "artifacts": value["artifacts"], }) if normalized["content_type"] in {"text", "article"}: if not isinstance(value["body_text"], str) or not value["body_complete"]: raise core.CollectorError("E_CONTENT_INCOMPLETE", "Text/article requires complete body text.") body = value["body_text"].replace("\r\n", "\n").replace("\r", "\n").rstrip("\n") + "\n" if len(body.encode("utf-8")) > config.refresh.max_text_bytes: raise core.CollectorError("E_CONTENT_LIMIT", "Text body exceeds the byte limit.") normalized["body_bytes"] = body.encode("utf-8") elif normalized["content_type"] == "video": if not normalized["bvid"] or not isinstance(value["duration_seconds"], (int, float)) or value["duration_seconds"] <= 0: raise core.CollectorError("E_CONTENT_INCOMPLETE", "Video requires BVID and positive duration.") if not isinstance(value["artifacts"], list): raise core.CollectorError("E_EVIDENCE_SCHEMA", "items.artifacts must be a list.") return normalized def _intake_artifact(config: core.CollectorConfig, pending: Mapping[str, Any], raw: Any, item_type: str) -> dict[str, Any]: value = _exact_keys(raw, {"kind", "sequence", "path", "extension", "bytes", "sha256", "source_url"}, "artifact") if value["kind"] not in {"image", "cover"} or not isinstance(value["sequence"], int): raise core.CollectorError("E_ARTIFACT", "Artifact kind/sequence is invalid.") if item_type == "video" and value["kind"] != "cover": raise core.CollectorError("E_ARTIFACT", "Video may contain only one cover artifact.") relative = PurePosixPath(value["path"]) if relative.is_absolute() or ".." in relative.parts or not relative.parts: raise core.CollectorError("E_ARTIFACT_PATH", "Artifact path must be a safe intake-relative path.", safety=True) intake_root = core.absolute_lexical(Path(pending["intake_root"])) source = core.absolute_lexical(intake_root.joinpath(*relative.parts)) core.lexical_lstat_chain(source, allow_missing_leaf=False) if not core.path_within(source, intake_root) or not source.is_file(): raise core.CollectorError("E_ARTIFACT_PATH", "Artifact is missing or outside intake root.", safety=True) extension = value["extension"] if extension not in {".jpg", ".jpeg", ".png", ".webp"} or source.suffix.lower() != extension: raise core.CollectorError("E_ARTIFACT", "Artifact extension is invalid.") if source.stat().st_size != value["bytes"] or core.sha256_file(source) != value["sha256"]: raise core.CollectorError("E_ARTIFACT_HASH", "Artifact bytes/hash mismatch.", safety=True) if value["bytes"] > config.refresh.max_image_bytes: raise core.CollectorError("E_ARTIFACT", "Artifact exceeds the byte limit.") head = source.read_bytes()[:12] if extension in {".jpg", ".jpeg"} and not head.startswith(b"\xff\xd8\xff"): raise core.CollectorError("E_ARTIFACT", "JPEG magic mismatch.") if extension == ".png" and not head.startswith(b"\x89PNG\r\n\x1a\n"): raise core.CollectorError("E_ARTIFACT", "PNG magic mismatch.") if extension == ".webp" and not (head.startswith(b"RIFF") and head[8:12] == b"WEBP"): raise core.CollectorError("E_ARTIFACT", "WebP magic mismatch.") core.validate_url(value["source_url"], "artifact.source_url", config.allowed_source_hosts) return {**value, "source": source} def _plan_content( config: core.CollectorConfig, pending: Mapping[str, Any], evidence: Mapping[str, Any], formal_tokens: set[str], now: datetime, ) -> list[dict[str, Any]]: planned: list[dict[str, Any]] = [] seen: set[str] = set() latest, token_map = core.latest_entities(core.load_manifest(config.manifest_path)) cutoff = core.parse_datetime(pending["window_start"], "window_start") end = core.parse_datetime(pending["window_end"], "window_end") for index, raw in enumerate(evidence["items"]): item = _normalize_evidence_item(raw, config, index) published = core.parse_datetime(item["published_at"], "published_at") if published < cutoff or published > end: continue keys = item["dedupe_keys"] if seen.intersection(keys): continue seen.update(keys) if formal_tokens.intersection(keys) or core.resolve_entity(keys, token_map): continue entity = core.entity_id_for_keys(keys) stem = core.sanitize_windows_component(core.suggested_base(item, config), 86) + f"_{entity[:8]}" artifacts: list[dict[str, Any]] = [] if item["content_type"] in {"text", "article"}: artifacts.append({"kind": "text", "sequence": 1, "payload": item["body_bytes"], "target": f"{stem}.txt"}) raw_artifacts = item["artifacts"] if item["content_type"] == "image" and not 1 <= len(raw_artifacts) <= config.refresh.max_images_per_item: raise core.CollectorError("E_CONTENT_INCOMPLETE", "Image item requires 1..20 original images.") if item["content_type"] == "video" and len(raw_artifacts) != 1: raise core.CollectorError("E_CONTENT_INCOMPLETE", "Video requires exactly one cover.") seen_sequences: set[int] = set() for raw_artifact in raw_artifacts: artifact = _intake_artifact(config, pending, raw_artifact, item["content_type"]) if artifact["sequence"] in seen_sequences: raise core.CollectorError("E_ARTIFACT", "Artifact sequence is duplicated.") seen_sequences.add(artifact["sequence"]) target = f"{stem}_{artifact['kind']}-{artifact['sequence']:02d}{artifact['extension']}" artifacts.append({**artifact, "payload": artifact["source"].read_bytes(), "target": target}) planned.append({"item": item, "entity_id": entity, "stem": stem, "artifacts": artifacts}) return sorted(planned, key=lambda row: row["entity_id"]) def _event_lines(config: core.CollectorConfig, planned: Sequence[Mapping[str, Any]], run_id: str, now: datetime, coverage: bool) -> tuple[bytes, bytes, list[dict[str, Any]]]: state_lines = b"" formal_lines = b"" created: list[dict[str, Any]] = [] for row in planned: item = row["item"] artifact_refs: list[dict[str, Any]] = [] for artifact in row["artifacts"]: payload = artifact["payload"] ref = {"kind": artifact["kind"], "sequence": artifact["sequence"], "path": artifact["target"], "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest()} artifact_refs.append(ref) created.append(ref) state = core.manifest_event( config, item, entity_id=row["entity_id"], status="TODO_QUEUED" if item["content_type"] == "video" else "CONTENT_SAVED", collected_at=now, suggested_stem=row["stem"], ) state["event_id"] = hashlib.sha256(f"{run_id}\n{row['entity_id']}\nstate".encode()).hexdigest()[:32] state["artifact_refs"] = artifact_refs formal = { "schema_version": 2, "event_type": "DYNAMIC_CONTENT_SAVED", "creator": config.creator_name, "creator_uid": config.creator_uid, "entity_id": row["entity_id"], "dynamic_id": item["dynamic_id"], "opus_id": item["opus_id"], "bvid": item["bvid"], "dedupe_keys": item["dedupe_keys"], "content_type": item["content_type"], "published_at": item["published_at"], "title": item["title"], "source_url": item["source_url"], "collected_at": core.canonical_datetime(now), "artifacts": artifact_refs, "duration_seconds": item["duration_seconds"], "page_run_id": run_id, "coverage_complete": coverage, "status": "SAVED", } state_lines += core.canonical_json_bytes(state) formal_lines += core.canonical_json_bytes(formal) return state_lines, formal_lines, created def _formal_lock_path(config: core.CollectorConfig) -> Path: assert config.refresh is not None return config.refresh.archive_dir / FORMAL_LOCK_NAME def _acquire_formal_lock(config: core.CollectorConfig, pending: dict[str, Any], now: datetime) -> None: path = _formal_lock_path(config) transaction_id = pending["transaction_identity"]["transaction_id"] record = { "schema_version": 1, "task_id": TASK_ID, "run_id": pending["run_id"], "owner_nonce": pending["owner_nonce"], "transaction_id": transaction_id, "recovery_generation": 0, "holder_pid": os.getpid(), "holder_process_created_at": core.process_created_at(os.getpid()), "lock_created_at": core.canonical_datetime(now), } payload = core.canonical_json_bytes(record, newline=False) claim = {"claim_state": "PLANNED", "lock_path": str(path), "lock_bytes": len(payload), "lock_sha256": hashlib.sha256(payload).hexdigest(), "lock_record": record} pending["transaction_identity"]["formal_lock_claim"] = claim _write_pending(config, pending) try: _durable_formal_lock_claim(config, claim, create=True) except core.CollectorError as exc: if exc.code == "E_ALREADY_EXISTS": raise core.CollectorError("E_FORMAL_LOCK_BUSY", "Formal manifest lock is held.", safety=True) from exc raise claim["claim_state"] = "HELD" pending["transaction_identity"]["formal_lock_claim"] = claim _write_pending(config, pending) def _formal_owner_state(record: Mapping[str, Any]) -> str: pid = record.get("holder_pid") created = record.get("holder_process_created_at") if not isinstance(pid, int) or not isinstance(created, str): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock owner identity is invalid.", safety=True) try: actual = core.process_created_at(pid) except ProcessLookupError: return "DEAD" except OSError as exc: raise core.CollectorError("E_FORMAL_LOCK_OWNER_UNPROVEN", "Formal lock owner cannot be proven.", safety=True) from exc if actual != created: return "PID_REUSED" if pid == os.getpid() and actual == core.process_created_at(os.getpid()): return "CURRENT" return "ALIVE" def _claim_payload(claim: Mapping[str, Any]) -> bytes: record = claim.get("lock_record") if not isinstance(record, Mapping): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock claim record is invalid.", safety=True) payload = core.canonical_json_bytes(record, newline=False) if len(payload) != claim.get("lock_bytes") or hashlib.sha256(payload).hexdigest() != claim.get("lock_sha256"): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock claim bytes drifted.", safety=True) return payload def _new_formal_claim(config: core.CollectorConfig, pending: dict[str, Any], generation: int, now: datetime) -> dict[str, Any]: record = { "schema_version": 1, "task_id": TASK_ID, "run_id": pending["run_id"], "owner_nonce": pending["owner_nonce"], "transaction_id": pending["transaction_identity"]["transaction_id"], "recovery_generation": generation, "holder_pid": os.getpid(), "holder_process_created_at": core.process_created_at(os.getpid()), "lock_created_at": core.canonical_datetime(now), } payload = core.canonical_json_bytes(record, newline=False) return { "claim_state": "PLANNED", "lock_path": str(_formal_lock_path(config)), "lock_bytes": len(payload), "lock_sha256": hashlib.sha256(payload).hexdigest(), "lock_record": record, } def _quarantine_root(config: core.CollectorConfig) -> Path: assert config.refresh is not None return config.refresh.archive_dir / f"{FORMAL_LOCK_NAME}.quarantine" def _directory_identity(path: Path) -> dict[str, Any]: lexical = core.absolute_lexical(path) core.lexical_lstat_chain(lexical, allow_missing_leaf=False) if not lexical.is_dir(): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Expected an ordinary recovery directory.", safety=True) stat = lexical.stat() return {"path": str(lexical), "device": int(stat.st_dev), "inode": int(stat.st_ino)} def _fsync_directory(path: Path) -> None: """Make a directory entry transition durable or fail closed.""" lexical = core.absolute_lexical(path) core.lexical_lstat_chain(lexical, allow_missing_leaf=False) if os.name != "nt": descriptor = os.open(lexical, os.O_RDONLY) try: os.fsync(descriptor) finally: os.close(descriptor) return import ctypes from ctypes import wintypes kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) create_file = kernel32.CreateFileW create_file.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE, ] create_file.restype = wintypes.HANDLE flush = kernel32.FlushFileBuffers flush.argtypes = [wintypes.HANDLE] flush.restype = wintypes.BOOL close = kernel32.CloseHandle close.argtypes = [wintypes.HANDLE] close.restype = wintypes.BOOL handle = create_file(str(lexical), 0xC0000000, 0x00000007, None, 3, 0x02000000, None) invalid = ctypes.c_void_p(-1).value if handle in (None, 0, invalid): raise core.CollectorError("E_FORMAL_LOCK", "Cannot open recovery directory for durable flush.", safety=True) try: if not flush(handle): raise core.CollectorError("E_FORMAL_LOCK", "Recovery directory durable flush failed.", safety=True) finally: close(handle) def _durable_formal_lock_claim( config: core.CollectorConfig, claim: Mapping[str, Any], *, create: bool, ) -> bytes: """Create/verify one claim and durably persist its directory entry before promotion.""" path = _formal_lock_path(config) if claim.get("lock_path") != str(path): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock path drifted.", safety=True) expected = _claim_payload(claim) if create: _create_new(path, expected) core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file(): raise core.CollectorError("E_FORMAL_LOCK", "Formal lock is not an ordinary file.", safety=True) actual = path.read_bytes() if actual != expected or len(actual) != claim.get("lock_bytes") or hashlib.sha256(actual).hexdigest() != claim.get("lock_sha256"): raise core.CollectorError("E_FORMAL_LOCK", "Formal lock exact readback mismatch.", safety=True) _fsync_directory(path.parent) core.lexical_lstat_chain(path, allow_missing_leaf=False) durable = path.read_bytes() if durable != expected or len(durable) != claim.get("lock_bytes") or hashlib.sha256(durable).hexdigest() != claim.get("lock_sha256"): raise core.CollectorError("E_FORMAL_LOCK", "Formal lock changed during durable persistence.", safety=True) return durable def _rename_no_overwrite(source: Path, target: Path) -> None: """Move one exact lock into quarantine without replacement and flush both parents.""" source = core.absolute_lexical(source) target = core.absolute_lexical(target) if target.exists(): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock quarantine target already exists.", safety=True) core.lexical_lstat_chain(source, allow_missing_leaf=False) core.lexical_lstat_chain(target, allow_missing_leaf=True) if os.stat(source.parent).st_dev != os.stat(target.parent).st_dev: raise core.CollectorError("E_FORMAL_LOCK_VOLUME_IDENTITY", "Formal lock quarantine is cross-volume.", safety=True) if os.name == "nt": import ctypes from ctypes import wintypes kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) move = kernel32.MoveFileExW move.argtypes = [wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD] move.restype = wintypes.BOOL if not move(str(source), str(target), 0x00000008): # MOVEFILE_WRITE_THROUGH; no REPLACE_EXISTING error = ctypes.get_last_error() raise core.CollectorError( "E_RECOVERY_AMBIGUOUS", f"No-overwrite quarantine rename failed with Win32 error {error}.", safety=True, ) else: try: os.link(source, target) except FileExistsError as exc: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock quarantine target appeared.", safety=True) from exc os.unlink(source) _fsync_directory(source.parent) if target.parent != source.parent: _fsync_directory(target.parent) def _ensure_quarantine_owner(config: core.CollectorConfig, pending: Mapping[str, Any]) -> Path: root = _quarantine_root(config) marker = root / ".owner.json" expected_value = { "schema_version": 1, "task_id": TASK_ID, "run_id": pending["run_id"], "owner_nonce": pending["owner_nonce"], } expected = core.canonical_json_bytes(expected_value, newline=False) if root.exists(): core.lexical_lstat_chain(root, allow_missing_leaf=False) if not root.is_dir() or not marker.is_file() or marker.read_bytes() != expected: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal quarantine owner marker is invalid.", safety=True) else: core.ensure_directory(root, create=True) _fsync_directory(root.parent) _create_new(marker, expected) _fsync_directory(root) core.lexical_lstat_chain(root, allow_missing_leaf=False) return root def _quarantine_binding(config: core.CollectorConfig, path: Path) -> dict[str, Any]: root = _quarantine_root(config) archive = core.absolute_lexical(config.refresh.archive_dir) path = core.absolute_lexical(path) if path.parent != core.absolute_lexical(root) or not core.path_within(path, archive): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Quarantine path is not mechanically bound.", safety=True) return { "quarantine_relative_path": path.relative_to(archive).as_posix(), "archive_directory_identity": _directory_identity(archive), "quarantine_directory_identity": _directory_identity(root), } def _validate_quarantine_binding( config: core.CollectorConfig, value: Mapping[str, Any], path: Path, ) -> None: expected = _quarantine_binding(config, path) for key, wanted in expected.items(): if value.get(key) != wanted: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Quarantine directory binding drifted.", safety=True) def _recover_failed_claim_intent( config: core.CollectorConfig, pending: dict[str, Any], takeover: dict[str, Any], ) -> None: intent = takeover.get("failed_claim_intent") if intent is None: return if not isinstance(intent, dict) or set(intent) != { "path", "quarantine_relative_path", "archive_directory_identity", "quarantine_directory_identity", "bytes", "sha256", "claim_attempt", }: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Failed-claim quarantine intent is invalid.", safety=True) path = _formal_lock_path(config) failed_path = core.absolute_lexical(Path(intent["path"])) _validate_quarantine_binding(config, intent, failed_path) old_claim = pending["transaction_identity"]["formal_lock_claim"] expected = _claim_payload(old_claim) if len(expected) != intent["bytes"] or hashlib.sha256(expected).hexdigest() != intent["sha256"]: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Failed-claim quarantine identity drifted.", safety=True) source_exists = path.exists() target_exists = failed_path.exists() if source_exists and target_exists: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Failed claim exists at both source and quarantine.", safety=True) if source_exists: if not path.is_file() or path.read_bytes() != expected: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Failed formal claim source drifted.", safety=True) _rename_no_overwrite(path, failed_path) elif not target_exists: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Failed formal claim disappeared before quarantine.", safety=True) core.lexical_lstat_chain(failed_path, allow_missing_leaf=False) if not failed_path.is_file() or failed_path.read_bytes() != expected or path.exists(): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Failed formal claim quarantine readback mismatch.", safety=True) record = { "path": str(failed_path), "quarantine_relative_path": intent["quarantine_relative_path"], "archive_directory_identity": intent["archive_directory_identity"], "quarantine_directory_identity": intent["quarantine_directory_identity"], "bytes": intent["bytes"], "sha256": intent["sha256"], "claim_attempt": intent["claim_attempt"], } failed = takeover.setdefault("failed_claims", []) if record not in failed: failed.append(record) takeover.pop("failed_claim_intent") pending["transaction_identity"]["takeover"] = takeover _write_pending(config, pending) def _rebind_takeover_claim( config: core.CollectorConfig, pending: dict[str, Any], takeover: dict[str, Any], now: datetime, *, isolate_existing: bool, advance_generation: bool, ) -> None: txn = pending["transaction_identity"] old_claim = txn["formal_lock_claim"] path = _formal_lock_path(config) old_payload = _claim_payload(old_claim) takeover.setdefault("failed_claims", []) if isolate_existing: root = _ensure_quarantine_owner(config, pending) attempt = int(takeover.get("claim_attempt", 0)) failed_path = root / ( f"{pending['run_id']}-g{old_claim['lock_record']['recovery_generation']}" f"-attempt{attempt}-{old_claim['lock_sha256']}.json" ) if failed_path.exists() or not path.is_file() or path.read_bytes() != old_payload: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Failed formal claim cannot be isolated exactly.", safety=True) binding = _quarantine_binding(config, failed_path) takeover["failed_claim_intent"] = { "path": str(failed_path), **binding, "bytes": len(old_payload), "sha256": hashlib.sha256(old_payload).hexdigest(), "claim_attempt": attempt, } txn["takeover"] = takeover _write_pending(config, pending) _recover_failed_claim_intent(config, pending, takeover) generation = int(takeover["next_generation"]) + (1 if advance_generation else 0) takeover["next_generation"] = generation takeover["claim_attempt"] = int(takeover.get("claim_attempt", 0)) + 1 new_claim = _new_formal_claim(config, pending, generation, now) txn["formal_lock_claim"] = new_claim takeover["phase"] = "CREATE_PLANNED" txn["takeover"] = takeover _write_pending(config, pending) _durable_formal_lock_claim(config, new_claim, create=True) new_claim["claim_state"] = "HELD" takeover["phase"] = "NEW_LOCK_HELD" txn["formal_lock_claim"] = new_claim txn["takeover"] = takeover _write_pending(config, pending) def _recovery_formal_lock(config: core.CollectorConfig, pending: dict[str, Any], now: datetime) -> None: """Reopen the same-run formal lock without deleting an unproven holder.""" txn = pending.get("transaction_identity") if not isinstance(txn, dict): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Transaction identity is missing.", safety=True) claim = txn.get("formal_lock_claim") if not isinstance(claim, dict) or claim.get("claim_state") not in {"PLANNED", "HELD"}: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock claim is missing.", safety=True) path = _formal_lock_path(config) if claim.get("lock_path") != str(path): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock path drifted.", safety=True) expected = _claim_payload(claim) takeover = txn.get("takeover") quarantine_root = _quarantine_root(config) if isinstance(takeover, dict): qpath = core.absolute_lexical(Path(takeover.get("quarantine_path", ""))) if not core.path_within(qpath, quarantine_root) or qpath.parent != quarantine_root: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock quarantine path drifted.", safety=True) old_payload = bytes.fromhex(takeover.get("old_lock_hex", "")) if hashlib.sha256(old_payload).hexdigest() != takeover.get("old_lock_sha256"): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock quarantine identity drifted.", safety=True) phase = takeover.get("phase") if phase == "TAKEOVER_PLANNED": _ensure_quarantine_owner(config, pending) _validate_quarantine_binding(config, takeover, qpath) if path.exists() and path.read_bytes() == old_payload and not qpath.exists(): if os.stat(path.parent).st_dev != os.stat(qpath.parent).st_dev: raise core.CollectorError("E_FORMAL_LOCK_VOLUME_IDENTITY", "Formal lock quarantine is cross-volume.", safety=True) _rename_no_overwrite(path, qpath) if not qpath.is_file() or qpath.read_bytes() != old_payload or path.exists(): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal takeover rename state is ambiguous.", safety=True) takeover["phase"] = "OLD_LOCK_QUARANTINED" txn["takeover"] = takeover _write_pending(config, pending) phase = "OLD_LOCK_QUARANTINED" if phase == "OLD_LOCK_QUARANTINED": if path.exists() or not qpath.is_file() or qpath.read_bytes() != old_payload: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal takeover quarantine state is ambiguous.", safety=True) new_claim = _new_formal_claim(config, pending, int(takeover["next_generation"]), now) txn["formal_lock_claim"] = new_claim takeover["phase"] = "CREATE_PLANNED" takeover["claim_attempt"] = int(takeover.get("claim_attempt", 0)) + 1 txn["takeover"] = takeover _write_pending(config, pending) _durable_formal_lock_claim(config, new_claim, create=True) new_claim["claim_state"] = "HELD" takeover["phase"] = "NEW_LOCK_HELD" txn["formal_lock_claim"] = new_claim txn["takeover"] = takeover _write_pending(config, pending) return if phase in {"CREATE_PLANNED", "NEW_LOCK_HELD"}: if not takeover.get("old_lock_never_created"): _validate_quarantine_binding(config, takeover, qpath) _recover_failed_claim_intent(config, pending, takeover) current_claim = txn["formal_lock_claim"] current_payload = _claim_payload(current_claim) if path.exists() and path.read_bytes() != current_payload: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Recovery claim bytes conflict.", safety=True) state = _formal_owner_state(current_claim["lock_record"]) if path.exists() and state == "CURRENT": _durable_formal_lock_claim(config, current_claim, create=False) current_claim["claim_state"] = "HELD" takeover["phase"] = "NEW_LOCK_HELD" _write_pending(config, pending) return if state == "ALIVE": raise core.CollectorError("E_FORMAL_LOCK_BUSY", "Recovery formal lock owner is alive.", safety=True) if not path.exists() and state == "CURRENT": _durable_formal_lock_claim(config, current_claim, create=True) current_claim["claim_state"] = "HELD" takeover["phase"] = "NEW_LOCK_HELD" _write_pending(config, pending) return if state in {"DEAD", "PID_REUSED"}: _rebind_takeover_claim( config, pending, takeover, now, isolate_existing=path.exists(), advance_generation=phase == "NEW_LOCK_HELD", ) return raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Recovery claim cannot be rebound safely.", safety=True) raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Unknown formal takeover phase.", safety=True) if not path.exists(): if claim["claim_state"] != "PLANNED": raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Held formal lock disappeared.", safety=True) state = _formal_owner_state(claim["lock_record"]) if state == "ALIVE": raise core.CollectorError("E_FORMAL_LOCK_BUSY", "Planned formal lock owner is alive.", safety=True) if state == "CURRENT": _durable_formal_lock_claim(config, claim, create=True) claim["claim_state"] = "HELD" _write_pending(config, pending) return new_claim = _new_formal_claim( config, pending, int(claim["lock_record"]["recovery_generation"]) + 1, now ) txn["formal_lock_claim"] = new_claim txn["takeover"] = { "phase": "CREATE_PLANNED", "old_lock_sha256": claim["lock_sha256"], "old_lock_hex": expected.hex(), "quarantine_path": str(quarantine_root / f"{pending['run_id']}-never-created-{claim['lock_sha256']}.json"), "next_generation": int(new_claim["lock_record"]["recovery_generation"]), "takeover_reason": "OWNER_DEAD_BEFORE_CREATE", "planned_at": core.canonical_datetime(now), "claim_attempt": 1, "old_lock_never_created": True, } _write_pending(config, pending) _durable_formal_lock_claim(config, new_claim, create=True) new_claim["claim_state"] = "HELD" txn["takeover"]["phase"] = "NEW_LOCK_HELD" _write_pending(config, pending) return else: core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file() or path.read_bytes() != expected: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock bytes are not the pending claim.", safety=True) state = _formal_owner_state(claim["lock_record"]) if state == "CURRENT": _durable_formal_lock_claim(config, claim, create=False) claim["claim_state"] = "HELD" _write_pending(config, pending) return if state == "ALIVE": raise core.CollectorError("E_FORMAL_LOCK_BUSY", "Formal lock owner is alive.", safety=True) _ensure_quarantine_owner(config, pending) quarantine = quarantine_root / f"{pending['run_id']}-g{claim['lock_record']['recovery_generation']}-{claim['lock_sha256']}.json" if quarantine.exists(): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Formal lock quarantine already exists.", safety=True) txn["takeover"] = { "phase": "TAKEOVER_PLANNED", "old_lock_sha256": claim["lock_sha256"], "old_lock_hex": expected.hex(), "quarantine_path": str(quarantine), "next_generation": int(claim["lock_record"]["recovery_generation"]) + 1, "takeover_reason": "OWNER_DEAD_PROVEN", "planned_at": core.canonical_datetime(now), "claim_attempt": 0, **_quarantine_binding(config, quarantine), } _write_pending(config, pending) _recovery_formal_lock(config, pending, now) def _release_formal_lock(config: core.CollectorConfig, pending: Mapping[str, Any]) -> None: claim = (pending.get("transaction_identity") or {}).get("formal_lock_claim") path = _formal_lock_path(config) if not claim: return if path.exists(): payload = path.read_bytes() if len(payload) != claim["lock_bytes"] or hashlib.sha256(payload).hexdigest() != claim["lock_sha256"]: raise core.CollectorError("E_FORMAL_LOCK", "Formal lock identity drifted.", safety=True) path.unlink() _fsync_directory(path.parent) takeover = (pending.get("transaction_identity") or {}).get("takeover") if isinstance(takeover, Mapping) and takeover.get("quarantine_path") and not takeover.get("old_lock_never_created"): quarantine = Path(str(takeover["quarantine_path"])) _validate_quarantine_binding(config, takeover, quarantine) if quarantine.exists(): expected = bytes.fromhex(str(takeover["old_lock_hex"])) if quarantine.is_file() and quarantine.read_bytes() == expected: quarantine.unlink() _fsync_directory(quarantine.parent) else: raise core.CollectorError("E_FORMAL_LOCK", "Formal lock quarantine cleanup identity drifted.", safety=True) if isinstance(takeover, Mapping): if takeover.get("failed_claim_intent") is not None: raise core.CollectorError("E_FORMAL_LOCK", "Unfinished failed-claim quarantine intent blocks cleanup.", safety=True) for failed in takeover.get("failed_claims", []): failed_path = Path(str(failed.get("path", ""))) _validate_quarantine_binding(config, failed, failed_path) if failed_path.exists(): payload = failed_path.read_bytes() if ( failed_path.is_file() and len(payload) == failed.get("bytes") and hashlib.sha256(payload).hexdigest() == failed.get("sha256") ): failed_path.unlink() _fsync_directory(failed_path.parent) else: raise core.CollectorError("E_FORMAL_LOCK", "Failed formal claim cleanup identity drifted.", safety=True) root = _quarantine_root(config) marker = root / ".owner.json" if root.exists() and set(root.iterdir()) == {marker}: expected = core.canonical_json_bytes( {"schema_version": 1, "task_id": TASK_ID, "run_id": pending["run_id"], "owner_nonce": pending["owner_nonce"]}, newline=False, ) if marker.is_file() and marker.read_bytes() == expected: marker.unlink() _fsync_directory(root) root.rmdir() _fsync_directory(root.parent) else: raise core.CollectorError("E_FORMAL_LOCK", "Formal quarantine owner cleanup identity drifted.", safety=True) elif root.exists() and not any(root.iterdir()): # Crash-reopen after the exact owner marker unlink but before rmdir. root.rmdir() _fsync_directory(root.parent) elif not root.exists(): _fsync_directory(root.parent) def _receipt_for_terminal( pending: Mapping[str, Any], state_id: Mapping[str, Any], formal_id: Mapping[str, Any] ) -> dict[str, Any]: txn = pending.get("transaction_identity") if not isinstance(txn, Mapping): intent = { "run_id": pending["run_id"], "business_commit_kind": "NO_FORMAL_CHANGE", "state_manifest": state_id, "formal_manifest": formal_id, } return { "phase": "NO_FORMAL_CHANGE", "transaction_id": hashlib.sha256(core.canonical_json_bytes(intent, newline=False)).hexdigest()[:32], "intent_sha256": hashlib.sha256(core.canonical_json_bytes(intent, newline=False)).hexdigest(), "business_commit_kind": "NO_FORMAL_CHANGE", "state_preimage": dict(state_id), "formal_preimage": dict(formal_id), "state_candidate": dict(state_id), "formal_candidate": dict(formal_id), "created_artifacts": [], "recovery_count": 0, } intent = { "transaction_id": txn["transaction_id"], "state_preimage": txn["state_preimage"], "formal_preimage": txn["formal_preimage"], "state_candidate": txn["state_candidate"], "formal_candidate": txn["formal_candidate"], "created_artifacts": txn["created_artifacts"], } return { "phase": "BUSINESS_COMMITTED", "transaction_id": txn["transaction_id"], "intent_sha256": hashlib.sha256(core.canonical_json_bytes(intent, newline=False)).hexdigest(), "business_commit_kind": "DUAL_MANIFEST_COMMIT", "state_preimage": txn["state_preimage"], "formal_preimage": txn["formal_preimage"], "state_candidate": txn["state_candidate"], "formal_candidate": txn["formal_candidate"], "created_artifacts": txn["created_artifacts"], "recovery_count": int(txn.get("recovery_count", 0)), } def _latest_index(slot: Mapping[str, Any], slot_payload: bytes) -> dict[str, Any]: return { "schema_version": SLOT_SCHEMA, "slot_index": slot["slot_index"], "hour_epoch": slot["hour_epoch"], "run_id": slot["run_id"], "slot_bytes": len(slot_payload), "slot_sha256": hashlib.sha256(slot_payload).hexdigest(), "status": slot["status"], "terminal_at": slot["terminal_at"], } def _write_readback(path: Path, payload: bytes, description: str) -> None: core.atomic_replace_bytes(path, payload) core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file() or path.read_bytes() != payload: raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", f"{description} durable readback mismatch.", safety=True) def _write_terminal_slot( config: core.CollectorConfig, pending: Mapping[str, Any], slot: Mapping[str, Any], payload: bytes, ) -> None: """Replace only this run's exact STARTED slot; never overwrite third content.""" started = core.canonical_json_bytes(_started_slot(config, pending), newline=False) path = _slot_path(config, int(slot["hour_epoch"])) if not path.exists(): raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "STARTED slot disappeared before terminal commit.", safety=True) core.lexical_lstat_chain(path, allow_missing_leaf=False) if not path.is_file(): raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Run slot is not a regular file.", safety=True) current = path.read_bytes() if current == payload: return if current != started: raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Run slot contains third content.", safety=True) _write_readback(path, payload, "terminal run slot") def _terminal_slot_from_planned( config: core.CollectorConfig, pending: Mapping[str, Any], planned: Mapping[str, Any], ) -> tuple[dict[str, Any], bytes]: required = { "terminal_at", "status", "error_code", "exit_code", "coverage_proof", "state_manifest", "formal_manifest", "artifact_tree_sha256", "refresh_action", "refresh_count", "page_authoritative", "coverage_complete", "evidence_sha256", "input_item_count", "new_item_count", "saved_artifact_count", "formal_manifest_changed", "transaction_receipt", } if set(planned) != required: raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Planned terminal schema is invalid.", safety=True) core.parse_datetime(planned["terminal_at"], "planned_terminal.terminal_at") if not isinstance(planned["transaction_receipt"], Mapping): raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Planned terminal receipt is invalid.", safety=True) if planned["transaction_receipt"] != _receipt_for_terminal( pending, planned["state_manifest"], planned["formal_manifest"] ): raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Planned terminal receipt identity drifted.", safety=True) if planned["state_manifest"] != _identity(config.manifest_path).as_dict() or planned["formal_manifest"] != _identity(config.refresh.formal_manifest).as_dict(): raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Planned terminal manifest identity drifted.", safety=True) if planned["formal_manifest_changed"] != ( planned["transaction_receipt"].get("business_commit_kind") == "DUAL_MANIFEST_COMMIT" ): raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Planned terminal business identity drifted.", safety=True) slot = _started_slot(config, pending) slot.update({ "run_state": "TERMINAL", "terminal_at": planned["terminal_at"], "status": planned["status"], "error_code": planned["error_code"], "exit_code": planned["exit_code"], "refresh_action": planned["refresh_action"], "refresh_count": planned["refresh_count"], "page_authoritative": planned["page_authoritative"], "coverage_complete": planned["coverage_complete"], "coverage_proof": planned["coverage_proof"], "evidence_sha256": planned["evidence_sha256"], "input_item_count": planned["input_item_count"], "new_item_count": planned["new_item_count"], "saved_artifact_count": planned["saved_artifact_count"], "state_manifest": planned["state_manifest"], "formal_manifest": planned["formal_manifest"], "artifact_tree_sha256": planned["artifact_tree_sha256"], "transaction_receipt": planned["transaction_receipt"], }) return slot, core.canonical_json_bytes(slot, newline=False) def _ensure_latest(config: core.CollectorConfig, slot: Mapping[str, Any], slot_payload: bytes) -> None: latest = _latest_index(slot, slot_payload) payload = core.canonical_json_bytes(latest, newline=False) path = _runs_dir(config) / "latest.json" if path.exists(): current, current_payload = _load_json_file(path, "latest run index") if current_payload == payload: return current_hour = current.get("hour_epoch") if not isinstance(current_hour, int) or current_hour >= latest["hour_epoch"]: raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Latest run index is conflicting or newer.", safety=True) _write_readback(path, payload, "latest run index") def _result_from_slot(config: core.CollectorConfig, slot: Mapping[str, Any]) -> dict[str, Any]: receipt = slot["transaction_receipt"] return { "status": slot["status"], "error_code": slot["error_code"], "exit_code": slot["exit_code"], "run_id": slot["run_id"], "creator_uid": slot["creator_uid"], "refresh_count": slot["refresh_count"], "page_authoritative": slot["page_authoritative"], "coverage_complete": slot["coverage_complete"], "new_items": slot["new_item_count"], "saved_artifacts": slot["saved_artifact_count"], "formal_manifest_changed": receipt["business_commit_kind"] == "DUAL_MANIFEST_COMMIT", "run_evidence_path": str(_slot_path(config, slot["hour_epoch"])), "no_new_confirmed": slot["status"] == "REFRESH_CONFIRMED_NO_NEW", "warnings": list(slot["warnings"]), } def _terminal( config: core.CollectorConfig, pending: dict[str, Any], *, now: datetime, status: str, error_code: str | None, exit_code: int, evidence: Mapping[str, Any] | None, evidence_hash: str | None, coverage: Mapping[str, Any] | None, input_count: int, new_count: int, created: Sequence[Mapping[str, Any]], formal_changed: bool, ) -> dict[str, Any]: terminal_at = core.canonical_datetime(now) transaction = pending.get("transaction_identity") state_id = _identity(config.manifest_path).as_dict() formal_id = _identity(config.refresh.formal_manifest).as_dict() artifact_tree = hashlib.sha256(core.canonical_json_bytes(list(created), newline=False)).hexdigest() receipt = _receipt_for_terminal(pending, state_id, formal_id) planned = pending.get("planned_terminal") if planned is None: planned = { "terminal_at": terminal_at, "status": status, "error_code": error_code, "exit_code": exit_code, "coverage_proof": coverage["proof"] if coverage else None, "state_manifest": state_id, "formal_manifest": formal_id, "artifact_tree_sha256": artifact_tree, "refresh_action": evidence.get("refresh_action") if evidence else None, "refresh_count": evidence.get("refresh_count", 0) if evidence else 0, "page_authoritative": bool(evidence and evidence.get("page_outcome") == "READABLE"), "coverage_complete": bool(coverage and coverage["coverage_complete"]), "evidence_sha256": evidence_hash, "input_item_count": input_count, "new_item_count": new_count, "saved_artifact_count": len(created), "formal_manifest_changed": formal_changed, "transaction_receipt": receipt, } pending["planned_terminal"] = planned _write_pending(config, pending) slot, slot_payload = _terminal_slot_from_planned(config, pending, planned) _write_terminal_slot(config, pending, slot, slot_payload) _ensure_latest(config, slot, slot_payload) pending["phase"] = "TERMINAL_RECORDED" pending["last_transition_at"] = planned["terminal_at"] _write_pending(config, pending) warnings: list[str] = [] try: _release_formal_lock(config, pending) _pending_path(config).unlink() except (OSError, core.CollectorError): warnings.append("W_PENDING_CLEANUP") result = _result_from_slot(config, slot) result["warnings"] = warnings return result def _recover_or_replay( config: core.CollectorConfig, config_path: Path, now: datetime, *, allow_final_evidence: bool = False, ) -> dict[str, Any] | None: pending = _load_pending(config) if pending is None: return None if pending["config_sha256"] != _canonical_config_hash(config_path): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Config changed during pending run.", safety=True) phase = pending["phase"] if phase == "TERMINAL_RECORDED": planned = pending.get("planned_terminal") if not isinstance(planned, Mapping): raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Recorded terminal plan is missing.", safety=True) slot, expected_payload = _terminal_slot_from_planned(config, pending, planned) slot_path = _slot_path(config, int(slot["hour_epoch"])) core.lexical_lstat_chain(slot_path, allow_missing_leaf=False) slot_payload = slot_path.read_bytes() if slot_payload != expected_payload: raise core.CollectorError("E_RUN_EVIDENCE_COMMIT", "Terminal slot does not match the frozen plan.", safety=True) _ensure_latest(config, slot, slot_payload) warnings: list[str] = [] try: _release_formal_lock(config, pending) _pending_path(config).unlink() except (OSError, core.CollectorError): warnings.append("W_PENDING_CLEANUP") result = _result_from_slot(config, slot) result["warnings"] = warnings return result if phase == "AWAITING_EVIDENCE": repaired_started = _ensure_started_slot(config, pending) final = Path(pending["evidence_path"]) if pending["schema_version"] == LEGACY_PENDING_SCHEMA: if final.exists(): raise core.CollectorError( "E_LEGACY_RECOVERY_ONLY", "Legacy browser evidence cannot enter the runtime-v2 commit path.", safety=True, ) if now <= core.parse_datetime(pending["deadline_at"], "deadline_at"): raise core.CollectorError( "E_LEGACY_RECOVERY_ONLY", "Legacy pending is recovery-only and cannot request another browser action.", safety=True, ) return _terminal( config, pending, now=now, status="REFRESH_FAILED_PAGE_UNREADABLE", error_code="E_LEGACY_EVIDENCE_MISSING_AFTER_DEADLINE", exit_code=4, evidence=None, evidence_hash=None, coverage=None, input_count=0, new_count=0, created=[], formal_changed=False, ) if final.exists(): if allow_final_evidence: return None raise core.CollectorError( "E_CONTROLLER_REQUIRED", "Unbound runtime-v2 evidence cannot enter refresh-commit.", safety=True, ) if repaired_started: return _begin_result(config, pending) if now <= core.parse_datetime(pending["deadline_at"], "deadline_at"): raise core.CollectorError("E_BUSY", "The current refresh is still awaiting evidence.", safety=True) return _terminal( config, pending, now=now, status="REFRESH_FAILED_PAGE_UNREADABLE", error_code="E_EVIDENCE_MISSING_AFTER_DEADLINE", exit_code=4, evidence=None, evidence_hash=None, coverage=None, input_count=0, new_count=0, created=[], formal_changed=False, ) if phase == "EVIDENCE_BOUND": evidence_path = Path(pending["evidence_path"]) if not evidence_path.exists(): raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Bound evidence disappeared.", safety=True) payload = evidence_path.read_bytes() identity = pending.get("evidence_identity") if not isinstance(identity, Mapping) or identity != {"bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest()}: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Bound evidence identity drifted.", safety=True) planned = pending.get("planned_terminal") if isinstance(planned, Mapping): evidence = _strict_json_bytes(payload, "bound browser evidence") coverage = None if planned.get("coverage_proof") is not None: coverage = { "coverage_complete": bool(planned["coverage_complete"]), "proof": planned["coverage_proof"], } receipt = planned.get("transaction_receipt") or {} return _terminal( config, pending, now=now, status=str(planned["status"]), error_code=planned.get("error_code"), exit_code=int(planned["exit_code"]), evidence=evidence, evidence_hash=identity["sha256"], coverage=coverage, input_count=int(planned["input_item_count"]), new_count=int(planned["new_item_count"]), created=list(receipt.get("created_artifacts", [])), formal_changed=bool(planned.get("formal_manifest_changed")), ) if allow_final_evidence: return None raise core.CollectorError("E_EVIDENCE_READY", "Bound evidence is ready for refresh-commit.", safety=True) if phase in {"TRANSACTION_INTENT", "BUSINESS_COMMITTED"}: txn = pending["transaction_identity"] _recovery_formal_lock(config, pending, now) state_now = _identity(config.manifest_path).as_dict() formal_now = _identity(config.refresh.formal_manifest).as_dict() if state_now == txn["state_candidate"] and formal_now == txn["formal_candidate"]: pending["phase"] = "BUSINESS_COMMITTED" _write_pending(config, pending) return _terminal( config, pending, now=now, status="NEW_ITEMS_SAVED", error_code=None, exit_code=0, evidence=None, evidence_hash=pending["evidence_identity"]["sha256"], coverage=None, input_count=txn["input_item_count"], new_count=txn["new_item_count"], created=txn["created_artifacts"], formal_changed=True, ) if formal_now == txn["formal_preimage"] and state_now in (txn["state_preimage"], txn["state_candidate"]): if state_now == txn["state_candidate"]: try: pre = bytes.fromhex(txn["state_preimage_hex"]) except (KeyError, ValueError, TypeError) as exc: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "State preimage bytes are not retained for rollback.", safety=True) from exc if hashlib.sha256(pre).hexdigest() != txn["state_preimage"]["sha256"]: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "State rollback preimage hash drifted.", safety=True) _replace_manifest(config.manifest_path, pre) for artifact in txn["created_artifacts"]: target = config.refresh.archive_dir / artifact["path"] if target.exists() and target.is_file() and target.stat().st_size == artifact["bytes"] and core.sha256_file(target) == artifact["sha256"]: target.unlink() return _terminal( config, pending, now=now, status="PARTIAL_DISCOVERY_UNCONFIRMED", error_code="E_TRANSACTION_ROLLED_BACK", exit_code=4, evidence=None, evidence_hash=pending["evidence_identity"]["sha256"], coverage=None, input_count=txn["input_item_count"], new_count=0, created=[], formal_changed=False, ) raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Incomplete content transaction needs exact offline recovery.", safety=True) return None def refresh_commit(config: core.CollectorConfig, config_path: Path, evidence_path: Path, now: datetime) -> dict[str, Any]: refresh = config.refresh assert refresh is not None _validate_refresh_roots(config, create_state=False) pending = _load_pending(config) if pending is None: raise core.CollectorError("E_NO_PENDING", "refresh-commit requires refresh-begin first.") recovered = _recover_or_replay(config, config_path, now, allow_final_evidence=True) if recovered is not None: return recovered if now > core.parse_datetime(pending["deadline_at"], "deadline_at") and pending.get("phase") != "EVIDENCE_BOUND": return _terminal( config, pending, now=now, status="REFRESH_FAILED_PAGE_UNREADABLE", error_code="E_OVERALL_DEADLINE", exit_code=4, evidence=None, evidence_hash=None, coverage=None, input_count=0, new_count=0, created=[], formal_changed=False, ) if pending.get("phase") != "EVIDENCE_BOUND": raise core.CollectorError( "E_CONTROLLER_REQUIRED", "Schema 3 evidence is accepted only after the trusted refresh-run controller binds it.", safety=True, ) evidence, payload, observation, identity_match = _validate_evidence(config, pending, evidence_path) evidence_hash = hashlib.sha256(payload).hexdigest() if pending.get("evidence_identity") != {"bytes": len(payload), "sha256": evidence_hash}: raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Bound evidence identity drifted.", safety=True) action_outcome = evidence["runtime_observation"]["refresh_action_outcome"] if evidence["page_outcome"] == "ACCESS_BLOCKED" or not identity_match: error_code = "E_ACCESS_BLOCKED" if evidence["page_outcome"] == "ACCESS_BLOCKED" else "E_CREATOR_MISMATCH" return _terminal(config, pending, now=now, status="REFRESH_BLOCKED_AUTH_OR_ACCESS", error_code=error_code, exit_code=3, evidence=evidence, evidence_hash=evidence_hash, coverage=None, input_count=len(evidence["items"]), new_count=0, created=[], formal_changed=False) if evidence["page_outcome"] != "READABLE": return _terminal(config, pending, now=now, status="REFRESH_FAILED_PAGE_UNREADABLE", error_code="E_PAGE_UNREADABLE", exit_code=4, evidence=evidence, evidence_hash=evidence_hash, coverage=None, input_count=len(evidence["items"]), new_count=0, created=[], formal_changed=False) effective_observation = observation if observation is not None and action_outcome != "CONFIRMED": effective_observation = dict(observation) effective_observation["coverage_complete"] = False _, formal_tokens, _ = load_formal_catalog(config) planned = _plan_content(config, pending, evidence, formal_tokens, now) if not planned: status = "REFRESH_CONFIRMED_NO_NEW" if action_outcome == "CONFIRMED" and effective_observation and effective_observation["coverage_complete"] else "PARTIAL_DISCOVERY_UNCONFIRMED" code = 0 if status == "REFRESH_CONFIRMED_NO_NEW" else 4 error = None if code == 0 else "E_COVERAGE_INCOMPLETE" return _terminal(config, pending, now=now, status=status, error_code=error, exit_code=code, evidence=evidence, evidence_hash=evidence_hash, coverage=effective_observation, input_count=len(evidence["items"]), new_count=0, created=[], formal_changed=False) state_pre = config.manifest_path.read_bytes() if config.manifest_path.exists() else b"" formal_pre = refresh.formal_manifest.read_bytes() if refresh.formal_manifest.exists() else b"" if state_pre and not state_pre.endswith(b"\n") or formal_pre and not formal_pre.endswith(b"\n"): raise core.CollectorError("E_MANIFEST", "Manifest preimage must end in LF.", safety=True) state_lines, formal_lines, created = _event_lines(config, planned, pending["run_id"], now, bool(effective_observation and effective_observation["coverage_complete"])) state_candidate = state_pre + state_lines formal_candidate = formal_pre + formal_lines txn = { "transaction_id": uuid.uuid4().hex, "state_preimage": FileIdentity(config.manifest_path.exists(), len(state_pre), hashlib.sha256(state_pre).hexdigest()).as_dict(), "state_preimage_hex": state_pre.hex(), "formal_preimage": FileIdentity(refresh.formal_manifest.exists(), len(formal_pre), hashlib.sha256(formal_pre).hexdigest()).as_dict(), "state_candidate": FileIdentity(True, len(state_candidate), hashlib.sha256(state_candidate).hexdigest()).as_dict(), "formal_candidate": FileIdentity(True, len(formal_candidate), hashlib.sha256(formal_candidate).hexdigest()).as_dict(), "created_artifacts": created, "input_item_count": len(evidence["items"]), "new_item_count": len(planned), "formal_lock_claim": None, "recovery_count": 0, } pending["transaction_identity"] = txn pending["phase"] = "TRANSACTION_INTENT" _write_pending(config, pending) _acquire_formal_lock(config, pending, now) if _identity(config.manifest_path).as_dict() != txn["state_preimage"] or _identity(refresh.formal_manifest).as_dict() != txn["formal_preimage"]: raise core.CollectorError("E_MANIFEST_RACE_REBEGIN", "Manifest changed before transaction commit.", safety=True) published: list[Path] = [] try: for row in planned: for artifact in row["artifacts"]: target = refresh.archive_dir / artifact["target"] _create_new(target, artifact["payload"]) published.append(target) if _identity(config.manifest_path).as_dict() != txn["state_preimage"] or _identity(refresh.formal_manifest).as_dict() != txn["formal_preimage"]: raise core.CollectorError("E_MANIFEST_RACE_REBEGIN", "Manifest changed during artifact staging.", safety=True) core.atomic_replace_bytes(config.manifest_path, state_candidate) core.atomic_replace_bytes(refresh.formal_manifest, formal_candidate) pending["phase"] = "BUSINESS_COMMITTED" _write_pending(config, pending) except BaseException: if _identity(refresh.formal_manifest).as_dict() == txn["formal_preimage"]: if _identity(config.manifest_path).as_dict() == txn["state_candidate"]: _replace_manifest(config.manifest_path, state_pre) for target in reversed(published): try: if target.is_file() and any(target.name == item["path"] and core.sha256_file(target) == item["sha256"] for item in created): target.unlink() except OSError: pass raise return _terminal(config, pending, now=now, status="NEW_ITEMS_SAVED", error_code=None, exit_code=0, evidence=evidence, evidence_hash=evidence_hash, coverage=effective_observation, input_count=len(evidence["items"]), new_count=len(planned), created=created, formal_changed=True)