chore(project-info): archive Bilibili dynamic b292f50b994e

MB-X Bilibili Pipeline

6 days ago 643d038b717c97958e7e9dac25fb67d56645f12a

 dev/project-dev/test/test_bili_dynamic_refresh.py

@@ -1,11 +1,14 @@
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
@@ -42,6 +45,7 @@
        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,
            {
@@ -93,20 +97,7 @@

    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):
            config = collector.load_config(self.config_path)
            pending = refresh._load_pending(config)
            if pending is not None:
                attestation = value["controller_attestation"]
                refresh._attest_controller_evidence(
                    pending, value,
                    action_dispatched=bool(attestation["action_dispatched"]),
                    monotonic_run_started_ms=int(attestation["monotonic_run_started_ms"]),
                    monotonic_action_started_ms=attestation["monotonic_action_started_ms"],
                    monotonic_action_finished_ms=attestation["monotonic_action_finished_ms"],
                    monotonic_observation_started_ms=attestation["monotonic_observation_started_ms"],
                    monotonic_observation_finished_ms=attestation["monotonic_observation_finished_ms"],
                    monotonic_evidence_write_started_ms=int(attestation["monotonic_evidence_write_started_ms"]),
                )
            self.sign_evidence(value)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")

@@ -116,12 +107,30 @@
            if path.exists():
                shutil.rmtree(path)
            path.mkdir()
        self.controller_keys.clear()

    def begin(self) -> dict[str, object]:
        code, result = collector.run(["--config", str(self.config_path), "refresh-begin", "--now", NOW])
        self.assertEqual(0, code)
        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 {
@@ -244,9 +253,6 @@
                "observation_count": observation_count,
            },
        }
        config = collector.load_config(self.config_path)
        pending = refresh._load_pending(config)
        assert pending is not None
        if action_outcome == "PRE_DISPATCH_ERROR":
            action_started = action_finished = observation_started = observation_finished = None
        else:
@@ -262,17 +268,22 @@
            action_finished or 0,
            observation_finished or 0,
        )
        refresh._attest_controller_evidence(
            pending,
            evidence,
            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,
        )
        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:
@@ -297,33 +308,36 @@
        }

    def commit(self, begin: dict[str, object], evidence: dict[str, object]) -> tuple[int, dict[str, object]]:
        if "controller_attestation" in evidence:
            config = collector.load_config(self.config_path)
            pending = refresh._load_pending(config)
            assert pending is not None
            attestation = evidence["controller_attestation"]
            assert isinstance(attestation, dict)
            refresh._attest_controller_evidence(
                pending,
                evidence,
                action_dispatched=bool(attestation["action_dispatched"]),
                monotonic_run_started_ms=int(attestation["monotonic_run_started_ms"]),
                monotonic_action_started_ms=attestation["monotonic_action_started_ms"],
                monotonic_action_finished_ms=attestation["monotonic_action_finished_ms"],
                monotonic_observation_started_ms=attestation["monotonic_observation_started_ms"],
                monotonic_observation_finished_ms=attestation["monotonic_observation_finished_ms"],
                monotonic_evidence_write_started_ms=int(attestation["monotonic_evidence_write_started_ms"]),
            )
        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
        refresh.bind_controller_evidence(
            config, pending, path,
            transitioned_at=collector.parse_datetime("2026-08-13T10:00:24+08:00", "test transitioned_at"),
        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"])]
        )
        return collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])

    @staticmethod
    def bound_card_item(identifier: str = "456") -> tuple[dict[str, object], dict[str, object]]:
@@ -447,10 +461,10 @@
        }
        node = {"position": 0, "node_fingerprint_sha256": "b" * 64, "reason_code": "PARSER_REJECTED"}
        observation = self.observation([card], [node])
        path = Path(str(begin["evidence_path"]))
        self.write_json(path, self.evidence(begin, outcome="READABLE", observations=[observation], marker=None))
        with self.assertRaisesRegex(collector.CollectorError, "positions overlap"):
            collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
            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:
@@ -460,10 +474,8 @@
            begin, outcome="READABLE", observations=[self.observation([card])],
            marker=self.end_marker(), items=[],
        )
        path = Path(str(begin["evidence_path"]))
        self.write_json(path, evidence)
        with self.assertRaises(collector.CollectorError) as failure:
            collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
            self.validate_for_test(begin, evidence)
        self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
        self.assertFalse((self.archive / "manifest.jsonl").exists())

@@ -474,10 +486,8 @@
            begin, outcome="READABLE", observations=[self.observation([])],
            marker=self.end_marker(), items=[item],
        )
        path = Path(str(begin["evidence_path"]))
        self.write_json(path, evidence)
        with self.assertRaises(collector.CollectorError) as failure:
            collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
            self.validate_for_test(begin, evidence)
        self.assertEqual("E_EVIDENCE_ITEM_BINDING", failure.exception.code)
        self.assertFalse((self.archive / "manifest.jsonl").exists())

