#!/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 secrets
|
import uuid
|
from dataclasses import dataclass
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path, PurePosixPath
|
from typing import Any, Iterable, Mapping, Sequence
|
|
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 _attest_controller_evidence(
|
pending: Mapping[str, Any],
|
evidence: dict[str, Any],
|
*,
|
action_dispatched: bool,
|
monotonic_run_started_ms: int,
|
monotonic_action_started_ms: int | None,
|
monotonic_action_finished_ms: int | None,
|
monotonic_observation_started_ms: int | None,
|
monotonic_observation_finished_ms: int | None,
|
monotonic_evidence_write_started_ms: int,
|
) -> dict[str, Any]:
|
"""Bind evidence to the source-controlled controller's actual call envelope.
|
|
The per-run capability is durable state that is never returned by
|
``refresh-begin``. It is not browser/session material; it only prevents a
|
caller-authored JSON document from selecting a successful runtime state.
|
"""
|
capability = pending.get("controller_capability")
|
if not isinstance(capability, str) or re.fullmatch(r"[0-9a-f]{64}", capability) is None:
|
raise core.CollectorError("E_RECOVERY_AMBIGUOUS", "Controller capability is invalid.", safety=True)
|
runtime = pending["runtime_contract"]
|
evidence["controller_attestation"] = {
|
"controller_id": runtime["controller_id"],
|
"controller_sha256": runtime["controller_sha256"],
|
"binding_algorithm": runtime["binding_algorithm"],
|
"action_dispatched": action_dispatched,
|
"monotonic_run_started_ms": monotonic_run_started_ms,
|
"monotonic_action_started_ms": monotonic_action_started_ms,
|
"monotonic_action_finished_ms": monotonic_action_finished_ms,
|
"monotonic_observation_started_ms": monotonic_observation_started_ms,
|
"monotonic_observation_finished_ms": monotonic_observation_finished_ms,
|
"monotonic_evidence_write_started_ms": monotonic_evidence_write_started_ms,
|
"binding_sha256": None,
|
}
|
digest = hmac.new(
|
bytes.fromhex(capability), _controller_attestation_payload(evidence), hashlib.sha256
|
).hexdigest()
|
evidence["controller_attestation"]["binding_sha256"] = digest
|
return evidence
|
|
|
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_capability",
|
}
|
schema = value.get("schema_version")
|
if schema == LEGACY_PENDING_SCHEMA:
|
legacy_required = required - {"runtime_contract", "controller_capability"}
|
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) -> 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
|
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_capability": secrets.token_hex(32),
|
}
|
_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,
|
) -> 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)
|
capability = pending["controller_capability"]
|
expected_binding = hmac.new(
|
bytes.fromhex(capability), _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)
|
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 bind_controller_evidence(
|
config: core.CollectorConfig,
|
pending: dict[str, Any],
|
path: Path,
|
*,
|
transitioned_at: datetime,
|
) -> None:
|
"""Validate and durably bind the controller-created evidence exactly once."""
|
if pending.get("schema_version") != PENDING_SCHEMA or pending.get("phase") != "AWAITING_EVIDENCE":
|
raise core.CollectorError("E_CONTROLLER_STATE", "Controller evidence can only bind the active runtime-v2 run.", safety=True)
|
deadline = core.parse_datetime(pending["deadline_at"], "pending.deadline_at")
|
if transitioned_at > deadline:
|
raise core.CollectorError(
|
"E_OVERALL_DEADLINE",
|
"Controller evidence cannot be bound after the total deadline.",
|
safety=True,
|
)
|
_, payload, _, _ = _validate_evidence(config, pending, path)
|
pending["evidence_identity"] = {"bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest()}
|
pending["phase"] = "EVIDENCE_BOUND"
|
pending["last_transition_at"] = core.canonical_datetime(transitioned_at)
|
_write_pending(config, pending)
|
if _load_pending(config) != pending:
|
raise core.CollectorError("E_CONTROLLER_STATE", "Controller evidence binding readback mismatch.", safety=True)
|
|
|
def _formal_tokens(event: Mapping[str, Any], config: core.CollectorConfig) -> list[str]:
|
stable = event.get("stable_id")
|
tokens: list[str] = []
|
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:
|
raise core.CollectorError("E_CATALOG_IDENTITY_CONFLICT", "Formal stable_id is invalid.", safety=True)
|
source = event.get("source_url")
|
if isinstance(source, str):
|
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,
|
)
|
evidence, payload, observation, identity_match = _validate_evidence(config, pending, evidence_path)
|
evidence_hash = hashlib.sha256(payload).hexdigest()
|
if pending.get("phase") != "EVIDENCE_BOUND":
|
raise core.CollectorError("E_CONTROLLER_REQUIRED", "Evidence must be durably bound by the trusted controller.", safety=True)
|
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)
|
