from __future__ import annotations

import hashlib
import importlib.util
import io
import json
import os
import sys
import tempfile
import unittest
from contextlib import redirect_stdout
from datetime import datetime
from pathlib import Path
from unittest import mock


MODULE_PATH = Path(__file__).resolve().parents[1] / "bili_dynamic_collector.py"
SPEC = importlib.util.spec_from_file_location("bili_dynamic_collector", MODULE_PATH)
assert SPEC and SPEC.loader
collector = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = collector
SPEC.loader.exec_module(collector)

FIXTURE = Path(__file__).resolve().parent / "fixtures" / "bili_dynamic_collector" / "dynamic_items.json"
NOW = "2026-08-04T12:00:00+08:00"


class BiliDynamicCollectorTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        self.config_path = self.root / "config.json"
        self.state_dir = self.root / "state"
        self.download_dir = self.root / "downloads"
        self.video_dir = self.root / "videos"
        self.download_dir.mkdir()
        self.video_dir.mkdir()
        self.write_json(
            self.config_path,
            {
                "schema_version": 1,
                "creator": {
                    "name": "青枫浦上Q",
                    "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",
                },
            },
        )

    def tearDown(self) -> None:
        self.temp.cleanup()

    @staticmethod
    def write_json(path: Path, value: object) -> None:
        path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")

    def run_check(self) -> tuple[int, dict[str, object]]:
        return collector.run(
            ["--config", str(self.config_path), "check", "--input", str(FIXTURE), "--now", NOW]
        )

    def run_main_json(self, argv: list[str]) -> tuple[int, dict[str, object], str]:
        stream = io.StringIO()
        with redirect_stdout(stream):
            code = collector.main(argv)
        text = stream.getvalue()
        return code, json.loads(text), text

    def queue_video(self) -> dict[str, object]:
        code, result = self.run_check()
        self.assertEqual(0, code)
        self.assertEqual("TODO_GENERATED", result["status"])
        manifest = collector.load_manifest(self.state_dir / "manifest.jsonl")
        return next(event for event in manifest if event["content_type"] == "video")

    def test_windows_name_is_safe_and_preserves_timestamp(self) -> None:
        self.assertEqual("_CON", collector.sanitize_windows_component("CON", 48))
        self.assertEqual("a_b_c", collector.sanitize_windows_component('a<b>c. ', 48))
        config = collector.load_config(self.config_path)
        item = collector.normalize_item(
            {
                "dynamic_id": "d1",
                "content_type": "text",
                "published_at": "2026-08-04T10:00:00+08:00",
                "title": "午盘: A/B?",
                "source_url": "https://www.bilibili.com/opus/d1",
            },
            config,
            0,
        )
        self.assertEqual("20260804-100000_文字_午盘_ A_B_", collector.suggested_base(item, config))

    def test_check_filters_window_and_is_idempotent(self) -> None:
        code, first = self.run_check()
        self.assertEqual(0, code)
        self.assertEqual("TODO_GENERATED", first["status"])
        self.assertEqual(2, first["new_items"])
        self.assertEqual(1, first["skipped_old"])
        self.assertEqual(1, first["skipped_future"])
        queue_path = Path(first["queue_path"])
        self.assertTrue(queue_path.is_file())
        queue = json.loads(queue_path.read_text(encoding="utf-8"))
        self.assertEqual(2, len(queue["items"]))
        self.assertTrue(queue["browser_interaction_required"])
        self.assertFalse(queue["authentication_data_allowed"])
        self.assertEqual(2, len(collector.load_manifest(self.state_dir / "manifest.jsonl")))

        code, second = self.run_check()
        self.assertEqual(0, code)
        self.assertEqual("NO_NEW_ITEMS", second["status"])
        self.assertEqual(0, second["new_items"])
        self.assertIsNone(second["queue_path"])
        self.assertEqual([queue_path], list((self.state_dir / "queues").glob("*.json")))
        self.assertEqual(2, len(collector.load_manifest(self.state_dir / "manifest.jsonl")))

    def test_safe_move_hash_manifest_and_handoff_are_idempotent(self) -> None:
        queued = self.queue_video()
        source = self.download_dir / "extension-result.mp4"
        payload = b"small synthetic completed video fixture"
        source.write_bytes(payload)
        completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
        os.utime(source, (completed_mtime, completed_mtime))
        mapping = self.root / "mapping.json"
        self.write_json(
            mapping,
            {
                "schema_version": 1,
                "items": [{"source_file": source.name, "bvid": "BV1TEST00001"}],
            },
        )

        code, moved = collector.run(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(0, code)
        self.assertEqual("COMPLETE", moved["status"])
        self.assertFalse(source.exists())
        target = Path(moved["items"][0]["target"])
        self.assertEqual(f"{queued['suggested_stem']}.mp4", target.name)
        self.assertEqual(payload, target.read_bytes())
        expected_hash = hashlib.sha256(payload).hexdigest()
        self.assertEqual(expected_hash, moved["items"][0]["sha256"])
        latest, _ = collector.latest_entities(collector.load_manifest(self.state_dir / "manifest.jsonl"))
        event = latest[queued["entity_id"]]
        self.assertEqual("VIDEO_MOVED", event["status"])
        self.assertEqual("READY_FOR_HANDOFF", event["video_processing_status"])
        self.assertEqual(expected_hash, event["sha256"])

        manifest_before_rerun = (self.state_dir / "manifest.jsonl").read_bytes()
        code, moved_again = collector.run(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(0, code)
        self.assertEqual("ALREADY_MOVED", moved_again["items"][0]["status"])
        self.assertEqual(expected_hash, moved_again["items"][0]["sha256"])
        self.assertEqual(manifest_before_rerun, (self.state_dir / "manifest.jsonl").read_bytes())
        self.assertEqual(payload, target.read_bytes())

        code, handoff = collector.run(
            ["--config", str(self.config_path), "handoff", "--now", NOW]
        )
        self.assertEqual(0, code)
        self.assertEqual("GENERATED", handoff["write"])
        draft = Path(handoff["handoff_path"])
        text = draft.read_text(encoding="utf-8")
        self.assertIn("message_type=video_processing_request", text)
        self.assertIn("status=PROCESSING_REQUESTED", text)
        self.assertRegex(text, r"handoff_id=HANDOFF-BILI-DYNAMIC-VIDEO-PROCESSING-[0-9A-F]{24}")
        for field in (
            "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",
            "scope:",
            "evidence:",
            "expected_action:",
        ):
            self.assertIn(field, text)
        self.assertNotIn("generated_at=", text)
        self.assertIn(json.dumps(str(target), ensure_ascii=False), text)
        self.assertNotIn("cookie", text.lower())

        code, repeated = collector.run(
            ["--config", str(self.config_path), "handoff", "--now", "2026-08-04T13:00:00+08:00"]
        )
        self.assertEqual(0, code)
        self.assertEqual("REUSED", repeated["write"])
        self.assertEqual(draft, Path(repeated["handoff_path"]))

        target.write_bytes(b"tampered after manifest")
        config = collector.load_config(self.config_path)
        with collector.StateLock(config.lock_path):
            with self.assertRaises(collector.CollectorError) as caught:
                collector.generate_handoff(config, None, collector.parse_now(NOW))
        self.assertEqual("E_HANDOFF_HASH", caught.exception.code)

    def test_existing_target_preflight_has_zero_business_side_effects(self) -> None:
        queued = self.queue_video()
        source = self.download_dir / "extension-result.mp4"
        source.write_bytes(b"new download")
        completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
        os.utime(source, (completed_mtime, completed_mtime))
        target = self.video_dir / f"{queued['suggested_stem']}.mp4"
        target.write_bytes(b"existing unrelated file")
        mapping = self.root / "mapping.json"
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": source.name, "dynamic_id": "dynamic-video-001"}]},
        )
        manifest_before = (self.state_dir / "manifest.jsonl").read_bytes()
        config = collector.load_config(self.config_path)
        with collector.StateLock(config.lock_path):
            with self.assertRaisesRegex(collector.CollectorError, "refusing to overwrite") as caught:
                collector.move_completed(config, mapping, collector.parse_now(NOW))
        self.assertEqual("E_TARGET_EXISTS", caught.exception.code)
        self.assertEqual(b"existing unrelated file", target.read_bytes())
        self.assertEqual(b"new download", source.read_bytes())
        self.assertEqual(manifest_before, (self.state_dir / "manifest.jsonl").read_bytes())
        latest, _ = collector.latest_entities(collector.load_manifest(self.state_dir / "manifest.jsonl"))
        self.assertEqual("TODO_QUEUED", latest[queued["entity_id"]]["status"])

        code, retry = self.run_check()
        self.assertEqual(0, code)
        self.assertEqual("NO_NEW_ITEMS", retry["status"])

    def test_incomplete_or_outside_download_is_a_safety_stop(self) -> None:
        queued = self.queue_video()
        incomplete = self.download_dir / "clip.part"
        incomplete.write_bytes(b"partial")
        mapping = self.root / "mapping.json"
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": incomplete.name, "bvid": queued["bvid"]}]},
        )
        code = collector.main(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(3, code)
        self.assertTrue(incomplete.exists())

        outside = self.root / "outside.mp4"
        outside.write_bytes(b"outside")
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": str(outside), "bvid": queued["bvid"]}]},
        )
        code = collector.main(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(3, code)
        self.assertTrue(outside.exists())

    def test_exact_selectors_and_batch_uniqueness_fail_before_business_writes(self) -> None:
        queued = self.queue_video()
        events = collector.load_manifest(self.state_dir / "manifest.jsonl")
        text_event = next(event for event in events if event["content_type"] == "text")
        source = self.download_dir / "one.mp4"
        source.write_bytes(b"one physical source")
        completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
        os.utime(source, (completed_mtime, completed_mtime))
        manifest_path = self.state_dir / "manifest.jsonl"

        cases = [
            (
                "unknown supplied selector",
                [{"source_file": source.name, "bvid": queued["bvid"], "dynamic_id": "unknown-dynamic"}],
                "E_MAPPING_SELECTOR_UNKNOWN",
            ),
            (
                "selectors disagree",
                [
                    {
                        "source_file": source.name,
                        "bvid": queued["bvid"],
                        "dynamic_id": text_event["dynamic_id"],
                    }
                ],
                "E_MAPPING_SELECTOR_CONFLICT",
            ),
            (
                "entity repeated",
                [
                    {"source_file": source.name, "bvid": queued["bvid"]},
                    {"source_file": source.name, "dynamic_id": queued["dynamic_id"]},
                ],
                "E_ENTITY_DUPLICATE",
            ),
        ]
        for label, items, expected_code in cases:
            with self.subTest(label=label):
                mapping = self.root / "mapping.json"
                self.write_json(mapping, {"schema_version": 1, "items": items})
                manifest_before = manifest_path.read_bytes()
                code, payload, _ = self.run_main_json(
                    ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
                )
                self.assertEqual(3, code)
                self.assertEqual(expected_code, payload["error_code"])
                self.assertEqual(manifest_before, manifest_path.read_bytes())
                self.assertEqual(b"one physical source", source.read_bytes())
                self.assertEqual([], list(self.video_dir.iterdir()))

        second = dict(queued)
        second.update(
            {
                "event_id": "second-video-event",
                "entity_id": "second-video-entity",
                "dynamic_id": "dynamic-video-002",
                "bvid": "BV1TEST00002",
                "source_url": "https://www.bilibili.com/video/BV1TEST00002",
                "dedupe_keys": [
                    "bvid:bv1test00002",
                    "dynamic:dynamic-video-002",
                    "url:https://www.bilibili.com/video/BV1TEST00002",
                ],
                "suggested_stem": str(queued["suggested_stem"]) + "_second",
            }
        )
        collector.append_manifest(manifest_path, [second])
        mapping = self.root / "same-source.json"
        self.write_json(
            mapping,
            {
                "schema_version": 1,
                "items": [
                    {"source_file": source.name, "bvid": queued["bvid"]},
                    {"source_file": str(source), "bvid": second["bvid"]},
                ],
            },
        )
        manifest_before = manifest_path.read_bytes()
        code, payload, _ = self.run_main_json(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(3, code)
        self.assertEqual("E_SOURCE_DUPLICATE", payload["error_code"])
        self.assertEqual(manifest_before, manifest_path.read_bytes())
        self.assertEqual(b"one physical source", source.read_bytes())
        self.assertEqual([], list(self.video_dir.iterdir()))

    def test_retained_source_warning_is_sticky_with_or_without_source(self) -> None:
        queued = self.queue_video()
        source = self.download_dir / "retained.mp4"
        payload_bytes = b"retained synthetic bytes"
        source.write_bytes(payload_bytes)
        completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
        os.utime(source, (completed_mtime, completed_mtime))
        mapping = self.root / "mapping.json"
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
        )
        path_class = type(source)
        original_unlink = path_class.unlink

        def fail_only_source(path: Path, *args: object, **kwargs: object) -> None:
            if path == source:
                raise PermissionError("injected source delete failure")
            original_unlink(path, *args, **kwargs)

        with mock.patch.object(path_class, "unlink", fail_only_source):
            code, first = collector.run(
                ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
            )
        self.assertEqual(4, code)
        self.assertEqual("VIDEO_MOVED_SOURCE_RETAINED", first["items"][0]["status"])
        target = Path(first["items"][0]["target"])
        expected_hash = hashlib.sha256(payload_bytes).hexdigest()
        self.assertEqual(expected_hash, first["items"][0]["sha256"])
        self.assertTrue(source.exists())
        manifest_before = (self.state_dir / "manifest.jsonl").read_bytes()

        code, second = collector.run(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(4, code)
        self.assertEqual("VIDEO_MOVED_SOURCE_RETAINED", second["items"][0]["status"])
        self.assertEqual(first["items"][0]["source_delete_error"], second["items"][0]["source_delete_error"])
        self.assertEqual(manifest_before, (self.state_dir / "manifest.jsonl").read_bytes())
        self.assertEqual(payload_bytes, target.read_bytes())

        source.unlink()
        code, third = collector.run(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(4, code)
        self.assertEqual("VIDEO_MOVED_SOURCE_RETAINED", third["items"][0]["status"])
        self.assertEqual(expected_hash, third["items"][0]["sha256"])
        self.assertEqual(first["items"][0]["source_delete_error"], third["items"][0]["source_delete_error"])
        self.assertEqual(manifest_before, (self.state_dir / "manifest.jsonl").read_bytes())
        self.assertEqual(payload_bytes, target.read_bytes())

    def test_disallowed_latest_statuses_fail_before_stage_b(self) -> None:
        queued = self.queue_video()
        source = self.download_dir / "sentinel.mp4"
        source.write_bytes(b"sentinel source")
        mapping = self.root / "mapping.json"
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
        )
        manifest_path = self.state_dir / "manifest.jsonl"
        status_cases: list[object] = [
            "PROCESSING_HANDOFF_CONFIRMED",
            "PROCESSING",
            "PROCESSING_FAILED",
            None,
            "",
            "UNKNOWN_STATE",
        ]
        for status in status_cases:
            with self.subTest(status=status):
                event = dict(queued)
                if status is None:
                    event.pop("status", None)
                else:
                    event["status"] = status
                manifest_path.write_bytes(collector.canonical_json_bytes(event))
                before = {
                    str(path.relative_to(self.root)): path.read_bytes()
                    for path in self.root.rglob("*")
                    if path.is_file()
                }
                code, payload, _ = self.run_main_json(
                    ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
                )
                after = {
                    str(path.relative_to(self.root)): path.read_bytes()
                    for path in self.root.rglob("*")
                    if path.is_file()
                }
                self.assertEqual(3, code)
                self.assertEqual("SAFETY_STOP", payload["status"])
                self.assertEqual("E_STATUS", payload["error_code"])
                self.assertEqual(before, after)
                self.assertEqual([], list(self.video_dir.iterdir()))
                self.assertFalse((self.state_dir / ".collector.lock").exists())

    def test_manifest_secret_gate_is_shared_by_all_commands(self) -> None:
        queued = self.queue_video()
        source = self.download_dir / "secret-check.mp4"
        source.write_bytes(b"source remains")
        mapping = self.root / "mapping.json"
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
        )
        manifest_path = self.state_dir / "manifest.jsonl"
        base_event = dict(queued)
        commands = [
            ["check", "--input", str(FIXTURE), "--now", NOW],
            ["move-completed", "--mapping", str(mapping), "--now", NOW],
            ["handoff", "--now", NOW],
        ]
        for index, command in enumerate(commands):
            with self.subTest(command=command[0]):
                event = dict(base_event)
                if index == 0:
                    event["cookie"] = "forbidden-manifest-value"
                else:
                    event["nested"] = {"token": "forbidden-manifest-value"}
                manifest_path.write_bytes(collector.canonical_json_bytes(event))
                before = manifest_path.read_bytes()
                code, payload, stdout = self.run_main_json(["--config", str(self.config_path), *command])
                self.assertEqual(3, code)
                self.assertEqual("E_SECRET_FIELD", payload["error_code"])
                self.assertNotIn("forbidden-manifest-value", stdout)
                self.assertEqual(before, manifest_path.read_bytes())
                self.assertEqual(b"source remains", source.read_bytes())
                self.assertEqual([], list(self.video_dir.iterdir()))

    def test_terminal_projection_rejects_missing_blank_or_malformed_evidence(self) -> None:
        queued = self.queue_video()
        mapping = self.root / "terminal-mapping.json"
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": "source-is-absent.mp4", "bvid": queued["bvid"]}]},
        )
        manifest_path = self.state_dir / "manifest.jsonl"
        valid_target = str(self.video_dir / "persisted-target.mp4")
        cases = [
            ("moved missing local", "VIDEO_MOVED", "local_file", None, True),
            ("moved blank local", "VIDEO_MOVED", "local_file", "   ", False),
            ("moved control local", "VIDEO_MOVED", "local_file", "bad\npath", False),
            ("moved missing sha", "VIDEO_MOVED", "sha256", None, True),
            ("moved blank sha", "VIDEO_MOVED", "sha256", "", False),
            ("moved malformed sha", "VIDEO_MOVED", "sha256", "A" * 64, False),
            ("retained missing reason", "VIDEO_MOVED_SOURCE_RETAINED", "failure_reason", None, True),
            ("retained blank reason", "VIDEO_MOVED_SOURCE_RETAINED", "failure_reason", " ", False),
            ("retained control reason", "VIDEO_MOVED_SOURCE_RETAINED", "failure_reason", "bad\rreason", False),
        ]
        for label, status, field, value, remove_field in cases:
            with self.subTest(label=label):
                event = dict(queued)
                event.update(
                    {
                        "status": status,
                        "local_file": valid_target,
                        "sha256": "a" * 64,
                        "failure_reason": "PermissionError: retained source",
                        "video_processing_status": "READY_FOR_HANDOFF",
                    }
                )
                if remove_field:
                    event.pop(field, None)
                else:
                    event[field] = value
                manifest_path.write_bytes(collector.canonical_json_bytes(event))
                before = {
                    str(path.relative_to(self.root)): path.read_bytes()
                    for path in self.root.rglob("*")
                    if path.is_file()
                }
                code, payload, stdout = self.run_main_json(
                    ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
                )
                after = {
                    str(path.relative_to(self.root)): path.read_bytes()
                    for path in self.root.rglob("*")
                    if path.is_file()
                }
                self.assertEqual(3, code)
                self.assertEqual("SAFETY_STOP", payload["status"])
                self.assertEqual("E_MANIFEST", payload["error_code"])
                self.assertEqual(before, after)
                self.assertNotIn("bad\npath", stdout)
                self.assertNotIn("bad\rreason", stdout)
                if isinstance(value, str) and value.strip():
                    self.assertNotIn(value, stdout)
                self.assertEqual([], list(self.video_dir.iterdir()))
                self.assertFalse((self.state_dir / ".collector.lock").exists())

    def test_handoff_rejects_control_or_invalid_manifest_scalars_without_writes(self) -> None:
        queued = self.queue_video()
        source = self.download_dir / "handoff-source.mp4"
        source_bytes = b"handoff scalar validation bytes"
        source.write_bytes(source_bytes)
        completed_mtime = datetime.fromisoformat("2026-08-04T11:30:00+08:00").timestamp()
        os.utime(source, (completed_mtime, completed_mtime))
        mapping = self.root / "handoff-mapping.json"
        self.write_json(
            mapping,
            {"schema_version": 1, "items": [{"source_file": source.name, "bvid": queued["bvid"]}]},
        )
        code, moved = collector.run(
            ["--config", str(self.config_path), "move-completed", "--mapping", str(mapping), "--now", NOW]
        )
        self.assertEqual(0, code)
        target = Path(moved["items"][0]["target"])
        latest, _ = collector.latest_entities(collector.load_manifest(self.state_dir / "manifest.jsonl"))
        moved_event = dict(latest[queued["entity_id"]])
        manifest_path = self.state_dir / "manifest.jsonl"
        cases = [
            ("entity newline", "entity_id", str(moved_event["entity_id"]) + "\nstatus=FORGED"),
            ("entity format", "entity_id", "not-an-internal-id"),
            ("bvid newline", "bvid", str(moved_event["bvid"]) + "\r\nstatus=FORGED"),
            ("bvid format", "bvid", "NOT-A-BVID"),
            ("published newline", "published_at", str(moved_event["published_at"]) + "\nstatus=FORGED"),
            ("published noncanonical", "published_at", "2026-08-04T03:00:00+00:00"),
            ("title newline", "title", str(moved_event["title"]) + "\nstatus=FORGED"),
            ("source url newline", "source_url", str(moved_event["source_url"]) + "\nstatus=FORGED"),
            ("local file newline", "local_file", str(moved_event["local_file"]) + "\nstatus=FORGED"),
            ("sha newline", "sha256", str(moved_event["sha256"]) + "\nstatus=FORGED"),
        ]
        for label, field, value in cases:
            with self.subTest(label=label):
                event = dict(moved_event)
                event[field] = value
                manifest_path.write_bytes(collector.canonical_json_bytes(event))
                before = {
                    str(path.relative_to(self.root)): path.read_bytes()
                    for path in self.root.rglob("*")
                    if path.is_file()
                }
                code, payload, stdout = self.run_main_json(
                    ["--config", str(self.config_path), "handoff", "--now", NOW]
                )
                after = {
                    str(path.relative_to(self.root)): path.read_bytes()
                    for path in self.root.rglob("*")
                    if path.is_file()
                }
                self.assertEqual(3, code)
                self.assertEqual("SAFETY_STOP", payload["status"])
                self.assertEqual("E_MANIFEST", payload["error_code"])
                self.assertNotIn("status=FORGED", stdout)
                self.assertNotIn(value, stdout)
                self.assertEqual(before, after)
                self.assertEqual(source_bytes, target.read_bytes())
                self.assertFalse((self.state_dir / ".collector.lock").exists())

    def test_native_handoff_route_is_required_and_validated_before_state_write(self) -> None:
        base = json.loads(self.config_path.read_text(encoding="utf-8"))
        for label, mutate in (
            ("missing field", lambda route: route.pop("target_thread_id")),
            ("invalid thread", lambda route: route.__setitem__("target_thread_id", "NOT-A-THREAD")),
            ("envelope injection", lambda route: route.__setitem__("target_ai_id", "media-processor\nstatus=FORGED")),
        ):
            with self.subTest(label=label):
                invalid = json.loads(json.dumps(base))
                mutate(invalid["native_handoff"])
                config = self.root / f"invalid-{label}.json"
                self.write_json(config, invalid)
                self.assertFalse(self.state_dir.exists())
                code, payload, _ = self.run_main_json(
                    ["--config", str(config), "check", "--input", str(FIXTURE), "--now", NOW]
                )
                self.assertEqual(2, code)
                self.assertEqual("E_CONFIG", payload["error_code"])
                self.assertFalse(self.state_dir.exists())

    def test_secret_fields_are_rejected_and_not_persisted(self) -> None:
        exported = json.loads(FIXTURE.read_text(encoding="utf-8"))
        exported["cookie"] = "forbidden-value"
        bad = self.root / "bad.json"
        self.write_json(bad, exported)
        code = collector.main(
            ["--config", str(self.config_path), "check", "--input", str(bad), "--now", NOW]
        )
        self.assertEqual(3, code)
        self.assertFalse((self.state_dir / "manifest.jsonl").exists())


if __name__ == "__main__":
    unittest.main()