@@ -489,10 +499,8 @@
            begin, outcome="READABLE", observations=[self.observation([card])],
            marker=self.end_marker(), items=[item],
        )
        path = Path(str(begin["evidence_path"]))
        self.write_json(path, evidence)
        with self.assertRaises(collector.CollectorError) as failure:
            collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
            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:
@@ -510,10 +518,8 @@
            begin, outcome="READABLE", observations=[self.observation([card, other_card])],
            marker=self.end_marker(), items=[item, other_item],
        )
        path = Path(str(begin["evidence_path"]))
        self.write_json(path, evidence)
        with self.assertRaises(collector.CollectorError) as failure:
            collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
            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:
@@ -565,24 +571,23 @@
        self.assertEqual(4, code)
        self.assertEqual("E_EVIDENCE_MISSING_AFTER_DEADLINE", terminal["error_code"])
        with self.assertRaises(collector.CollectorError) as occupied:
            collector.run(["--config", str(self.config_path), "refresh-begin", "--now", "2026-08-13T10:02:10+08:00"])
            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)
        config = collector.load_config(self.config_path)
        pending = refresh._load_pending(config)
        assert pending is not None
        refresh.bind_controller_evidence(
            config, pending, path,
            transitioned_at=collector.parse_datetime("2026-08-13T10:00:24+08:00", "test transitioned_at"),
        )
        self.bind_for_test(begin, evidence)
        original = refresh._write_readback
        injected = {"done": False}

@@ -628,7 +633,7 @@
        self.assertTrue(config.lock_path.is_file())
        self.assertFalse((config.state_dir / ".collector.lock.owner.json").exists())

    def test_frozen_formal_catalog_shape_is_48_22_16_5_1_read_only(self) -> None:
    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")
@@ -648,7 +653,7 @@
            ),
        )
        _, _, counts = refresh.load_formal_catalog(rebound)
        self.assertEqual({"events": 48, "components": 22, "saved": 16, "video": 5, "retryable": 1}, counts)
        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:
@@ -723,7 +728,10 @@
        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))
                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)
@@ -1174,10 +1182,8 @@
                self.set_runtime_diagnostic(evidence, "ACTION_PRE_DISPATCH")
                evidence["runtime_observation"]["observation_outcome"] = observation_outcome
                evidence["runtime_observation"]["observation_count"] = 1
                path = Path(str(begin["evidence_path"]))
                self.write_json(path, evidence)
                with self.assertRaises(collector.CollectorError) as invalid:
                    collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
                    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()
