from __future__ import annotations import hashlib import hmac import importlib.util import io import json import gc import os import shutil import secrets import subprocess import sys import tempfile import time import unittest from pathlib import Path PROJECT_DEV = Path(__file__).resolve().parents[1] if str(PROJECT_DEV) not in sys.path: sys.path.insert(0, str(PROJECT_DEV)) COLLECTOR_PATH = PROJECT_DEV / "bili_dynamic_collector.py" SPEC = importlib.util.spec_from_file_location("bili_dynamic_collector", COLLECTOR_PATH) assert SPEC and SPEC.loader collector = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = collector SPEC.loader.exec_module(collector) import bili_dynamic_refresh as refresh import bili_dynamic_refresh_controller as controller NOW = "2026-08-13T10:00:00+08:00" class BiliDynamicRefreshTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.state = self.root / "state" self.archive = self.root / "archive" self.intake = self.root / "intake" self.downloads = self.root / "downloads" self.videos = self.root / "videos" for path in (self.archive, self.intake, self.downloads, self.videos): path.mkdir() self.config_path = self.root / "config.json" self.controller_keys: dict[str, bytes] = {} self.write_json( self.config_path, { "schema_version": 1, "creator": { "name": "青枫浦上Q", "uid": "1420210197", "dynamic_url": "https://space.bilibili.com/1420210197/dynamic", }, "timezone": "Asia/Shanghai", "window_hours": 72, "minimum_complete_age_seconds": 0, "title_max_length": 48, "allowed_source_hosts": [ "space.bilibili.com", "www.bilibili.com", "t.bilibili.com", "b23.tv" ], "allowed_video_extensions": [".mp4", ".mkv", ".mov", ".webm"], "native_handoff": { "project_id": "project-info", "source_ai_id": "video-downloader", "source_thread_id": "019fcc5d-798f-7ea1-8325-3a4d1f2dc5a5", "source_role_instance_id": "case_analysis.video_downloader", "target_ai_id": "media-processor", "target_thread_id": "019fb7a4-bdfd-79f2-bd6b-e67e2b7d8efd", "target_role_instance_id": "case_analysis.media_processor", "reply_thread_id": "019fcc5d-798f-7ea1-8325-3a4d1f2dc5a5", }, "paths": {"state_dir": "state", "download_dir": "downloads", "video_dir": "videos"}, "refresh": { "archive_dir": "archive", "formal_manifest": "archive/manifest.jsonl", "intake_dir": "intake", "overall_deadline_seconds": 120, "refresh_action_timeout_seconds": 35, "observation_timeout_seconds": 45, "page_internal_settle_timeout_seconds": 15, "max_refresh_count": 1, "run_history_slots": 168, "max_items": 200, "max_images_per_item": 20, "max_image_bytes": 20971520, "max_text_bytes": 2097152, }, }, ) def tearDown(self) -> None: self.temp.cleanup() def write_json(self, path: Path, value: object) -> None: if isinstance(value, dict) and value.get("schema_version") == 3 and isinstance(value.get("controller_attestation"), dict): self.sign_evidence(value) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8") def reset_runtime_fixture(self) -> None: """Reset only disposable per-test runtime/formal trees between matrix rows.""" for path in (self.state, self.archive, self.intake): if path.exists(): shutil.rmtree(path) path.mkdir() self.controller_keys.clear() def begin(self) -> dict[str, object]: key = secrets.token_bytes(32) config = collector.load_config(self.config_path) result = refresh.refresh_begin( config, self.config_path, collector.parse_datetime(NOW, "test now"), _controller_key_commitment=hashlib.sha256(key).hexdigest(), ) self.assertEqual("BROWSER_REFRESH_REQUIRED", result["status"]) self.controller_keys[str(result["run_id"])] = key return result def sign_evidence(self, evidence: dict[str, object]) -> None: key = self.controller_keys.get(str(evidence.get("run_id"))) attestation = evidence.get("controller_attestation") if key is None or not isinstance(attestation, dict): return attestation["binding_sha256"] = None attestation["binding_sha256"] = hmac.new( key, refresh._controller_attestation_payload(evidence), hashlib.sha256 ).hexdigest() def formal_pending(self, config: collector.CollectorConfig) -> dict[str, object]: return { "schema_version": 2, "run_id": "a" * 32, "owner_nonce": "b" * 32, "phase": "TRANSACTION_INTENT", "task_id": refresh.TASK_ID, "creator_uid": config.creator_uid, "creator_dynamic_url": config.creator_dynamic_url, "started_at": "2026-08-13T02:00:00Z", "deadline_at": "2026-08-13T02:00:30Z", "window_start": "2026-08-10T02:00:00Z", "window_end": "2026-08-13T02:00:00Z", "config_sha256": hashlib.sha256(self.config_path.read_bytes()).hexdigest(), "state_manifest_preimage": refresh._identity(config.manifest_path).as_dict(), "formal_manifest_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(), "evidence_path": str(config.state_dir / "refresh" / "incoming" / ("a" * 32 + ".json")), "intake_root": str(config.refresh.intake_dir / ("a" * 32)), "evidence_identity": {"bytes": 2, "sha256": hashlib.sha256(b"{}").hexdigest()}, "planned_terminal": None, "transaction_identity": { "transaction_id": "c" * 32, "state_preimage": refresh._identity(config.manifest_path).as_dict(), "formal_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(), "state_candidate": refresh._identity(config.manifest_path).as_dict(), "formal_candidate": refresh._identity(config.refresh.formal_manifest).as_dict(), "created_artifacts": [], "input_item_count": 0, "new_item_count": 0, "formal_lock_claim": None, }, "last_transition_at": "2026-08-13T02:00:00Z", } def extractor(self) -> dict[str, object]: contract_raw = refresh.OBSERVATION_CONTRACT.read_bytes() contract = json.loads(contract_raw) return { "contract_id": contract["contract_id"], "contract_sha256": hashlib.sha256(contract_raw).hexdigest(), "parser_version": contract["parser_version"], "parser_sha256": hashlib.sha256(refresh.EXTRACTOR_SOURCE.read_bytes()).hexdigest(), } def evidence( self, begin: dict[str, object], *, outcome: str, observations: list[dict[str, object]] | None, marker: dict[str, object] | None, items: list[dict[str, object]] | None = None, discovery: str = "NOT_USED", action_outcome: str = "CONFIRMED", observation_outcome: str | None = None, refresh_count: int = 1, observation_count: int | None = None, ) -> dict[str, object]: outcome_map = { "READABLE": "READABLE", "UNREADABLE_TIMEOUT": "TIMEOUT", "UNREADABLE_ERROR": "ERROR", "ACCESS_BLOCKED": "ACCESS_BLOCKED", } observation_outcome = observation_outcome or outcome_map[outcome] if action_outcome == "PRE_DISPATCH_ERROR": refresh_started = refresh_finished = read_finished = "2026-08-13T02:00:01Z" action_elapsed = observation_elapsed = 0 observation_count = 0 refresh_action = None else: refresh_started = "2026-08-13T02:00:01Z" refresh_finished = "2026-08-13T02:00:16Z" read_finished = "2026-08-13T02:00:24Z" action_elapsed = 15000 observation_elapsed = 8000 observation_count = 1 if observation_count is None else observation_count refresh_action = "reload" evidence = { "schema_version": 3, "run_id": begin["run_id"], "transport": "codex_chrome_visible_page", "requested_url": "https://space.bilibili.com/1420210197/dynamic", "final_url": "https://space.bilibili.com/1420210197/dynamic", "refresh_action": refresh_action, "refresh_count": refresh_count, "refresh_started_at": refresh_started, "refresh_finished_at": refresh_finished, "read_finished_at": read_finished, "page_outcome": outcome, "page_title": "青枫浦上Q个人动态-青枫浦上Q动态记录-哔哩哔哩视频", "creator": {"uid": "1420210197", "name": "青枫浦上Q", "profile_url": "https://space.bilibili.com/1420210197"}, "extractor": self.extractor(), "page_observation": None if observations is None else { "schema_version": 1, "limits": json.loads(refresh.OBSERVATION_CONTRACT.read_text(encoding="utf-8"))["limits"], "observations": observations, "terminal_marker": marker, }, "items": items or [], "discovery_summary": {"status": discovery, "item_count": 0}, "safe_diagnostics": { "code": "OBSERVATION_TIMEOUT" if outcome == "UNREADABLE_TIMEOUT" else "OBSERVATION_ERROR" if outcome == "UNREADABLE_ERROR" else "ACCESS_INTERSTITIAL" if outcome == "ACCESS_BLOCKED" else "NONE", "overall_deadline_seconds": 120, "refresh_action_timeout_seconds": 35, "observation_timeout_seconds": 45, }, "runtime_contract": { "contract_id": begin["runtime_contract_id"], "contract_bytes": Path(refresh.RUNTIME_CONTRACT).stat().st_size, "contract_sha256": begin["runtime_contract_sha256"], }, "runtime_observation": { "refresh_action_outcome": action_outcome, "refresh_action_elapsed_ms": action_elapsed, "refresh_count": refresh_count, "observation_outcome": observation_outcome, "observation_elapsed_ms": observation_elapsed, "observation_count": observation_count, }, } if action_outcome == "PRE_DISPATCH_ERROR": action_started = action_finished = observation_started = observation_finished = None else: action_started = 1000 action_finished = 1000 + action_elapsed if observation_count: observation_started = action_finished observation_finished = observation_started + observation_elapsed else: observation_started = observation_finished = None write_started = max( 1000, action_finished or 0, observation_finished or 0, ) runtime = collector.load_config(self.config_path).refresh assert runtime is not None evidence["controller_attestation"] = { "controller_id": "bili-supported-chrome-controller-v1", "controller_sha256": hashlib.sha256(refresh.CONTROLLER_SOURCE.read_bytes()).hexdigest(), "binding_algorithm": "hmac-sha256-controller-envelope-v1", "action_dispatched": action_outcome != "PRE_DISPATCH_ERROR", "monotonic_run_started_ms": 0, "monotonic_action_started_ms": action_started, "monotonic_action_finished_ms": action_finished, "monotonic_observation_started_ms": observation_started, "monotonic_observation_finished_ms": observation_finished, "monotonic_evidence_write_started_ms": write_started, "binding_sha256": None, } self.sign_evidence(evidence) return evidence def set_runtime_diagnostic(self, evidence: dict[str, object], code: str) -> None: diagnostics = evidence["safe_diagnostics"] assert isinstance(diagnostics, dict) diagnostics["code"] = code @staticmethod def observation(cards: list[dict[str, object]], unparsed: list[dict[str, object]] | None = None) -> dict[str, object]: unparsed = unparsed or [] return { "ordinal": 0, "observed_at": "2026-08-13T02:00:24Z", "cursor_before": 0, "cursor_after": len(cards) + len(unparsed), "visible_node_count": len(cards) + len(unparsed), "complete_card_count": len(cards), "unparsed_node_count": len(unparsed), "cards": cards, "unparsed_nodes": unparsed, "limit_hit": "NONE", } def commit(self, begin: dict[str, object], evidence: dict[str, object]) -> tuple[int, dict[str, object]]: path = self.bind_for_test(begin, evidence) return collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"]) def bind_for_test(self, begin: dict[str, object], evidence: dict[str, object]) -> Path: self.sign_evidence(evidence) path = Path(str(begin["evidence_path"])) self.write_json(path, evidence) config = collector.load_config(self.config_path) pending = refresh._load_pending(config) assert pending is not None key = self.controller_keys[str(begin["run_id"])] value, payload, _, _ = refresh._validate_evidence(config, pending, path, _controller_key=key) pending["evidence_identity"] = { "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest() } pending["controller_binding_sha256"] = value["controller_attestation"]["binding_sha256"] pending["phase"] = "EVIDENCE_BOUND" pending["last_transition_at"] = "2026-08-13T02:00:24Z" refresh._write_pending(config, pending) return path def validate_for_test(self, begin: dict[str, object], evidence: dict[str, object]) -> None: path = Path(str(begin["evidence_path"])) self.write_json(path, evidence) config = collector.load_config(self.config_path) pending = refresh._load_pending(config) assert pending is not None refresh._validate_evidence( config, pending, path, _controller_key=self.controller_keys[str(begin["run_id"])] ) @staticmethod def bound_card_item(identifier: str = "456") -> tuple[dict[str, object], dict[str, object]]: source = f"https://www.bilibili.com/opus/{identifier}" card = { "position": 0, "identifiers": {"dynamic_id": identifier, "opus_id": identifier, "bvid": None}, "stable_keys": [f"dynamic:{identifier}", f"opus:{identifier}", f"url:{source}"], "published_at": "2026-08-13T01:55:00Z", "content_type": "text", "source_url": source, } item = { "dynamic_id": identifier, "opus_id": identifier, "bvid": None, "content_type": "text", "published_at": "2026-08-13T01:55:00Z", "title": "offline body", "source_url": source, "body_text": "offline body", "body_complete": True, "duration_seconds": None, "artifacts": [], } return card, item @staticmethod def end_marker() -> dict[str, object]: selector = "DYNAMIC_FEED_END_TEXT" text = "没有更多动态了" return { "observation_ordinal": 0, "kind": "EXACT_END_OF_FEED", "selector_id": selector, "normalized_text": text, "marker_sha256": hashlib.sha256(f"{selector}\n{text}".encode()).hexdigest(), } def test_real_timeout_and_412_replay_never_claim_no_new(self) -> None: begin = self.begin() before = (self.archive / "manifest.jsonl").read_bytes() if (self.archive / "manifest.jsonl").exists() else b"" evidence = self.evidence(begin, outcome="UNREADABLE_TIMEOUT", observations=None, marker=None, discovery="BLOCKED_412") code, result = self.commit(begin, evidence) self.assertEqual(4, code) self.assertEqual("REFRESH_FAILED_PAGE_UNREADABLE", result["status"]) self.assertFalse(result["no_new_confirmed"]) self.assertFalse(result["formal_manifest_changed"]) after = (self.archive / "manifest.jsonl").read_bytes() if (self.archive / "manifest.jsonl").exists() else b"" self.assertEqual(before, after) self.assertFalse((self.state / "refresh" / "pending.json").exists()) slot = json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8")) self.assertEqual("TERMINAL", slot["run_state"]) self.assertEqual("BLOCKED_412", evidence["discovery_summary"]["status"]) def test_empty_page_requires_exact_marker_for_no_new(self) -> None: begin = self.begin() config = collector.load_config(self.config_path) pending_before = refresh._load_pending(config) self.assertIsNotNone(pending_before) observation = self.observation([]) marker = { "observation_ordinal": 0, "kind": "EXACT_END_OF_FEED", "selector_id": "DYNAMIC_FEED_END_TEXT", "normalized_text": "没有更多动态了", "marker_sha256": hashlib.sha256("DYNAMIC_FEED_END_TEXT\n没有更多动态了".encode()).hexdigest(), } code, result = self.commit(begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=marker)) self.assertEqual(0, code) self.assertEqual("REFRESH_CONFIRMED_NO_NEW", result["status"]) self.assertTrue(result["coverage_complete"]) slot_path = Path(str(result["run_evidence_path"])) slot_payload = slot_path.read_bytes() slot = json.loads(slot_payload) receipt = slot["transaction_receipt"] self.assertEqual("NO_FORMAL_CHANGE", receipt["business_commit_kind"]) self.assertEqual(receipt["state_preimage"], receipt["state_candidate"]) self.assertEqual(receipt["formal_preimage"], receipt["formal_candidate"]) latest_path = self.state / "refresh" / "runs" / "latest.json" latest = json.loads(latest_path.read_text(encoding="utf-8")) self.assertEqual( {"schema_version", "slot_index", "hour_epoch", "run_id", "slot_bytes", "slot_sha256", "status", "terminal_at"}, set(latest), ) self.assertEqual(len(slot_payload), latest["slot_bytes"]) self.assertEqual(hashlib.sha256(slot_payload).hexdigest(), latest["slot_sha256"]) latest_path.unlink() replay_pending = dict(pending_before) replay_pending["phase"] = "TERMINAL_RECORDED" replay_pending["planned_terminal"] = {key: slot[key] for key in ( "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", "transaction_receipt", )} replay_pending["planned_terminal"]["formal_manifest_changed"] = False refresh._write_pending(config, replay_pending, create=True) replay = refresh._recover_or_replay( config, self.config_path, collector.parse_now("2026-08-13T10:00:26+08:00") ) self.assertEqual("REFRESH_CONFIRMED_NO_NEW", replay["status"]) self.assertEqual(latest, json.loads(latest_path.read_text(encoding="utf-8"))) def test_unparsed_node_is_mutually_exclusive_and_forces_partial(self) -> None: begin = self.begin() node = {"position": 0, "node_fingerprint_sha256": "a" * 64, "reason_code": "PARSER_REJECTED"} observation = self.observation([], [node]) marker = { "observation_ordinal": 0, "kind": "EXACT_END_OF_FEED", "selector_id": "DYNAMIC_FEED_END_TEXT", "normalized_text": "已经到底了", "marker_sha256": hashlib.sha256("DYNAMIC_FEED_END_TEXT\n已经到底了".encode()).hexdigest(), } code, result = self.commit(begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=marker)) self.assertEqual(4, code) self.assertEqual("PARTIAL_DISCOVERY_UNCONFIRMED", result["status"]) self.assertFalse(result["coverage_complete"]) def test_node_position_overlap_fails_before_formal_write(self) -> None: begin = self.begin() card = { "position": 0, "identifiers": {"dynamic_id": "123", "opus_id": None, "bvid": None}, "stable_keys": ["dynamic:123", "url:https://www.bilibili.com/opus/123"], "published_at": "2026-08-13T01:55:00Z", "content_type": "text", "source_url": "https://www.bilibili.com/opus/123", } node = {"position": 0, "node_fingerprint_sha256": "b" * 64, "reason_code": "PARSER_REJECTED"} observation = self.observation([card], [node]) with self.assertRaisesRegex(collector.CollectorError, "positions overlap"): self.validate_for_test( begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=None) ) self.assertFalse((self.archive / "manifest.jsonl").exists()) def test_complete_observed_card_with_missing_item_cannot_claim_no_new(self) -> None: begin = self.begin() card, _ = self.bound_card_item() evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([card])], marker=self.end_marker(), items=[], ) with self.assertRaises(collector.CollectorError) as failure: self.validate_for_test(begin, evidence) self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code) self.assertFalse((self.archive / "manifest.jsonl").exists()) def test_extra_item_without_observed_card_fails_closed(self) -> None: begin = self.begin() _, item = self.bound_card_item() evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(), items=[item], ) with self.assertRaises(collector.CollectorError) as failure: self.validate_for_test(begin, evidence) self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code) self.assertFalse((self.archive / "manifest.jsonl").exists()) def test_observation_item_content_binding_mismatch_fails_closed(self) -> None: begin = self.begin() card, item = self.bound_card_item() item["content_type"] = "article" evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([card])], marker=self.end_marker(), items=[item], ) with self.assertRaises(collector.CollectorError) as failure: self.validate_for_test(begin, evidence) self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code) def test_observation_overlapping_component_identity_fails_closed(self) -> None: begin = self.begin() card, item = self.bound_card_item() other_card, other_item = self.bound_card_item("457") other_card["identifiers"]["dynamic_id"] = "456" other_card["stable_keys"] = [ "dynamic:456", "opus:457", "url:https://www.bilibili.com/opus/457", ] other_item["dynamic_id"] = "456" for position, row in enumerate((card, other_card)): row["position"] = position evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([card, other_card])], marker=self.end_marker(), items=[item, other_item], ) with self.assertRaises(collector.CollectorError) as failure: self.validate_for_test(begin, evidence) self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code) def test_new_text_commits_both_manifests_and_artifact(self) -> None: begin = self.begin() card = { "position": 0, "identifiers": {"dynamic_id": "456", "opus_id": "456", "bvid": None}, "stable_keys": ["dynamic:456", "opus:456", "url:https://www.bilibili.com/opus/456"], "published_at": "2026-08-13T01:55:00Z", "content_type": "text", "source_url": "https://www.bilibili.com/opus/456", } item = { "dynamic_id": "456", "opus_id": "456", "bvid": None, "content_type": "text", "published_at": "2026-08-13T01:55:00Z", "title": "离线合成正文", "source_url": "https://www.bilibili.com/opus/456", "body_text": "仅测试文本", "body_complete": True, "duration_seconds": None, "artifacts": [], } observation = self.observation([card]) marker = { "observation_ordinal": 0, "kind": "EXACT_END_OF_FEED", "selector_id": "DYNAMIC_FEED_END_TEXT", "normalized_text": "没有更多动态了", "marker_sha256": hashlib.sha256("DYNAMIC_FEED_END_TEXT\n没有更多动态了".encode()).hexdigest(), } code, result = self.commit(begin, self.evidence(begin, outcome="READABLE", observations=[observation], marker=marker, items=[item])) self.assertEqual(0, code) self.assertEqual("NEW_ITEMS_SAVED", result["status"]) self.assertEqual(1, result["new_items"]) self.assertEqual(1, result["saved_artifacts"]) self.assertEqual(1, len(collector.load_manifest(self.state / "manifest.jsonl"))) formal = (self.archive / "manifest.jsonl").read_text(encoding="utf-8") self.assertIn('"event_type":"DYNAMIC_CONTENT_SAVED"', formal) artifacts = [path for path in self.archive.glob("*.txt")] self.assertEqual(1, len(artifacts)) self.assertEqual("仅测试文本\n", artifacts[0].read_text(encoding="utf-8")) slot = json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8")) self.assertEqual("DUAL_MANIFEST_COMMIT", slot["transaction_receipt"]["business_commit_kind"]) self.assertEqual("BUSINESS_COMMITTED", slot["transaction_receipt"]["phase"]) def test_hourly_slot_and_dead_pending_recovery_are_bounded(self) -> None: begin = self.begin() with self.assertRaises(collector.CollectorError) as busy: collector.run(["--config", str(self.config_path), "refresh-begin", "--now", "2026-08-13T10:00:10+08:00"]) self.assertEqual("E_BUSY", busy.exception.code) code, terminal = collector.run(["--config", str(self.config_path), "refresh-begin", "--now", "2026-08-13T10:02:01+08:00"]) self.assertEqual(4, code) self.assertEqual("E_EVIDENCE_MISSING_AFTER_DEADLINE", terminal["error_code"]) with self.assertRaises(collector.CollectorError) as occupied: refresh.refresh_begin( collector.load_config(self.config_path), self.config_path, collector.parse_now("2026-08-13T10:02:10+08:00"), _controller_key_commitment=hashlib.sha256(b"next-controller").hexdigest(), ) self.assertEqual("E_RUN_HOUR_OCCUPIED", occupied.exception.code) def test_latest_commit_failure_retains_pending_and_reopen_rebuilds_index(self) -> None: begin = self.begin() config = collector.load_config(self.config_path) observation = self.observation([]) evidence = self.evidence( begin, outcome="READABLE", observations=[observation], marker=self.end_marker(), items=[] ) path = Path(str(begin["evidence_path"])) self.write_json(path, evidence) self.bind_for_test(begin, evidence) original = refresh._write_readback injected = {"done": False} def fail_latest(target: Path, payload: bytes, description: str) -> None: if description == "latest run index" and not injected["done"]: injected["done"] = True raise OSError("injected latest failure") original(target, payload, description) refresh._write_readback = fail_latest try: with self.assertRaisesRegex(OSError, "injected latest failure"): collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"]) finally: refresh._write_readback = original self.assertTrue(refresh._pending_path(config).is_file()) self.assertFalse((self.state / "refresh" / "runs" / "latest.json").exists()) replay = refresh._recover_or_replay( config, self.config_path, collector.parse_now("2026-08-13T10:00:26+08:00"), allow_final_evidence=True, ) self.assertEqual("REFRESH_CONFIRMED_NO_NEW", replay["status"]) self.assertTrue((self.state / "refresh" / "runs" / "latest.json").is_file()) self.assertFalse(refresh._pending_path(config).exists()) def test_state_lock_is_kernel_backed_and_owner_metadata_recovers(self) -> None: config = collector.load_config(self.config_path) config.state_dir.mkdir() stale = { "pid": 99999999, "process_created_at": "0", "run_id": None, "acquired_at": "2026-08-13T00:00:00Z", } self.write_json(config.state_dir / ".collector.lock.owner.json", stale) with collector.StateLock(config.lock_path): self.assertEqual(1, config.lock_path.stat().st_size) self.assertTrue((config.state_dir / ".collector.lock.owner.json").is_file()) with self.assertRaises(collector.CollectorError) as locked: with collector.StateLock(config.lock_path): pass self.assertEqual("E_STATE_LOCKED", locked.exception.code) self.assertTrue(config.lock_path.is_file()) self.assertFalse((config.state_dir / ".collector.lock.owner.json").exists()) def test_current_formal_catalog_shape_is_140_111_103_8_0_read_only(self) -> None: formal = Path(__file__).resolve().parents[3] / "ana-data" / "news-青枫浦上Q" / "manifest.jsonl" if not formal.exists(): self.skipTest("formal read-only acceptance snapshot is not present") before = (formal.stat().st_size, hashlib.sha256(formal.read_bytes()).hexdigest()) config = collector.load_config(self.config_path) archive = formal.parent rebound = collector.CollectorConfig( config.creator_name, config.creator_uid, config.creator_dynamic_url, config.timezone_name, config.window_hours, config.state_dir, config.download_dir, config.video_dir, config.minimum_complete_age_seconds, config.title_max_length, config.allowed_source_hosts, config.allowed_video_extensions, config.native_handoff, collector.RefreshConfig( archive, formal, config.refresh.intake_dir, config.refresh.overall_deadline_seconds, config.refresh.refresh_action_timeout_seconds, config.refresh.observation_timeout_seconds, config.refresh.page_internal_settle_timeout_seconds, 1, 168, 200, 20, config.refresh.max_image_bytes, config.refresh.max_text_bytes, ), ) _, _, counts = refresh.load_formal_catalog(rebound) self.assertEqual({"events": 140, "components": 111, "saved": 103, "video": 8, "retryable": 0}, counts) self.assertEqual(before, (formal.stat().st_size, hashlib.sha256(formal.read_bytes()).hexdigest())) def test_mixed_history_union_saved_precedence_and_schema2_identity(self) -> None: artifact = self.archive / "saved.txt" artifact.write_bytes(b"saved\n") digest = hashlib.sha256(artifact.read_bytes()).hexdigest() config = collector.load_config(self.config_path) pending = { "schema_version": 1, "creator": config.creator_name, "creator_uid": config.creator_uid, "item_type": "article", "stable_id": "700", "source_url": "https://www.bilibili.com/opus/700", "published_at": "2026-08-12T01:00:00Z", "collected_at": "2026-08-12T02:00:00Z", "status": "CONTENT_ACCESS_PENDING_CHROME", } saved = { **pending, "status": "SAVED", "path": "saved.txt", "bytes": artifact.stat().st_size, "sha256": digest.upper(), } formal = self.archive / "manifest.jsonl" formal.write_bytes(collector.canonical_json_bytes(pending) + collector.canonical_json_bytes(saved)) _, tokens, counts = refresh.load_formal_catalog(config) self.assertEqual({"events": 2, "components": 1, "saved": 1, "video": 0, "retryable": 0}, counts) self.assertEqual({"opus:700", "url:https://www.bilibili.com/opus/700"}, tokens) def test_schema2_status_entity_and_artifact_identity_fail_closed(self) -> None: artifact = self.archive / "new.txt" artifact.write_bytes(b"new\n") source = "https://www.bilibili.com/opus/701" keys = ["dynamic:701", "opus:701", f"url:{source}"] config = collector.load_config(self.config_path) event = { "schema_version": 2, "event_type": "DYNAMIC_CONTENT_SAVED", "creator": config.creator_name, "creator_uid": config.creator_uid, "entity_id": collector.entity_id_for_keys(keys), "dynamic_id": "701", "opus_id": "701", "bvid": None, "dedupe_keys": keys, "content_type": "text", "published_at": "2026-08-12T01:00:00Z", "title": "saved", "source_url": source, "collected_at": "2026-08-12T02:00:00Z", "artifacts": [{"kind": "text", "sequence": 1, "path": "new.txt", "bytes": 4, "sha256": hashlib.sha256(b"new\n").hexdigest()}], "duration_seconds": None, "page_run_id": "a" * 32, "coverage_complete": True, "status": "SAVED", } mutations = ( ("status", {**event, "status": "UNKNOWN"}, "E_CATALOG_STATUS"), ("integer_uid", {**event, "creator_uid": int(config.creator_uid)}, "E_CATALOG_STATUS"), ("entity", {**event, "entity_id": "0" * 24}, "E_CATALOG_IDENTITY_CONFLICT"), ("artifact", {**event, "artifacts": [{**event["artifacts"][0], "path": "absent.txt"}]}, "E_CATALOG_ARTIFACT"), ) for label, bad, expected in mutations: with self.subTest(label=label): (self.archive / "manifest.jsonl").write_bytes(collector.canonical_json_bytes(bad)) with self.assertRaises(collector.CollectorError) as failure: refresh.load_formal_catalog(config) self.assertEqual(expected, failure.exception.code) def test_schema1_stable_id_must_be_one_to_thirty_two_ascii_digits(self) -> None: config = collector.load_config(self.config_path) base = { "schema_version": 1, "creator": config.creator_name, "creator_uid": config.creator_uid, "item_type": "article", "source_url": "https://www.bilibili.com/opus/123", "published_at": "2026-08-12T01:00:00Z", "collected_at": "2026-08-12T02:00:00Z", "status": "CONTENT_ACCESS_PENDING_CHROME", } for stable_id in ("1" * 33, "１２３"): with self.subTest(stable_id=stable_id): row = {**base, "stable_id": stable_id} (self.archive / "manifest.jsonl").write_bytes(collector.canonical_json_bytes(row)) with self.assertRaises(collector.CollectorError) as failure: refresh.load_formal_catalog(config) self.assertEqual("E_CATALOG_IDENTITY_CONFLICT", failure.exception.code) def test_pending_before_started_rebuilds_same_run_and_third_slot_is_untouched(self) -> None: config = collector.load_config(self.config_path) original = refresh._ensure_started_slot refresh._ensure_started_slot = lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("after pending")) try: with self.assertRaisesRegex(OSError, "after pending"): refresh.refresh_begin( config, self.config_path, collector.parse_now(NOW), _controller_key_commitment=hashlib.sha256(b"test-controller").hexdigest(), ) finally: refresh._ensure_started_slot = original pending = refresh._load_pending(config) self.assertIsNotNone(pending) run_id = pending["run_id"] result = refresh.refresh_begin(config, self.config_path, collector.parse_now("2026-08-13T10:00:01+08:00")) self.assertEqual(run_id, result["run_id"]) self.assertEqual("RUN_STARTED", json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8"))["run_state"]) Path(str(result["run_evidence_path"])).write_bytes(b'{"third":true}') before = Path(str(result["run_evidence_path"])).read_bytes() with self.assertRaises(collector.CollectorError) as failure: refresh._recover_or_replay(config, self.config_path, collector.parse_now("2026-08-13T10:00:02+08:00")) self.assertEqual("E_RECOVERY_AMBIGUOUS", failure.exception.code) self.assertEqual(before, Path(str(result["run_evidence_path"])).read_bytes()) def test_terminal_never_overwrites_third_slot_and_recorded_replay_binds_plan(self) -> None: begin = self.begin() slot_path = Path(str(begin["run_evidence_path"])) third = b'{"third":"terminal"}' evidence = self.evidence(begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker()) original_write_pending = refresh._write_pending injected = {"done": False} def inject_third(config, pending, *, create=False): original_write_pending(config, pending, create=create) if pending.get("planned_terminal") is not None and not injected["done"]: injected["done"] = True slot_path.write_bytes(third) refresh._write_pending = inject_third try: with self.assertRaises(collector.CollectorError) as failure: self.commit(begin, evidence) finally: refresh._write_pending = original_write_pending self.assertEqual("E_RUN_EVIDENCE_COMMIT", failure.exception.code) self.assertEqual(third, slot_path.read_bytes()) config = collector.load_config(self.config_path) pending = refresh._load_pending(config) self.assertIsNotNone(pending) self.assertIsNotNone(pending["planned_terminal"]) slot, payload = refresh._terminal_slot_from_planned(config, pending, pending["planned_terminal"]) slot_path.write_bytes(payload) pending["phase"] = "TERMINAL_RECORDED" pending["planned_terminal"]["transaction_receipt"]["intent_sha256"] = "0" * 64 refresh._write_pending(config, pending) before = slot_path.read_bytes() with self.assertRaises(collector.CollectorError) as replay_failure: refresh._recover_or_replay(config, self.config_path, collector.parse_now("2026-08-13T10:00:26+08:00")) self.assertEqual("E_RUN_EVIDENCE_COMMIT", replay_failure.exception.code) self.assertEqual(before, slot_path.read_bytes()) def test_mixed_history_cross_line_identity_conflict_fails_closed(self) -> None: config = collector.load_config(self.config_path) first = { "schema_version": 1, "creator": config.creator_name, "creator_uid": config.creator_uid, "item_type": "video", "stable_id": "BV1HA3o6oEJJ", "source_url": "https://www.bilibili.com/video/BV1HA3o6oEJJ", "published_at": "2026-08-12T01:00:00Z", "collected_at": "2026-08-12T02:00:00Z", "status": "VIDEO_DOWNLOAD_PENDING_EXTENSION", } second = {**first, "stable_id": "BV1DVMX6XEPq"} (self.archive / "manifest.jsonl").write_bytes( collector.canonical_json_bytes(first) + collector.canonical_json_bytes(second) ) with self.assertRaises(collector.CollectorError) as failure: refresh.load_formal_catalog(config) self.assertEqual("E_CATALOG_IDENTITY_CONFLICT", failure.exception.code) def test_dead_formal_lock_takeover_uses_archive_quarantine_and_current_identity(self) -> None: config = collector.load_config(self.config_path) config.state_dir.mkdir() pending = { "schema_version": 2, "run_id": "a" * 32, "owner_nonce": "b" * 32, "phase": "TRANSACTION_INTENT", "task_id": refresh.TASK_ID, "creator_uid": config.creator_uid, "creator_dynamic_url": config.creator_dynamic_url, "started_at": "2026-08-13T02:00:00Z", "deadline_at": "2026-08-13T02:00:30Z", "window_start": "2026-08-10T02:00:00Z", "window_end": "2026-08-13T02:00:00Z", "config_sha256": hashlib.sha256(self.config_path.read_bytes()).hexdigest(), "state_manifest_preimage": refresh._identity(config.manifest_path).as_dict(), "formal_manifest_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(), "evidence_path": str(config.state_dir / "refresh" / "incoming" / ("a" * 32 + ".json")), "intake_root": str(config.refresh.intake_dir / ("a" * 32)), "evidence_identity": {"bytes": 2, "sha256": hashlib.sha256(b"{}").hexdigest()}, "planned_terminal": None, "transaction_identity": { "transaction_id": "c" * 32, "state_preimage": refresh._identity(config.manifest_path).as_dict(), "formal_preimage": refresh._identity(config.refresh.formal_manifest).as_dict(), "state_candidate": refresh._identity(config.manifest_path).as_dict(), "formal_candidate": refresh._identity(config.refresh.formal_manifest).as_dict(), "created_artifacts": [], "input_item_count": 0, "new_item_count": 0, "formal_lock_claim": None, }, "last_transition_at": "2026-08-13T02:00:00Z", } old_record = { "schema_version": 1, "task_id": refresh.TASK_ID, "run_id": pending["run_id"], "owner_nonce": pending["owner_nonce"], "transaction_id": pending["transaction_identity"]["transaction_id"], "recovery_generation": 0, "holder_pid": 99999999, "holder_process_created_at": "0", "lock_created_at": "2026-08-13T02:00:00Z", } old_payload = collector.canonical_json_bytes(old_record, newline=False) old_claim = { "claim_state": "HELD", "lock_path": str(refresh._formal_lock_path(config)), "lock_bytes": len(old_payload), "lock_sha256": hashlib.sha256(old_payload).hexdigest(), "lock_record": old_record, } pending["transaction_identity"]["formal_lock_claim"] = old_claim refresh._write_pending(config, pending, create=True) refresh._create_new(refresh._formal_lock_path(config), old_payload) refresh._recovery_formal_lock(config, pending, collector.parse_now("2026-08-13T10:00:25+08:00")) claim = pending["transaction_identity"]["formal_lock_claim"] self.assertEqual("HELD", claim["claim_state"]) self.assertEqual(1, claim["lock_record"]["recovery_generation"]) self.assertEqual(os.getpid(), claim["lock_record"]["holder_pid"]) takeover = pending["transaction_identity"]["takeover"] self.assertEqual("NEW_LOCK_HELD", takeover["phase"]) quarantine = Path(takeover["quarantine_path"]) self.assertEqual(config.refresh.archive_dir, quarantine.parent.parent) self.assertEqual(old_payload, quarantine.read_bytes()) self.assertEqual(collector.canonical_json_bytes(claim["lock_record"], newline=False), refresh._formal_lock_path(config).read_bytes()) lock_path = refresh._formal_lock_path(config) lock_path.unlink() claim["claim_state"] = "PLANNED" claim["lock_record"]["holder_pid"] = 99999999 claim["lock_record"]["holder_process_created_at"] = "0" dead_payload = collector.canonical_json_bytes(claim["lock_record"], newline=False) claim["lock_bytes"] = len(dead_payload) claim["lock_sha256"] = hashlib.sha256(dead_payload).hexdigest() takeover["phase"] = "CREATE_PLANNED" prior_generation = takeover["next_generation"] prior_attempt = takeover["claim_attempt"] refresh._write_pending(config, pending) refresh._recovery_formal_lock(config, pending, collector.parse_now("2026-08-13T10:00:26+08:00")) claim = pending["transaction_identity"]["formal_lock_claim"] takeover = pending["transaction_identity"]["takeover"] self.assertEqual(prior_generation, claim["lock_record"]["recovery_generation"]) self.assertEqual(prior_attempt + 1, takeover["claim_attempt"]) self.assertEqual(os.getpid(), claim["lock_record"]["holder_pid"]) lock_path.unlink() claim["claim_state"] = "PLANNED" claim["lock_record"]["holder_pid"] = 99999999 claim["lock_record"]["holder_process_created_at"] = "0" dead_payload = collector.canonical_json_bytes(claim["lock_record"], newline=False) claim["lock_bytes"] = len(dead_payload) claim["lock_sha256"] = hashlib.sha256(dead_payload).hexdigest() takeover["phase"] = "CREATE_PLANNED" refresh._write_pending(config, pending) refresh._create_new(lock_path, dead_payload) original_rename = refresh._rename_no_overwrite rename_calls: list[tuple[Path, Path]] = [] def fail_after_intent(source: Path, target: Path) -> None: rename_calls.append((source, target)) raise OSError("crash before failed-claim rename") refresh._rename_no_overwrite = fail_after_intent try: with self.assertRaisesRegex(OSError, "crash before failed-claim rename"): refresh._recovery_formal_lock(config, pending, collector.parse_now("2026-08-13T10:00:27+08:00")) finally: refresh._rename_no_overwrite = original_rename persisted = refresh._load_pending(config) intent = persisted["transaction_identity"]["takeover"]["failed_claim_intent"] self.assertEqual(str(rename_calls[0][1]), intent["path"]) self.assertEqual(dead_payload, lock_path.read_bytes()) self.assertFalse(Path(intent["path"]).exists()) refresh._recovery_formal_lock(config, persisted, collector.parse_now("2026-08-13T10:00:27+08:00")) pending = persisted takeover = pending["transaction_identity"]["takeover"] self.assertEqual(1, len(takeover["failed_claims"])) failed = takeover["failed_claims"][0] self.assertEqual(dead_payload, Path(failed["path"]).read_bytes()) self.assertTrue((refresh._quarantine_root(config) / ".owner.json").is_file()) refresh._formal_lock_path(config).unlink() refresh._release_formal_lock(config, pending) self.assertFalse(refresh._formal_lock_path(config).exists()) self.assertFalse(quarantine.exists()) self.assertFalse(refresh._quarantine_root(config).exists()) def test_quarantine_rename_is_no_overwrite_and_empty_cleanup_reopens(self) -> None: config = collector.load_config(self.config_path) source = refresh._formal_lock_path(config) root = refresh._quarantine_root(config) root.mkdir() target = root / "third.json" source.write_bytes(b"owned") target.write_bytes(b"third") with self.assertRaises(collector.CollectorError) as collision: refresh._rename_no_overwrite(source, target) self.assertEqual("E_RECOVERY_AMBIGUOUS", collision.exception.code) self.assertEqual(b"owned", source.read_bytes()) self.assertEqual(b"third", target.read_bytes()) target.unlink() flushed: list[Path] = [] original_fsync = refresh._fsync_directory refresh._fsync_directory = lambda path: flushed.append(Path(path)) try: refresh._rename_no_overwrite(source, target) finally: refresh._fsync_directory = original_fsync self.assertFalse(source.exists()) self.assertEqual(b"owned", target.read_bytes()) self.assertIn(self.archive, flushed) self.assertIn(root, flushed) target.unlink() pending = { "run_id": "a" * 32, "owner_nonce": "b" * 32, "transaction_identity": { "formal_lock_claim": { "lock_bytes": 0, "lock_sha256": hashlib.sha256(b"").hexdigest() }, "takeover": {"old_lock_never_created": True, "failed_claims": []}, }, } refresh._release_formal_lock(config, pending) self.assertFalse(root.exists()) def test_formal_claim_requires_exact_readback_and_parent_durability_before_held(self) -> None: config = collector.load_config(self.config_path) config.state_dir.mkdir() now = collector.parse_now("2026-08-13T10:00:25+08:00") formal_before = refresh._identity(config.refresh.formal_manifest).as_dict() def prepare() -> tuple[dict[str, object], dict[str, object], Path]: pending = self.formal_pending(config) claim = refresh._new_formal_claim(config, pending, 1, now) pending["transaction_identity"]["formal_lock_claim"] = claim pending["transaction_identity"]["takeover"] = { "phase": "CREATE_PLANNED", "old_lock_sha256": hashlib.sha256(b"").hexdigest(), "old_lock_hex": "", "quarantine_path": str(refresh._quarantine_root(config) / "never-created.json"), "next_generation": 1, "takeover_reason": "TEST", "planned_at": collector.canonical_datetime(now), "claim_attempt": 1, "old_lock_never_created": True, } refresh._write_pending(config, pending, create=True) return pending, claim, refresh._formal_lock_path(config) pending, _, lock = prepare() original_create = refresh._create_new refresh._create_new = lambda path, payload: path.write_bytes(b"third-party-drift") try: with self.assertRaises(collector.CollectorError) as drift: refresh._recovery_formal_lock(config, pending, now) finally: refresh._create_new = original_create self.assertEqual("E_FORMAL_LOCK", drift.exception.code) self.assertEqual(b"third-party-drift", lock.read_bytes()) persisted = refresh._load_pending(config) self.assertEqual("CREATE_PLANNED", persisted["transaction_identity"]["takeover"]["phase"]) self.assertEqual("PLANNED", persisted["transaction_identity"]["formal_lock_claim"]["claim_state"]) with self.assertRaises(collector.CollectorError) as replay_drift: refresh._recovery_formal_lock(config, persisted, now) self.assertEqual("E_RECOVERY_AMBIGUOUS", replay_drift.exception.code) self.assertEqual(b"third-party-drift", lock.read_bytes()) lock.unlink() refresh._pending_path(config).unlink() pending, _, lock = prepare() original_read = Path.read_bytes def fail_lock_read(path: Path) -> bytes: if path == lock: raise OSError("injected formal lock readback failure") return original_read(path) Path.read_bytes = fail_lock_read try: with self.assertRaisesRegex(OSError, "readback failure"): refresh._recovery_formal_lock(config, pending, now) finally: Path.read_bytes = original_read self.assertEqual("CREATE_PLANNED", refresh._load_pending(config)["transaction_identity"]["takeover"]["phase"]) lock.unlink() refresh._pending_path(config).unlink() pending, claim, lock = prepare() original_fsync = refresh._fsync_directory def fail_parent_fsync(path: Path) -> None: if Path(path) == self.archive: raise OSError("injected parent directory fsync failure") original_fsync(path) refresh._fsync_directory = fail_parent_fsync try: with self.assertRaisesRegex(OSError, "parent directory fsync failure"): refresh._recovery_formal_lock(config, pending, now) finally: refresh._fsync_directory = original_fsync persisted = refresh._load_pending(config) self.assertEqual("CREATE_PLANNED", persisted["transaction_identity"]["takeover"]["phase"]) self.assertEqual("PLANNED", persisted["transaction_identity"]["formal_lock_claim"]["claim_state"]) self.assertEqual(refresh._claim_payload(claim), lock.read_bytes()) refresh._recovery_formal_lock(config, persisted, now) replayed = refresh._load_pending(config) self.assertEqual("NEW_LOCK_HELD", replayed["transaction_identity"]["takeover"]["phase"]) self.assertEqual("HELD", replayed["transaction_identity"]["formal_lock_claim"]["claim_state"]) self.assertEqual(formal_before, refresh._identity(config.refresh.formal_manifest).as_dict()) refresh._release_formal_lock(config, replayed) def test_formal_claim_real_child_kill_reopens_create_and_new_lock_held(self) -> None: config = collector.load_config(self.config_path) config.state_dir.mkdir() formal_before = refresh._identity(config.refresh.formal_manifest).as_dict() script = r'''import hashlib, sys, time from pathlib import Path sys.path.insert(0, sys.argv[1]) import bili_dynamic_collector as core import bili_dynamic_refresh as refresh config = core.load_config(Path(sys.argv[2])) marker = Path(sys.argv[3]) mode = sys.argv[4] pending = refresh._load_pending(config) now = core.parse_now("2026-08-13T10:00:25+08:00") claim = refresh._new_formal_claim(config, pending, 1, now) pending["transaction_identity"]["formal_lock_claim"] = claim pending["transaction_identity"]["takeover"] = { "phase": "CREATE_PLANNED", "old_lock_sha256": hashlib.sha256(b"").hexdigest(), "old_lock_hex": "", "quarantine_path": str(refresh._quarantine_root(config) / "never-created.json"), "next_generation": 1, "takeover_reason": "CHILD_KILL_TEST", "planned_at": core.canonical_datetime(now), "claim_attempt": 1, "old_lock_never_created": True, } refresh._write_pending(config, pending) if mode == "CREATE_PLANNED": original = refresh._fsync_directory def stop_at_parent(path): if Path(path) == config.refresh.archive_dir and refresh._formal_lock_path(config).exists(): marker.write_text("created", encoding="utf-8") time.sleep(120) original(path) refresh._fsync_directory = stop_at_parent refresh._recovery_formal_lock(config, pending, now) marker.write_text("held", encoding="utf-8") time.sleep(120) ''' for mode, marker_text in (("CREATE_PLANNED", "created"), ("NEW_LOCK_HELD", "held")): with self.subTest(mode=mode): pending = self.formal_pending(config) refresh._write_pending(config, pending, create=True) marker = self.root / f"{mode}.marker" process = subprocess.Popen( [sys.executable, "-B", "-c", script, str(PROJECT_DEV), str(self.config_path), str(marker), mode], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) deadline = time.monotonic() + 15 while time.monotonic() < deadline and not marker.exists() and process.poll() is None: time.sleep(0.05) self.assertTrue(marker.exists(), f"child failed before {mode} marker; exit={process.poll()}") self.assertEqual(marker_text, marker.read_text(encoding="utf-8")) process.kill() process.wait(timeout=10) child_pid = process.pid if os.name == "nt": process._handle.Close() del process gc.collect() dead_deadline = time.monotonic() + 10 while time.monotonic() < dead_deadline: try: collector.process_created_at(child_pid) except ProcessLookupError: break time.sleep(0.05) else: self.fail(f"child process identity remained queryable after kill: {child_pid}") persisted = refresh._load_pending(config) expected_phase = "CREATE_PLANNED" if mode == "CREATE_PLANNED" else "NEW_LOCK_HELD" self.assertEqual(expected_phase, persisted["transaction_identity"]["takeover"]["phase"]) self.assertEqual(formal_before, refresh._identity(config.refresh.formal_manifest).as_dict()) refresh._recovery_formal_lock(config, persisted, collector.parse_now("2026-08-13T10:00:26+08:00")) recovered = refresh._load_pending(config) self.assertEqual("NEW_LOCK_HELD", recovered["transaction_identity"]["takeover"]["phase"]) self.assertEqual("HELD", recovered["transaction_identity"]["formal_lock_claim"]["claim_state"]) self.assertEqual(formal_before, refresh._identity(config.refresh.formal_manifest).as_dict()) refresh._release_formal_lock(config, recovered) refresh._pending_path(config).unlink() def test_runtime_v2_config_and_begin_bind_exact_contract(self) -> None: begin = self.begin() self.assertEqual("bili-supported-chrome-visible-runtime-v2", begin["runtime_contract_id"]) self.assertEqual(120, begin["overall_deadline_seconds"]) self.assertEqual(35, begin["refresh_action_timeout_seconds"]) self.assertEqual(45, begin["observation_timeout_seconds"]) self.assertEqual(15, begin["page_internal_settle_timeout_seconds"]) pending = refresh._load_pending(collector.load_config(self.config_path)) self.assertEqual(3, pending["schema_version"]) self.assertEqual(begin["runtime_contract_sha256"], pending["runtime_contract"]["contract_sha256"]) config = json.loads(self.config_path.read_text(encoding="utf-8")) config["refresh"]["timeout_seconds"] = 30 legacy = self.root / "legacy.json" self.write_json(legacy, config) with self.assertRaises(collector.CollectorError) as invalid: collector.load_config(legacy) self.assertEqual("E_CONFIG", invalid.exception.code) def test_runtime_pre_dispatch_error_is_zero_action_unreadable(self) -> None: begin = self.begin() evidence = self.evidence( begin, outcome="UNREADABLE_ERROR", observations=None, marker=None, action_outcome="PRE_DISPATCH_ERROR", observation_outcome="NOT_ATTEMPTED", refresh_count=0, ) self.set_runtime_diagnostic(evidence, "ACTION_PRE_DISPATCH") before = refresh._identity(self.archive / "manifest.jsonl").as_dict() code, result = self.commit(begin, evidence) self.assertEqual(4, code) self.assertEqual("REFRESH_FAILED_PAGE_UNREADABLE", result["status"]) self.assertEqual(0, result["refresh_count"]) self.assertEqual(before, refresh._identity(self.archive / "manifest.jsonl").as_dict()) def test_runtime_pre_dispatch_claimed_observation_fails_closed(self) -> None: for observation_outcome in ("READABLE", "ERROR"): with self.subTest(observation_outcome=observation_outcome): begin = self.begin() evidence = self.evidence( begin, outcome="UNREADABLE_ERROR", observations=None, marker=None, action_outcome="PRE_DISPATCH_ERROR", observation_outcome="NOT_ATTEMPTED", refresh_count=0, ) self.set_runtime_diagnostic(evidence, "ACTION_PRE_DISPATCH") evidence["runtime_observation"]["observation_outcome"] = observation_outcome evidence["runtime_observation"]["observation_count"] = 1 with self.assertRaises(collector.CollectorError) as invalid: self.validate_for_test(begin, evidence) self.assertEqual("E_EVIDENCE_SCHEMA", invalid.exception.code) refresh._pending_path(collector.load_config(self.config_path)).unlink() Path(str(begin["run_evidence_path"])).unlink() def test_runtime_post_dispatch_error_matrix(self) -> None: begin = self.begin() readable_empty = self.evidence( begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(), action_outcome="POST_DISPATCH_ERROR", ) self.set_runtime_diagnostic(readable_empty, "ACTION_POST_DISPATCH") code, result = self.commit(begin, readable_empty) self.assertEqual(4, code) self.assertEqual("PARTIAL_DISCOVERY_UNCONFIRMED", result["status"]) self.assertFalse(result["no_new_confirmed"]) def test_runtime_post_dispatch_read_error_and_deadline_exhausted(self) -> None: cases = ( ("ERROR", "UNREADABLE_ERROR", "ACTION_POST_DISPATCH", 1), ("DEADLINE_EXHAUSTED", "UNREADABLE_TIMEOUT", "ACTION_POST_DISPATCH", 0), ) for observation_outcome, page_outcome, code_name, observation_count in cases: with self.subTest(observation_outcome=observation_outcome): begin = self.begin() evidence = self.evidence( begin, outcome=page_outcome, observations=None, marker=None, action_outcome="POST_DISPATCH_ERROR", observation_outcome=observation_outcome, observation_count=observation_count, ) self.set_runtime_diagnostic(evidence, code_name) if observation_count == 0: evidence["runtime_observation"]["observation_elapsed_ms"] = 0 evidence["read_finished_at"] = evidence["refresh_finished_at"] code, result = self.commit(begin, evidence) self.assertEqual(4, code) self.assertEqual("REFRESH_FAILED_PAGE_UNREADABLE", result["status"]) if observation_outcome == "ERROR": Path(str(result["run_evidence_path"])).unlink() (self.state / "refresh" / "runs" / "latest.json").unlink() def test_runtime_timeout_cannot_confirm_no_new(self) -> None: begin = self.begin() evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(), action_outcome="TIMEOUT", ) self.set_runtime_diagnostic(evidence, "ACTION_TIMEOUT") code, result = self.commit(begin, evidence) self.assertEqual(4, code) self.assertEqual("PARTIAL_DISCOVERY_UNCONFIRMED", result["status"]) self.assertFalse(result["no_new_confirmed"]) def test_runtime_supported_control_latency_is_inside_new_budgets(self) -> None: begin = self.begin() evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(), ) evidence["refresh_finished_at"] = "2026-08-13T02:00:22.500000Z" evidence["read_finished_at"] = "2026-08-13T02:00:44Z" evidence["runtime_observation"]["refresh_action_elapsed_ms"] = 21500 evidence["runtime_observation"]["observation_elapsed_ms"] = 21500 evidence["controller_attestation"]["monotonic_action_finished_ms"] = 22500 evidence["controller_attestation"]["monotonic_observation_started_ms"] = 22500 evidence["controller_attestation"]["monotonic_observation_finished_ms"] = 44000 evidence["controller_attestation"]["monotonic_evidence_write_started_ms"] = 44000 code, result = self.commit(begin, evidence) self.assertEqual(0, code) self.assertEqual("REFRESH_CONFIRMED_NO_NEW", result["status"]) def test_runtime_post_dispatch_error_readable_new_saves_without_coverage(self) -> None: begin = self.begin() card, item = self.bound_card_item("987654") evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([card])], marker=self.end_marker(), items=[item], action_outcome="POST_DISPATCH_ERROR", ) self.set_runtime_diagnostic(evidence, "ACTION_POST_DISPATCH") code, result = self.commit(begin, evidence) self.assertEqual(0, code) self.assertEqual("NEW_ITEMS_SAVED", result["status"]) self.assertFalse(result["coverage_complete"]) def test_reviewer_trusted_controller_calls_once_and_caller_forgery_fails(self) -> None: input_stream = io.StringIO('{"page_authoritative":true,"no_new":true}\n') output_stream = io.StringIO() with self.assertRaises(collector.CollectorError) as failure: controller.run_product( collector.load_config(self.config_path), self.config_path, collector.parse_datetime(NOW, "now"), input_stream=input_stream, output_stream=output_stream, ) self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", failure.exception.code) self.assertEqual({"authoritative": False, "saved": False, "no_new": False}, failure.exception.details) self.assertEqual(0, input_stream.tell()) self.assertEqual("", output_stream.getvalue()) self.assertFalse((self.state / "refresh" / "pending.json").exists()) def test_reviewer_real_js_fixture_adapts_to_accepted_schema(self) -> None: runner = PROJECT_DEV / "test" / "fixtures" / "bili_dynamic_collector" / "page_extract_fixture_runner.js" for fixture_case in ("new", "empty", "unparsed", "identity", "access"): with self.subTest(fixture_case=fixture_case): self.reset_runtime_fixture() raw = None if fixture_case != "access": completed = subprocess.run( ["node", str(runner), str(refresh.EXTRACTOR_SOURCE), fixture_case], capture_output=True, check=True, timeout=10, ) raw = json.loads(completed.stdout.decode("utf-8")) def response(request_id: int, result: object, *, error: str | None = None) -> str: return json.dumps({ "schema_version": 1, "type": "supported_chrome_response", "request_id": request_id, "ok": error is None, "result": result, "error_code": error, }, ensure_ascii=False) protocol_input = "\n".join([ response(1, [{"tab_id": "fixture-tab", "url": "https://space.bilibili.com/1420210197/dynamic"}]), response(2, None), response(3, raw, error="ACCESS_BLOCKED" if fixture_case == "access" else None), ]) + "\n" completed = subprocess.run( [ sys.executable, "-B", str(COLLECTOR_PATH), "--config", str(self.config_path), "refresh-run", "--now", NOW, ], input=protocol_input, text=True, capture_output=True, timeout=20, ) lines = [json.loads(line) for line in completed.stdout.splitlines()] requests = [line for line in lines if line.get("type") == "supported_chrome_request"] terminal = lines[-1] self.assertEqual([], requests) self.assertEqual((3, "SAFETY_STOP", "E_TRUSTED_ADAPTER_REQUIRED"), (completed.returncode, terminal["status"], terminal["error_code"])) self.assertEqual({"authoritative": False, "saved": False, "no_new": False}, terminal["details"]) self.assertFalse((self.state / "refresh" / "pending.json").exists()) def test_reviewer_public_cli_cannot_create_or_bind_caller_schema3(self) -> None: self.assertFalse(hasattr(refresh, "_attest_controller_evidence")) self.assertFalse(hasattr(refresh, "bind_controller_evidence")) self.assertFalse(hasattr(controller, "_attest_controller_evidence")) self.assertFalse(hasattr(controller, "bind_controller_evidence")) with self.assertRaises(collector.CollectorError) as blocked: collector.run(["--config", str(self.config_path), "refresh-begin", "--now", NOW]) self.assertEqual("E_CONTROLLER_ENTRY_REQUIRED", blocked.exception.code) config = collector.load_config(self.config_path) self.assertFalse(refresh._pending_path(config).exists()) begin = self.begin() pending = refresh._load_pending(config) self.assertNotIn("controller_capability", pending) self.assertRegex(str(pending["controller_key_commitment"]), r"^[0-9a-f]{64}$") forged = self.evidence( begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker() ) path = Path(str(begin["evidence_path"])) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(forged, ensure_ascii=False), encoding="utf-8") with self.assertRaises(collector.CollectorError) as rejected: collector.run([ "--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00", ]) self.assertEqual("E_CONTROLLER_REQUIRED", rejected.exception.code) self.assertFalse((self.archive / "manifest.jsonl").exists()) def test_reviewer_expired_pending_is_zero_adapter_calls_and_zero_evidence(self) -> None: begin = self.begin() protocol_input = io.StringIO('{"page_authoritative":true}\n') protocol_output = io.StringIO() with self.assertRaises(collector.CollectorError) as expired: controller.run_product( collector.load_config(self.config_path), self.config_path, collector.parse_datetime("2026-08-13T10:02:00.001+08:00", "wall"), input_stream=protocol_input, output_stream=protocol_output, ) self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", expired.exception.code) self.assertEqual(0, protocol_input.tell()) self.assertEqual("", protocol_output.getvalue()) self.assertFalse(Path(str(begin["evidence_path"])).exists()) pending = refresh._load_pending(collector.load_config(self.config_path)) self.assertEqual("AWAITING_EVIDENCE", pending["phase"]) def test_reviewer_deadline_plus_epsilon_never_binds_or_saves(self) -> None: begin = self.begin() evidence = self.evidence(begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker()) path = Path(str(begin["evidence_path"])) self.write_json(path, evidence) before = refresh._identity(self.archive / "manifest.jsonl").as_dict() code, result = collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:02:01+08:00"]) self.assertEqual((4, "E_OVERALL_DEADLINE"), (code, result["error_code"])) self.assertEqual(before, refresh._identity(self.archive / "manifest.jsonl").as_dict()) def test_reviewer_controller_commit_crossing_deadline_removes_unbound_evidence(self) -> None: begin = self.begin() with self.assertRaises(collector.CollectorError) as failure: controller.run_product( collector.load_config(self.config_path), self.config_path, collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"), ) self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", failure.exception.code) self.assertFalse(Path(str(begin["evidence_path"])).exists()) pending = refresh._load_pending(collector.load_config(self.config_path)) self.assertIsNotNone(pending) self.assertEqual("AWAITING_EVIDENCE", pending["phase"]) self.assertFalse((self.archive / "manifest.jsonl").exists()) def test_reviewer_legacy_pending_never_requests_browser_action(self) -> None: for slot_present in (False, True): for after_deadline in (False, True): for entrypoint in ("begin", "commit", "controller"): with self.subTest(slot_present=slot_present, after_deadline=after_deadline, entrypoint=entrypoint): self.reset_runtime_fixture() begin = self.begin() config = collector.load_config(self.config_path) pending = refresh._load_pending(config) assert pending is not None pending["schema_version"] = refresh.LEGACY_PENDING_SCHEMA pending.pop("runtime_contract") pending.pop("controller_key_commitment") pending.pop("controller_binding_sha256") refresh._write_pending(config, pending) if not slot_present: Path(str(begin["run_evidence_path"])).unlink() now_text = "2026-08-13T10:02:01+08:00" if after_deadline else "2026-08-13T10:00:30+08:00" if entrypoint == "controller": with self.assertRaises(collector.CollectorError) as failure: controller.run_product(config, self.config_path, collector.parse_datetime(now_text, "now")) self.assertEqual("E_TRUSTED_ADAPTER_REQUIRED", failure.exception.code) continue if entrypoint == "begin": invoke = lambda: refresh.refresh_begin(config, self.config_path, collector.parse_datetime(now_text, "now")) else: invoke = lambda: refresh.refresh_commit(config, self.config_path, Path(str(begin["evidence_path"])), collector.parse_datetime(now_text, "now")) if after_deadline: result = invoke() self.assertEqual(("REFRESH_FAILED_PAGE_UNREADABLE", "E_LEGACY_EVIDENCE_MISSING_AFTER_DEADLINE"), (result["status"], result["error_code"])) self.assertNotEqual("BROWSER_REFRESH_REQUIRED", result["status"]) else: with self.assertRaises(collector.CollectorError) as failure: invoke() self.assertEqual("E_LEGACY_RECOVERY_ONLY", failure.exception.code) self.assertFalse((self.archive / "manifest.jsonl").exists()) def test_reviewer_identity_mismatch_is_durable_blocked_terminal(self) -> None: identities = { "uid": lambda evidence: evidence["creator"].__setitem__("uid", "1"), "name": lambda evidence: evidence["creator"].__setitem__("name", "其他用户"), "profile": lambda evidence: evidence["creator"].__setitem__("profile_url", "https://space.bilibili.com/1"), "final_url": lambda evidence: evidence.__setitem__("final_url", "https://space.bilibili.com/1/dynamic"), } outcomes = { "CONFIRMED": "NONE", "TIMEOUT": "ACTION_TIMEOUT", "POST_DISPATCH_ERROR": "ACTION_POST_DISPATCH", } for identity_name, mutate in identities.items(): for action_outcome, diagnostic in outcomes.items(): with self.subTest(identity=identity_name, action_outcome=action_outcome): self.reset_runtime_fixture() begin = self.begin() evidence = self.evidence( begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker(), action_outcome=action_outcome, ) self.set_runtime_diagnostic(evidence, diagnostic) mutate(evidence) code, result = self.commit(begin, evidence) self.assertEqual((3, "REFRESH_BLOCKED_AUTH_OR_ACCESS", "E_CREATOR_MISMATCH"), (code, result["status"], result["error_code"])) slot = json.loads(Path(str(result["run_evidence_path"])).read_text(encoding="utf-8")) self.assertEqual("TERMINAL", slot["run_state"]) self.assertFalse(refresh._pending_path(collector.load_config(self.config_path)).exists()) self.assertFalse(result["no_new_confirmed"]) self.assertFalse((self.archive / "manifest.jsonl").exists()) if __name__ == "__main__": unittest.main()