#!/usr/bin/env python3
|
"""Trusted, injectable controller for one supported Chrome refresh run.
|
|
Production orchestration supplies a reviewed adapter for the supported Chrome
|
calls. Tests use a deterministic fake implementing the same four methods. This
|
module owns dispatch state, monotonic timing, extraction adaptation, attestation,
|
and the no-overwrite evidence commit; callers never select runtime outcomes.
|
"""
|
from __future__ import annotations
|
|
import hashlib
|
import json
|
import time
|
from datetime import datetime, timedelta, timezone
|
from pathlib import Path
|
from typing import Any, Callable, Mapping, Protocol
|
|
import bili_dynamic_collector as core
|
import bili_dynamic_refresh as refresh
|
|
|
class SupportedChromeAdapter(Protocol):
|
def open_tabs(self) -> list[Mapping[str, Any]]: ...
|
def reload(self, tab: Mapping[str, Any], timeout_seconds: int) -> None: ...
|
def goto(self, url: str, timeout_seconds: int) -> Mapping[str, Any]: ...
|
def evaluate(self, tab: Mapping[str, Any], source: str, timeout_seconds: int) -> Mapping[str, Any]: ...
|
|
|
def _millis(clock: Callable[[], float]) -> int:
|
return int(clock() * 1000)
|
|
|
def _sha_material(value: str) -> str:
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
def _adapt_extract(config: core.CollectorConfig, pending: Mapping[str, Any], raw: Mapping[str, Any], observed_at: str) -> tuple[dict[str, Any], list[dict[str, Any]], dict[str, Any] | None]:
|
cards: list[dict[str, Any]] = []
|
items: list[dict[str, Any]] = []
|
positions: list[tuple[int, str, Mapping[str, Any]]] = []
|
for entry in raw.get("cards", []):
|
positions.append((int(entry["position"]), "card", entry))
|
for entry in raw.get("unparsed_nodes", []):
|
positions.append((int(entry["position"]), "unparsed", entry))
|
positions.sort()
|
remap = {old: index for index, (old, _, _) in enumerate(positions)}
|
for old_position, kind, entry in positions:
|
if kind != "card":
|
continue
|
normalized = core.normalize_item(entry, config, len(cards))
|
cards.append({
|
"position": remap[old_position],
|
"identifiers": {"dynamic_id": normalized["dynamic_id"], "opus_id": normalized["opus_id"], "bvid": normalized["bvid"]},
|
"stable_keys": normalized["dedupe_keys"], "published_at": normalized["published_at"],
|
"content_type": normalized["content_type"], "source_url": normalized["source_url"],
|
})
|
items.append({
|
**{key: normalized[key] for key in ("dynamic_id", "opus_id", "bvid", "content_type", "published_at", "title", "source_url")},
|
"body_text": entry.get("body_text") or "", "body_complete": bool(entry.get("body_complete")),
|
"duration_seconds": entry.get("duration_seconds"), "artifacts": [],
|
})
|
unparsed = [{"position": remap[old], "node_fingerprint_sha256": _sha_material(str(entry["fingerprint_material"])), "reason_code": entry["reason_code"]} for old, kind, entry in positions if kind == "unparsed"]
|
observation = {"ordinal": 0, "observed_at": observed_at, "cursor_before": 0, "cursor_after": len(positions), "visible_node_count": len(positions), "complete_card_count": len(cards), "unparsed_node_count": len(unparsed), "cards": cards, "unparsed_nodes": unparsed, "limit_hit": raw.get("limit_hit", "NONE")}
|
marker = None
|
if raw.get("terminal_marker_text"):
|
text = str(raw["terminal_marker_text"])
|
marker = {"observation_ordinal": 0, "kind": "EXACT_END_OF_FEED", "selector_id": "DYNAMIC_FEED_END_TEXT", "normalized_text": text, "marker_sha256": hashlib.sha256(f"DYNAMIC_FEED_END_TEXT\n{text}".encode()).hexdigest()}
|
return observation, items, marker
|
|
|
def _commit_create_new(path: Path, evidence: Mapping[str, Any]) -> None:
|
payload = core.canonical_json_bytes(evidence, newline=False)
|
core.ensure_directory(path.parent, create=True)
|
partial = path.parent / f".{path.name}.{hashlib.sha256(payload).hexdigest()[:16]}.partial"
|
core.lexical_lstat_chain(partial, allow_missing_leaf=True)
|
try:
|
with partial.open("xb") as stream:
|
stream.write(payload); stream.flush(); __import__("os").fsync(stream.fileno())
|
refresh._rename_no_overwrite(partial, path)
|
refresh._fsync_directory(path.parent)
|
if path.read_bytes() != payload:
|
raise core.CollectorError("E_EVIDENCE_COMMIT", "Controller evidence durable readback mismatch.", safety=True)
|
finally:
|
try: partial.unlink()
|
except FileNotFoundError: pass
|
|
|
def _discard_exact_created(path: Path, evidence: Mapping[str, Any]) -> None:
|
"""Remove only the exact evidence created by this controller invocation."""
|
payload = core.canonical_json_bytes(evidence, newline=False)
|
core.lexical_lstat_chain(path, allow_missing_leaf=False)
|
if not path.is_file() or path.read_bytes() != payload:
|
raise core.CollectorError(
|
"E_EVIDENCE_COMMIT",
|
"Late controller evidence no longer matches the owned payload.",
|
safety=True,
|
)
|
path.unlink()
|
refresh._fsync_directory(path.parent)
|
if path.exists():
|
raise core.CollectorError(
|
"E_EVIDENCE_COMMIT",
|
"Late controller evidence cleanup did not persist.",
|
safety=True,
|
)
|
|
|
def run_once(config: core.CollectorConfig, begin: Mapping[str, Any], adapter: SupportedChromeAdapter, *, monotonic: Callable[[], float] = time.monotonic, wall_now: Callable[[], datetime] = lambda: datetime.now(timezone.utc)) -> Path:
|
pending = refresh._load_pending(config)
|
if not pending or pending["run_id"] != begin.get("run_id") or pending["schema_version"] != refresh.PENDING_SCHEMA:
|
raise core.CollectorError("E_CONTROLLER_PENDING", "Controller requires the exact active runtime-v2 pending.", safety=True)
|
run_ms = _millis(monotonic); deadline_ms = run_ms + config.refresh.overall_deadline_seconds * 1000
|
action_start = action_finish = observation_start = observation_finish = None
|
action_dispatched = False; action = None; action_outcome = "PRE_DISPATCH_ERROR"; observation_outcome = "NOT_ATTEMPTED"; raw = None
|
try:
|
tabs = [tab for tab in adapter.open_tabs() if tab.get("url") == config.creator_dynamic_url]
|
except Exception:
|
tabs = None
|
if tabs is None or len(tabs) > 1:
|
tab = None
|
else:
|
try:
|
action_start = _millis(monotonic); action_dispatched = True
|
if tabs: action = "reload"; tab = tabs[0]; adapter.reload(tab, config.refresh.refresh_action_timeout_seconds)
|
else: action = "navigate"; tab = adapter.goto(config.creator_dynamic_url, config.refresh.refresh_action_timeout_seconds)
|
action_outcome = "CONFIRMED"
|
except TimeoutError:
|
action_outcome = "TIMEOUT"; tab = tabs[0] if tabs else {"url": config.creator_dynamic_url}
|
except Exception:
|
action_outcome = "POST_DISPATCH_ERROR"; tab = tabs[0] if tabs else {"url": config.creator_dynamic_url}
|
finally:
|
if action_dispatched: action_finish = _millis(monotonic)
|
if action_dispatched and action_finish is not None and action_finish < deadline_ms:
|
try:
|
observation_start = _millis(monotonic)
|
extractor_call = (
|
refresh.EXTRACTOR_SOURCE.read_text(encoding="utf-8")
|
+ "\nprojectInfoCollectVisibleDynamicNodes({page_internal_settle_timeout_ms:15000})"
|
)
|
raw = adapter.evaluate(tab, extractor_call, min(config.refresh.observation_timeout_seconds, max(0, (deadline_ms - observation_start) // 1000)))
|
observation_outcome = "READABLE"
|
except TimeoutError: observation_outcome = "TIMEOUT"
|
except PermissionError: observation_outcome = "ACCESS_BLOCKED"
|
except Exception: observation_outcome = "ERROR"
|
finally: observation_finish = _millis(monotonic)
|
elif action_dispatched:
|
observation_outcome = "DEADLINE_EXHAUSTED"
|
write_ms = _millis(monotonic)
|
if write_ms > deadline_ms:
|
raise core.CollectorError("E_OVERALL_DEADLINE", "Runtime controller exceeded its total deadline.", safety=True)
|
wall_started = core.parse_datetime(pending["started_at"], "pending.started_at")
|
action_wall_ms = 0 if action_start is None else action_finish - action_start
|
observation_wall_ms = 0 if observation_start is None else observation_finish - observation_start
|
refresh_finished_at = wall_started + timedelta(milliseconds=action_wall_ms)
|
read_finished_at = refresh_finished_at + timedelta(milliseconds=observation_wall_ms)
|
observed_at = core.canonical_datetime(read_finished_at)
|
observation = items = marker = None
|
if observation_outcome == "READABLE" and raw is not None:
|
observation, items, marker = _adapt_extract(config, pending, raw, observed_at)
|
page_outcome = {"READABLE":"READABLE", "TIMEOUT":"UNREADABLE_TIMEOUT", "ERROR":"UNREADABLE_ERROR", "ACCESS_BLOCKED":"ACCESS_BLOCKED", "NOT_ATTEMPTED":"UNREADABLE_ERROR", "DEADLINE_EXHAUSTED":"UNREADABLE_TIMEOUT"}[observation_outcome]
|
diagnostics = {"PRE_DISPATCH_ERROR":"ACTION_PRE_DISPATCH", "POST_DISPATCH_ERROR":"ACTION_POST_DISPATCH", "TIMEOUT":"ACTION_TIMEOUT"}.get(action_outcome) or {"READABLE":"NONE", "TIMEOUT":"OBSERVATION_TIMEOUT", "ERROR":"OBSERVATION_ERROR", "ACCESS_BLOCKED":"ACCESS_INTERSTITIAL", "DEADLINE_EXHAUSTED":"DEADLINE_EXHAUSTED"}.get(observation_outcome, "ACTION_PRE_DISPATCH")
|
contract_raw = refresh.OBSERVATION_CONTRACT.read_bytes(); runtime_raw = refresh.RUNTIME_CONTRACT.read_bytes()
|
creator = raw.get("creator", {}) if raw else {"uid": None, "name": "", "profile_url": None}
|
evidence = {"schema_version": 3, "run_id": pending["run_id"], "transport": "codex_chrome_visible_page", "requested_url": config.creator_dynamic_url, "final_url": raw.get("final_url", config.creator_dynamic_url) if raw else config.creator_dynamic_url, "refresh_action": action, "refresh_count": 1 if action_dispatched else 0, "refresh_started_at": pending["started_at"], "refresh_finished_at": core.canonical_datetime(refresh_finished_at), "read_finished_at": observed_at, "page_outcome": page_outcome, "page_title": raw.get("page_title", "") if raw else "", "creator": creator, "extractor": {"contract_id": json.loads(contract_raw)["contract_id"], "contract_sha256": hashlib.sha256(contract_raw).hexdigest(), "parser_version": json.loads(contract_raw)["parser_version"], "parser_sha256": hashlib.sha256(refresh.EXTRACTOR_SOURCE.read_bytes()).hexdigest()}, "page_observation": None if observation is None else {"schema_version": 1, "limits": json.loads(contract_raw)["limits"], "observations": [observation], "terminal_marker": marker}, "items": items or [], "discovery_summary": {"status": "NOT_USED", "item_count": 0}, "safe_diagnostics": {"code": diagnostics, "overall_deadline_seconds": 120, "refresh_action_timeout_seconds": 35, "observation_timeout_seconds": 45}, "runtime_contract": {"contract_id": pending["runtime_contract"]["contract_id"], "contract_bytes": len(runtime_raw), "contract_sha256": hashlib.sha256(runtime_raw).hexdigest()}, "runtime_observation": {"refresh_action_outcome": action_outcome, "refresh_action_elapsed_ms": action_wall_ms, "refresh_count": 1 if action_dispatched else 0, "observation_outcome": observation_outcome, "observation_elapsed_ms": observation_wall_ms, "observation_count": 0 if observation_start is None else 1}}
|
refresh._attest_controller_evidence(pending, evidence, action_dispatched=action_dispatched, monotonic_run_started_ms=run_ms, monotonic_action_started_ms=action_start, monotonic_action_finished_ms=action_finish, monotonic_observation_started_ms=observation_start, monotonic_observation_finished_ms=observation_finish, monotonic_evidence_write_started_ms=write_ms)
|
path = Path(pending["evidence_path"])
|
_commit_create_new(path, evidence)
|
committed_ms = _millis(monotonic)
|
transitioned_at = wall_now()
|
wall_deadline = core.parse_datetime(pending["deadline_at"], "pending.deadline_at")
|
if committed_ms > deadline_ms or transitioned_at > wall_deadline:
|
_discard_exact_created(path, evidence)
|
raise core.CollectorError(
|
"E_OVERALL_DEADLINE",
|
"Controller evidence did not durably complete within the total deadline.",
|
safety=True,
|
)
|
refresh.bind_controller_evidence(config, pending, path, transitioned_at=transitioned_at)
|
return path
|