@@ -1261,127 +1267,109 @@
        self.assertFalse(result["coverage_complete"])

    def test_reviewer_trusted_controller_calls_once_and_caller_forgery_fails(self) -> None:
        begin = self.begin()
        calls = {"tabs": 0, "reload": 0, "goto": 0, "evaluate": 0}

        class Fake:
            def open_tabs(inner) -> list[dict[str, object]]:
                calls["tabs"] += 1
                return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]

            def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
                calls["reload"] += 1

            def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
                calls["goto"] += 1
                return {"url": url}

            def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
                calls["evaluate"] += 1
                return {
                    "schema_version": 1,
                    "ready_state": "complete", "visibility_state": "visible",
                    "final_url": "https://space.bilibili.com/1420210197/dynamic",
                    "page_title": "青枫浦上Q个人动态-青枫浦上Q动态记录-哔哩哔哩视频",
                    "creator": {"uid": "1420210197", "name": "青枫浦上Q", "profile_url": "https://space.bilibili.com/1420210197"},
                    "cards": [], "unparsed_nodes": [], "terminal_marker_text": "已经到底了", "limit_hit": "NONE",
                }

        ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 0.023, 0.024])
        path = controller.run_once(
            collector.load_config(self.config_path), begin, Fake(), monotonic=lambda: next(ticks),
            wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
        )
        self.assertEqual({"tabs": 1, "reload": 1, "goto": 0, "evaluate": 1}, calls)
        code, result = collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
        self.assertEqual((0, "REFRESH_CONFIRMED_NO_NEW"), (code, result["status"]))

        Path(str(result["run_evidence_path"])).unlink()
        (self.state / "refresh" / "runs" / "latest.json").unlink()
        begin = self.begin()
        forged = self.evidence(begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker())
        forged["controller_attestation"]["binding_sha256"] = "0" * 64
        path = Path(str(begin["evidence_path"]))
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(forged), encoding="utf-8")
        input_stream = io.StringIO('{"page_authoritative":true,"no_new":true}\n')
        output_stream = io.StringIO()
        with self.assertRaises(collector.CollectorError) as failure:
            collector.run(["--config", str(self.config_path), "refresh-commit", "--input", str(path), "--now", "2026-08-13T10:00:25+08:00"])
        self.assertIn(failure.exception.code, {"E_CONTROLLER_ATTESTATION", "E_CONTROLLER_REQUIRED"})
            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"
        expected = {
            "new": (0, "NEW_ITEMS_SAVED", None),
            "empty": (0, "REFRESH_CONFIRMED_NO_NEW", None),
            "unparsed": (4, "PARTIAL_DISCOVERY_UNCONFIRMED", "E_COVERAGE_INCOMPLETE"),
            "identity": (3, "REFRESH_BLOCKED_AUTH_OR_ACCESS", "E_CREATOR_MISMATCH"),
        }
        for fixture_case, wanted in expected.items():
        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(
                    ["node", str(runner), str(refresh.EXTRACTOR_SOURCE), fixture_case],
                    capture_output=True, check=True, timeout=10,
                    [
                        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,
                )
                raw = json.loads(completed.stdout.decode("utf-8"))
                begin = self.begin()
                calls = {"reload": 0, "evaluate": 0}
                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())

                class Fake:
                    def open_tabs(inner) -> list[dict[str, object]]:
                        return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]
    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())

                    def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
                        calls["reload"] += 1

                    def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
                        raise AssertionError("exact fixture tab must use reload")

                    def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
                        calls["evaluate"] += 1
                        self.assertIn("projectInfoCollectVisibleDynamicNodes({page_internal_settle_timeout_ms:15000})", source)
                        return raw

                ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 0.023, 0.024])
                path = controller.run_once(
                    collector.load_config(self.config_path), begin, Fake(),
                    monotonic=lambda: next(ticks),
                    wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
                )
                code, result = collector.run([
                    "--config", str(self.config_path), "refresh-commit", "--input", str(path),
                    "--now", "2026-08-13T10:00:25+08:00",
                ])
                self.assertEqual(wanted, (code, result["status"], result["error_code"]))
                self.assertEqual({"reload": 1, "evaluate": 1}, calls)

        self.reset_runtime_fixture()
        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}$")

        class AccessBlocked:
            def open_tabs(inner) -> list[dict[str, object]]:
                return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]

            def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
                return None

            def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
                raise AssertionError("exact fixture tab must use reload")

            def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
                raise PermissionError("synthetic access interstitial")

        ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 0.023, 0.024])
        path = controller.run_once(
            collector.load_config(self.config_path), begin, AccessBlocked(),
            monotonic=lambda: next(ticks),
            wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
        forged = self.evidence(
            begin, outcome="READABLE", observations=[self.observation([])], marker=self.end_marker()
        )
        code, result = collector.run([
            "--config", str(self.config_path), "refresh-commit", "--input", str(path),
            "--now", "2026-08-13T10:00:25+08:00",
        ])
        self.assertEqual((3, "REFRESH_BLOCKED_AUTH_OR_ACCESS", "E_ACCESS_BLOCKED"), (code, result["status"], result["error_code"]))
        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()
@@ -1395,34 +1383,12 @@

    def test_reviewer_controller_commit_crossing_deadline_removes_unbound_evidence(self) -> None:
        begin = self.begin()

        class Fake:
            def open_tabs(inner) -> list[dict[str, object]]:
                return [{"url": "https://space.bilibili.com/1420210197/dynamic"}]

            def reload(inner, tab: dict[str, object], timeout_seconds: int) -> None:
                return None

            def goto(inner, url: str, timeout_seconds: int) -> dict[str, object]:
                raise AssertionError("exact fixture tab must use reload")

            def evaluate(inner, tab: dict[str, object], source: str, timeout_seconds: int) -> dict[str, object]:
                return {
                    "schema_version": 1, "ready_state": "complete", "visibility_state": "visible",
                    "final_url": "https://space.bilibili.com/1420210197/dynamic",
                    "page_title": "青枫浦上Q个人动态-青枫浦上Q动态记录-哔哩哔哩视频",
                    "creator": {"uid": "1420210197", "name": "青枫浦上Q", "profile_url": "https://space.bilibili.com/1420210197"},
                    "cards": [], "unparsed_nodes": [], "terminal_marker_text": "已经到底了", "limit_hit": "NONE",
                }

        ticks = iter([0.0, 0.001, 0.011, 0.012, 0.022, 119.999, 120.001])
        with self.assertRaises(collector.CollectorError) as failure:
            controller.run_once(
                collector.load_config(self.config_path), begin, Fake(),
                monotonic=lambda: next(ticks),
                wall_now=lambda: collector.parse_datetime("2026-08-13T10:00:24+08:00", "wall"),
            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_OVERALL_DEADLINE", failure.exception.code)
        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)
@@ -1441,19 +1407,16 @@
                        assert pending is not None
                        pending["schema_version"] = refresh.LEGACY_PENDING_SCHEMA
                        pending.pop("runtime_contract")
                        pending.pop("controller_capability")
                        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":
                            class NeverCalled:
                                def open_tabs(inner) -> list[dict[str, object]]:
                                    raise AssertionError("legacy pending must not touch the browser adapter")

                            with self.assertRaises(collector.CollectorError) as failure:
                                controller.run_once(config, begin, NeverCalled())
                            self.assertEqual("E_CONTROLLER_PENDING", failure.exception.code)
                                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"))