#!/usr/bin/env python3
"""test_kb_deep_dive.py — V37.9.16 每日深度分析单测

覆盖维度：
  1. Tier 分类（一档/二档/三档）
  2. Entry 解析（从 H2 section 抽 title/link/stars/abstract）
  3. 评分 + 排序（⭐≥4 门槛 + 主题加权 + 摘要长度 tie-breaker）
  4. PDF / HTML fetcher（lazy import + degrade）
  5. Prompt builder（full_text vs abstract-only，grounding 约束）
  6. Output builder（markdown / WA / Discord 三格式）
  7. run() orchestrator（各种路径：ok / no_candidates / llm_failed）
  8. Shell 脚本守卫（反模式防御）
"""
import json
import os
import re
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime
from unittest.mock import patch, MagicMock

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import kb_deep_dive as m


# ══════════════════════════════════════════════════════════════════════
# 1. Tier 分类
# ══════════════════════════════════════════════════════════════════════
class TestClassifyTier(unittest.TestCase):
    def test_tier1_arxiv(self):
        self.assertEqual(m.classify_tier("arxiv_monitor"), 1)

    def test_tier1_acl(self):
        self.assertEqual(m.classify_tier("acl_anthology"), 1)

    def test_tier1_hf(self):
        self.assertEqual(m.classify_tier("hf_papers"), 1)

    def test_tier2_rss(self):
        self.assertEqual(m.classify_tier("rss_blogs"), 2)

    def test_tier2_ontology(self):
        self.assertEqual(m.classify_tier("ontology_sources"), 2)

    def test_tier3_hn_default(self):
        self.assertEqual(m.classify_tier("run_hn_fixed"), 3)

    def test_tier3_unknown(self):
        self.assertEqual(m.classify_tier("nonexistent_source"), 3)


# ══════════════════════════════════════════════════════════════════════
# 2. Entry 解析器
# ══════════════════════════════════════════════════════════════════════
class TestParseEntriesFromSection(unittest.TestCase):
    def test_single_entry(self):
        section = """
*Attention Is All You Need*
作者：Vaswani 等 | 日期：2017
链接：https://arxiv.org/abs/1706.03762
贡献：提出 Transformer 架构
价值：⭐⭐⭐⭐⭐
"""
        entries = m.parse_entries_from_section(section, "arxiv_monitor", "📄 ArXiv")
        self.assertEqual(len(entries), 1)
        self.assertEqual(entries[0]["title"], "Attention Is All You Need")
        self.assertEqual(entries[0]["link"], "https://arxiv.org/abs/1706.03762")
        self.assertEqual(entries[0]["stars"], 5)
        self.assertEqual(entries[0]["source_id"], "arxiv_monitor")

    def test_multiple_entries(self):
        section = """
*论文A*
链接：https://arxiv.org/abs/1111.1111
价值：⭐⭐⭐

*论文B*
链接：https://arxiv.org/abs/2222.2222
价值：⭐⭐⭐⭐⭐

*论文C*
价值：⭐⭐
"""
        entries = m.parse_entries_from_section(section, "arxiv_monitor", "📄 ArXiv")
        self.assertEqual(len(entries), 3)
        self.assertEqual(entries[0]["stars"], 3)
        self.assertEqual(entries[1]["stars"], 5)
        self.assertEqual(entries[2]["stars"], 2)

    def test_empty_section(self):
        self.assertEqual(m.parse_entries_from_section("", "x", "X"), [])

    def test_no_stars_defaults_zero(self):
        section = "*No stars here*\n链接：https://example.com"
        entries = m.parse_entries_from_section(section, "x", "X")
        self.assertEqual(entries[0]["stars"], 0)

    def test_fallback_url_without_link_label(self):
        """链接行不以'链接：'开头时也能抓到 URL。"""
        section = "*Paper*\nhttps://aclanthology.org/2024.acl-long.1/\n价值：⭐⭐⭐⭐"
        entries = m.parse_entries_from_section(section, "acl_anthology", "📝 ACL")
        self.assertTrue(entries[0]["link"].startswith("https://aclanthology.org/"))

    def test_abstract_accumulates(self):
        section = """
*Paper X*
贡献：核心贡献句1
要点：核心贡献句2
"""
        entries = m.parse_entries_from_section(section, "x", "X")
        self.assertIn("核心贡献句1", entries[0]["abstract"])
        self.assertIn("核心贡献句2", entries[0]["abstract"])


# ══════════════════════════════════════════════════════════════════════
# 3. 评分 + 排序
# ══════════════════════════════════════════════════════════════════════
class TestScoreEntry(unittest.TestCase):
    def test_low_stars_returns_negative(self):
        """⭐<MIN_STARS 必须返回负数以被排除。"""
        e = {"stars": 3, "title": "X", "abstract": ""}
        self.assertLess(m.score_entry(e), 0)

    def test_high_stars_base_score(self):
        e = {"stars": 5, "title": "plain", "abstract": ""}
        # 5*10 + 0 + 0 = 50
        self.assertEqual(m.score_entry(e), 50)

    def test_topic_keyword_boost(self):
        e = {"stars": 4, "title": "ontology engine", "abstract": ""}
        # 4*10 + 10 (ontology) = 50
        self.assertEqual(m.score_entry(e), 50)

    def test_multiple_topic_boosts(self):
        e = {
            "stars": 4,
            "title": "agent runtime with tool calling and ontology",
            "abstract": "",
        }
        # 4*10 + 10 (agent runtime) + 8 (tool calling) + 10 (ontology) = 68
        self.assertEqual(m.score_entry(e), 68)

    def test_abstract_length_tiebreaker(self):
        e_short = {"stars": 5, "title": "x", "abstract": "y" * 50}
        e_long = {"stars": 5, "title": "x", "abstract": "y" * 1000}
        self.assertLess(m.score_entry(e_short), m.score_entry(e_long))

    def test_tie_breaker_bonus_capped(self):
        """摘要长度奖励封顶 MAX_ABSTRACT_BONUS=10。"""
        e_capped = {"stars": 5, "title": "x", "abstract": "y" * 2000}
        # 5*10 + 0 + min(2000//100, 10) = 50 + 10 = 60
        self.assertEqual(m.score_entry(e_capped), 60)


# ══════════════════════════════════════════════════════════════════════
# 4. PDF URL 派生
# ══════════════════════════════════════════════════════════════════════
class TestPickTopTierFallback(unittest.TestCase):
    """V37.9.17 方案 C: tier-aware fallback — TIER 1/2 优先，TIER 3 fallback。

    背景：V37.9.16 首跑选中 ai_leaders_x ⭐5 走 abstract_only，用户感知质量低于
    预期。方案 C 保证 TIER 1/2 候选永远优先于 TIER 3，即使 score 较低。
    """

    def _entry(self, source_id, stars, title="x"):
        return {
            "title": title,
            "link": "https://example.com",
            "stars": stars,
            "abstract": "",
            "source_id": source_id,
            "source_label": source_id,
        }

    def test_empty_returns_none(self):
        self.assertIsNone(m.pick_top([]))

    def test_tier1_wins_over_higher_score_tier3(self):
        """关键场景：arxiv ⭐4 必须胜过 X tweet ⭐5（tier-aware）。"""
        tier3 = self._entry("ai_leaders_x", 5, "X tweet")
        tier1 = self._entry("arxiv_monitor", 4, "Arxiv paper")
        # candidates 已按 score 排序时 tier3 在前（V37.9.16 行为）
        result = m.pick_top([tier3, tier1])
        self.assertEqual(result["source_id"], "arxiv_monitor")
        self.assertEqual(result["title"], "Arxiv paper")

    def test_tier2_wins_over_tier3(self):
        tier3 = self._entry("run_hn_fixed", 5, "HN")
        tier2 = self._entry("rss_blogs", 4, "Blog post")
        result = m.pick_top([tier3, tier2])
        self.assertEqual(result["source_id"], "rss_blogs")

    def test_tier1_beats_tier2_when_both_present(self):
        """TIER 1+2 同桶内按 score 排序（不区分 1 vs 2）— 输入顺序即决定。"""
        tier1 = self._entry("arxiv_monitor", 4, "Arxiv")
        tier2 = self._entry("rss_blogs", 5, "Blog")
        # 输入按 score 排序：tier2 score 高在前 → 同桶内 tier2 胜出
        result = m.pick_top([tier2, tier1])
        self.assertEqual(result["source_id"], "rss_blogs")
        # 反过来 tier1 在前则 tier1 胜出
        result2 = m.pick_top([tier1, tier2])
        self.assertEqual(result2["source_id"], "arxiv_monitor")

    def test_only_tier3_fallback(self):
        """TIER 1+2 全空 → 回退 TIER 3（保留 V37.9.16 行为，没新候选时仍推送）。"""
        tier3a = self._entry("ai_leaders_x", 5, "X tweet A")
        tier3b = self._entry("run_hn_fixed", 4, "HN B")
        result = m.pick_top([tier3a, tier3b])
        self.assertEqual(result["source_id"], "ai_leaders_x")
        self.assertEqual(result["title"], "X tweet A")

    def test_v37_9_16_blood_lesson_scenario(self):
        """直接复现 V37.9.16 首跑场景：tier3 X tweet ⭐5 + tier1 论文 ⭐4
        必须选 tier1 论文（方案 C 兑现）。"""
        x_tweet = self._entry("ai_leaders_x", 5, "国家AI动员")
        arxiv = self._entry("arxiv_monitor", 4, "Some Paper")
        hf = self._entry("hf_papers", 4, "HF Paper")
        # 模拟 collect_today_candidates 已排序输出（X tweet score 最高在前）
        result = m.pick_top([x_tweet, arxiv, hf])
        self.assertIn(result["source_id"], ("arxiv_monitor", "hf_papers"))
        self.assertNotEqual(result["source_id"], "ai_leaders_x")


class TestPdfUrlDerivation(unittest.TestCase):
    def test_arxiv_abs_to_pdf(self):
        self.assertEqual(
            m.arxiv_url_to_pdf("https://arxiv.org/abs/2401.12345"),
            "https://arxiv.org/pdf/2401.12345.pdf",
        )

    def test_arxiv_abs_trailing_slash(self):
        self.assertEqual(
            m.arxiv_url_to_pdf("https://arxiv.org/abs/2401.12345/"),
            "https://arxiv.org/pdf/2401.12345.pdf",
        )

    def test_arxiv_already_pdf(self):
        url = "https://arxiv.org/pdf/2401.12345.pdf"
        self.assertEqual(m.arxiv_url_to_pdf(url), url)

    def test_acl_url_to_pdf(self):
        self.assertEqual(
            m.acl_url_to_pdf("https://aclanthology.org/2024.acl-long.123/"),
            "https://aclanthology.org/2024.acl-long.123.pdf",
        )

    def test_non_arxiv_returns_none(self):
        self.assertIsNone(m.arxiv_url_to_pdf("https://example.com/paper.html"))


# ══════════════════════════════════════════════════════════════════════
# 5. PDF 预处理（切 References + 去图表噪声 + 截断）
# ══════════════════════════════════════════════════════════════════════
class TestPreprocessPdfText(unittest.TestCase):
    def test_cuts_references(self):
        raw = "Body content here.\n\nReferences\n[1] Smith et al. 2024\n[2] Doe 2023\n"
        out = m.preprocess_pdf_text(raw)
        self.assertIn("Body content", out)
        self.assertNotIn("Smith et al", out)

    def test_cuts_acknowledgments(self):
        raw = "Core text.\n\nAcknowledgments\nWe thank...\n"
        out = m.preprocess_pdf_text(raw)
        self.assertIn("Core text", out)
        self.assertNotIn("We thank", out)

    def test_cuts_chinese_references(self):
        raw = "正文内容.\n\n参考文献\n[1] 某某 2024"
        out = m.preprocess_pdf_text(raw)
        self.assertIn("正文", out)
        self.assertNotIn("某某 2024", out)

    def test_truncates_to_max_chars(self):
        raw = "x" * 50000
        out = m.preprocess_pdf_text(raw, max_chars=1000)
        self.assertLessEqual(len(out), 1100)  # +" ...[truncated]"
        self.assertIn("[truncated]", out)

    def test_removes_figure_captions(self):
        raw = "Main text.\nFigure 1: A diagram showing something important.\nMore main text."
        out = m.preprocess_pdf_text(raw)
        self.assertIn("Main text", out)
        self.assertIn("More main text", out)


# ══════════════════════════════════════════════════════════════════════
# 6. Fetch 降级路径（lazy import 失败 → degrade）
# ══════════════════════════════════════════════════════════════════════
class TestFetchDegrade(unittest.TestCase):
    def test_tier3_source_not_fetched(self):
        entry = {
            "source_id": "run_hn_fixed",
            "title": "x",
            "abstract": "this is an HN discussion",
            "link": "https://news.ycombinator.com/item?id=1",
            "stars": 5,
        }
        mode, text, reason = m.fetch_full_text(entry)
        self.assertEqual(mode, "abstract_only")
        self.assertIn("tier3", reason)
        self.assertEqual(text, "this is an HN discussion")

    def test_no_link_degrades(self):
        entry = {
            "source_id": "arxiv_monitor",
            "title": "x",
            "abstract": "abstract only",
            "link": "",
            "stars": 5,
        }
        mode, text, reason = m.fetch_full_text(entry)
        self.assertEqual(mode, "abstract_only")

    def test_pdf_fetch_lazy_import_or_http_fail_degrades(self):
        """即使 pdfplumber 未装或 HTTP 失败，必须 degrade 不 raise。"""
        entry = {
            "source_id": "arxiv_monitor",
            "title": "x",
            "abstract": "fallback abstract",
            "link": "https://arxiv.org/abs/invalid.12345",
            "stars": 5,
        }
        # 不走真 HTTP — 用 patch 模拟 urlopen 抛错保证路径被覆盖
        with patch("kb_deep_dive._urlopen") as mock_open:
            mock_open.side_effect = m.urllib.error.URLError("mocked fail")
            mode, text, reason = m.fetch_full_text(entry)
        self.assertEqual(mode, "abstract_only")
        self.assertEqual(text, "fallback abstract")


# ══════════════════════════════════════════════════════════════════════
# 7. Prompt builder
# ══════════════════════════════════════════════════════════════════════
class TestPromptBuilders(unittest.TestCase):
    def _entry(self):
        return {
            "title": "Test Paper",
            "source_label": "📄 ArXiv",
            "link": "https://arxiv.org/abs/2401.12345",
            "stars": 5,
            "source_id": "arxiv_monitor",
        }

    def test_full_text_prompt_includes_structured_sections(self):
        p = m.build_full_text_prompt(self._entry(), "full body content")
        self.assertIn("核心论点", p)
        self.assertIn("论证链", p)
        self.assertIn("实验/证据", p)
        self.assertIn("局限性", p)
        self.assertIn("full body content", p)

    def test_abstract_prompt_forbids_speculation(self):
        p = m.build_abstract_only_prompt(self._entry(), "just the abstract")
        self.assertIn("严禁推测方法细节", p)
        self.assertIn("基于摘要", p)
        self.assertIn("just the abstract", p)

    def test_both_prompts_include_grounding(self):
        e = self._entry()
        self.assertIn("严格约束", m.build_full_text_prompt(e, "x"))
        self.assertIn("严格约束", m.build_abstract_only_prompt(e, "x"))

    def test_build_prompt_for_entry_dispatches(self):
        e = self._entry()
        full = m.build_prompt_for_entry(e, "full_text", "body")
        abs_p = m.build_prompt_for_entry(e, "abstract_only", "body")
        self.assertIn("完整", full)
        self.assertIn("摘要", abs_p)


# ══════════════════════════════════════════════════════════════════════
# 8. Output builder
# ══════════════════════════════════════════════════════════════════════
class TestOutputBuilders(unittest.TestCase):
    def _entry(self):
        return {
            "title": "Paper X",
            "source_label": "📄 ArXiv",
            "link": "https://arxiv.org/abs/2401.12345",
            "stars": 5,
            "source_id": "arxiv_monitor",
        }

    def test_markdown_has_frontmatter(self):
        md = m.build_deep_dive_markdown(
            self._entry(), "full_text", "LLM analysis content here", "", "2026-04-24"
        )
        self.assertTrue(md.startswith("---\n"))
        self.assertIn("date: 2026-04-24", md)
        self.assertIn("type: deep_dive", md)
        self.assertIn("LLM analysis content here", md)

    def test_markdown_includes_degrade_notice(self):
        md = m.build_deep_dive_markdown(
            self._entry(),
            "abstract_only",
            "analysis",
            "PDF fetch failed: HTTP 404",
            "2026-04-24",
        )
        self.assertIn("抓取降级原因", md)
        self.assertIn("HTTP 404", md)

    def test_wa_message_bounded(self):
        long_content = "a" * 5000
        wa = m.build_deep_dive_wa(self._entry(), "full_text", long_content, "2026-04-24")
        # V37.9.35: budget bumped 1400→4000 (WhatsApp client folding confirmed); +50 buffer
        self.assertLessEqual(len(wa), 4050)

    def test_discord_has_markdown_formatting(self):
        disc = m.build_deep_dive_discord(
            self._entry(), "full_text", "content", "2026-04-24"
        )
        self.assertIn("**", disc)
        self.assertIn("Paper X", disc)

    def test_abstract_mode_tag_in_push(self):
        wa = m.build_deep_dive_wa(
            self._entry(), "abstract_only", "content", "2026-04-24"
        )
        self.assertIn("摘要级", wa)
        disc = m.build_deep_dive_discord(
            self._entry(), "abstract_only", "content", "2026-04-24"
        )
        self.assertIn("摘要级", disc)


# ══════════════════════════════════════════════════════════════════════
# 8b. V37.9.21 — Multi-part WA splitting (Dream-style)
# ══════════════════════════════════════════════════════════════════════
class TestSplitTextIntoChunks(unittest.TestCase):
    """V37.9.21: chunk splitter for long LLM content."""

    def test_short_text_returns_single_chunk(self):
        chunks = m._split_text_into_chunks("hello world", 1000)
        self.assertEqual(chunks, ["hello world"])

    def test_empty_text_returns_empty_list(self):
        self.assertEqual(m._split_text_into_chunks("", 1000), [])

    def test_long_text_splits_at_paragraph_boundary(self):
        # First half + paragraph break + second half — splitter prefers \n\n
        text = ("a" * 800) + "\n\n" + ("b" * 800)
        chunks = m._split_text_into_chunks(text, 1000)
        self.assertEqual(len(chunks), 2)
        # First chunk should end at the paragraph (no b's)
        self.assertNotIn("b", chunks[0])
        # Second chunk should be the b's
        self.assertNotIn("a", chunks[1])

    def test_long_text_splits_at_line_boundary_when_no_paragraph(self):
        text = "\n".join(["line " + ("x" * 100) for _ in range(20)])
        chunks = m._split_text_into_chunks(text, 1000)
        self.assertGreater(len(chunks), 1)
        # Each chunk should be at most max_chunk + overhead from boundary search
        for chunk in chunks:
            self.assertLessEqual(len(chunk), 1100)

    def test_no_boundary_falls_back_to_hard_cut(self):
        # Pure 'a' string with no separators
        text = "a" * 3000
        chunks = m._split_text_into_chunks(text, 1000)
        self.assertEqual(len(chunks), 3)
        # Each chunk is exactly 1000 chars (hard cut)
        for chunk in chunks:
            self.assertEqual(len(chunk), 1000)

    def test_preserves_full_content(self):
        text = "Section 1\n\n" + ("a" * 500) + "\n\nSection 2\n\n" + ("b" * 500)
        chunks = m._split_text_into_chunks(text, 600)
        # All a's and b's must appear somewhere across chunks
        joined = "".join(chunks)
        self.assertEqual(joined.count("a"), 500)
        self.assertEqual(joined.count("b"), 500)

    def test_chinese_punctuation_boundary(self):
        # No paragraph or line breaks, but Chinese sentence punctuation
        text = ("内容" * 100) + "。" + ("更多" * 100) + "。" + ("结尾" * 100)
        chunks = m._split_text_into_chunks(text, 300)
        self.assertGreater(len(chunks), 1)
        # First chunk should end at one of the 。 boundaries
        self.assertTrue(chunks[0].endswith("。") or len(chunks[0]) == 300)


class TestBuildDeepDiveWaParts(unittest.TestCase):
    """V37.9.21: multi-part WA builder."""

    def _entry(self):
        return {
            "title": "Test Paper Title",
            "source_label": "📄 ArXiv",
            "link": "https://arxiv.org/abs/2401.12345",
            "stars": 5,
            "source_id": "arxiv_monitor",
        }

    def test_short_content_single_part(self):
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", "短分析。", "2026-04-27"
        )
        self.assertEqual(len(parts), 1)
        # Single-part: NO [i/N] indicator
        self.assertNotIn("[1/", parts[0])
        # Has header
        self.assertIn("Test Paper Title", parts[0])
        self.assertIn("⭐", parts[0])
        # Has link
        self.assertIn("arxiv.org", parts[0])

    def test_long_content_multi_part(self):
        # V37.9.35: bumped budget 1400→4000, need longer content to trigger split.
        # Each "段落内容。" is 5 chars, 1500 × 5 = 7500 chars > 3800 body budget.
        long_content = "段落内容。" * 1500  # ~7500 chars
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", long_content, "2026-04-27"
        )
        self.assertGreater(len(parts), 1, "Long content should split into multiple parts")

    def test_each_part_has_indexed_header(self):
        long_content = "x" * 5000
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", long_content, "2026-04-27"
        )
        total = len(parts)
        self.assertGreater(total, 1)
        for idx, part in enumerate(parts, start=1):
            # Each part must have [i/N] indicator
            self.assertIn(f"[{idx}/{total}]", part,
                f"Part {idx} missing [{idx}/{total}] indicator")
            # Each part has title for context
            self.assertIn("Test Paper Title", part)

    def test_link_only_on_first_part(self):
        long_content = "y" * 5000
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", long_content, "2026-04-27"
        )
        self.assertGreater(len(parts), 1)
        # Part 1 has link
        self.assertIn("arxiv.org/abs/2401.12345", parts[0])
        # Parts 2+ don't repeat link (saves chars)
        for part in parts[1:]:
            self.assertNotIn("arxiv.org/abs/2401.12345", part)

    def test_each_part_within_budget(self):
        long_content = "z" * 8000
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", long_content, "2026-04-27"
        )
        for idx, part in enumerate(parts, start=1):
            self.assertLessEqual(len(part), m._WA_BUDGET_PER_PART + 50,
                f"Part {idx} exceeds budget: {len(part)} > {m._WA_BUDGET_PER_PART}")

    def test_abstract_mode_tag_on_each_part(self):
        # V37.9.35: longer content for new 4000 budget
        long_content = "段落。" * 1500  # ~4500 chars
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "abstract_only", long_content, "2026-04-27"
        )
        self.assertGreater(len(parts), 1)
        for part in parts:
            self.assertIn("摘要级", part,
                "All parts must show abstract_only mode tag for context")

    def test_empty_llm_content_returns_empty_list(self):
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", "", "2026-04-27"
        )
        self.assertEqual(parts, [])

    def test_full_content_preserved_across_parts(self):
        """Critical: splitter must NOT lose content (no truncation)."""
        # Use distinctive markers so we can verify full preservation
        markers = [f"MARKER_{i:04d}" for i in range(50)]
        long_content = "\n".join(
            f"{m_str}: {'x' * 60}" for m_str in markers
        )
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", long_content, "2026-04-27"
        )
        # Concatenate body across all parts and check every marker is present
        joined = "\n".join(parts)
        for marker in markers:
            self.assertIn(marker, joined,
                f"Content lost: {marker} not in any part")

    def test_backward_compat_build_deep_dive_wa_returns_first_part(self):
        """build_deep_dive_wa() (legacy) returns parts[0] for backward compat."""
        long_content = "abc " * 1000
        wa = m.build_deep_dive_wa(self._entry(), "full_text", long_content, "2026-04-27")
        parts = m.build_deep_dive_wa_parts(
            self._entry(), "full_text", long_content, "2026-04-27"
        )
        self.assertEqual(wa, parts[0])


# ══════════════════════════════════════════════════════════════════════
# 9. run() orchestrator — 端到端路径
# ══════════════════════════════════════════════════════════════════════
class TestRunOrchestrator(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.kb_dir = self.tmp.name
        os.makedirs(os.path.join(self.kb_dir, "sources"), exist_ok=True)

        # Fake registry
        self.registry_path = os.path.join(self.tmp.name, "jobs_registry.yaml")
        with open(self.registry_path, "w", encoding="utf-8") as f:
            f.write(
                "jobs:\n"
                "  - id: arxiv_monitor\n"
                "    enabled: true\n"
                "    kb_source_file: arxiv_daily.md\n"
                "    kb_source_label: ArXiv\n"
            )
        self.today = datetime(2026, 4, 24)
        self.today_str = "2026-04-24"

    def tearDown(self):
        self.tmp.cleanup()

    def _write_arxiv_source(self, entries_text):
        path = os.path.join(self.kb_dir, "sources", "arxiv_daily.md")
        with open(path, "w", encoding="utf-8") as f:
            f.write(f"## {self.today_str}\n\n{entries_text}\n")

    def test_no_candidates_status(self):
        """无合格候选 → status=no_candidates。"""
        self._write_arxiv_source("*Low paper*\n链接：x\n价值：⭐⭐")
        out = m.run(
            self.kb_dir, self.registry_path, today=self.today,
            llm_caller=lambda p: (True, "content", ""),
            fetcher=lambda e: ("full_text", "body", ""),
        )
        self.assertEqual(out["status"], "no_candidates")
        self.assertEqual(out["candidates_count"], 0)

    def test_ok_path_full_text(self):
        self._write_arxiv_source(
            "*Great paper on ontology engines*\n"
            "链接：https://arxiv.org/abs/2401.12345\n"
            "贡献：本体工程新框架\n"
            "价值：⭐⭐⭐⭐⭐"
        )
        out = m.run(
            self.kb_dir, self.registry_path, today=self.today,
            llm_caller=lambda p: (True, "深度分析结果", ""),
            fetcher=lambda e: ("full_text", "FULL BODY", ""),
        )
        self.assertEqual(out["status"], "ok")
        self.assertEqual(out["mode"], "full_text")
        self.assertIn("深度分析结果", out["markdown"])
        self.assertIn("Great paper on ontology engines", out["pick"]["title"])
        self.assertGreaterEqual(out["candidates_count"], 1)

    def test_llm_failed_does_not_produce_artifacts(self):
        self._write_arxiv_source(
            "*Paper*\n链接：https://arxiv.org/abs/x\n价值：⭐⭐⭐⭐⭐"
        )
        out = m.run(
            self.kb_dir, self.registry_path, today=self.today,
            llm_caller=lambda p: (False, "", "timeout after 120s"),
            fetcher=lambda e: ("full_text", "BODY", ""),
        )
        self.assertEqual(out["status"], "llm_failed")
        self.assertIn("timeout", out["reason"])
        # Must NOT produce success-only artifacts
        self.assertNotIn("markdown", out)
        self.assertNotIn("wa_message", out)

    def test_degrade_path_abstract_only(self):
        self._write_arxiv_source(
            "*Paper with fetch fail*\n链接：https://arxiv.org/abs/bad\n贡献：摘要内容\n价值：⭐⭐⭐⭐⭐"
        )
        out = m.run(
            self.kb_dir, self.registry_path, today=self.today,
            llm_caller=lambda p: (True, "摘要级分析", ""),
            fetcher=lambda e: ("abstract_only", e.get("abstract", ""), "PDF fetch failed: HTTP 404"),
        )
        self.assertEqual(out["status"], "ok")
        self.assertEqual(out["mode"], "abstract_only")
        self.assertIn("HTTP 404", out["degrade_reason"])
        self.assertIn("摘要级", out["discord_message"])

    def test_collector_failed_on_missing_registry(self):
        out = m.run(
            self.kb_dir, "/nonexistent/registry.yaml", today=self.today,
        )
        self.assertEqual(out["status"], "collector_failed")

    def test_ok_path_returns_wa_parts_list(self):
        """V37.9.21: run() must return wa_parts as list[str] for multi-part send.
        V37.9.35: bumped content from ~2400 → ~7000 chars after budget 1400→4000."""
        self._write_arxiv_source(
            "*Long paper analysis*\n"
            "链接：https://arxiv.org/abs/2401.99999\n"
            "贡献：详细贡献\n"
            "价值：⭐⭐⭐⭐⭐"
        )
        long_llm = "深度分析段落。\n\n" * 600  # ~7200 chars guaranteed multi-part at new budget
        out = m.run(
            self.kb_dir, self.registry_path, today=self.today,
            llm_caller=lambda p: (True, long_llm, ""),
            fetcher=lambda e: ("full_text", "BODY", ""),
        )
        self.assertEqual(out["status"], "ok")
        self.assertIn("wa_parts", out)
        self.assertIsInstance(out["wa_parts"], list)
        self.assertGreater(len(out["wa_parts"]), 1,
            "Long LLM content should split into multiple WA parts")
        # Backward-compat: wa_message still equals parts[0]
        self.assertEqual(out["wa_message"], out["wa_parts"][0])


# ══════════════════════════════════════════════════════════════════════
# 10. Shell 脚本守卫
# ══════════════════════════════════════════════════════════════════════
class TestKbDeepDiveShellGuards(unittest.TestCase):
    def setUp(self):
        self.script_path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "kb_deep_dive.sh"
        )
        with open(self.script_path, "r", encoding="utf-8") as f:
            self.content = f.read()

    def test_script_exists(self):
        self.assertTrue(os.path.isfile(self.script_path))

    def test_script_has_system_alert_prefix(self):
        self.assertIn("[SYSTEM_ALERT]", self.content)

    def test_script_uses_env_var_heredoc_pattern(self):
        """V37.5.1 反模式防御：禁止 `echo ... | python3 - << 'PYEOF'`
        因 pipe+heredoc 会让 stdin 被 heredoc 覆盖。"""
        # 除注释外不得有 `| python3 -`
        lines = [l for l in self.content.split("\n") if not l.strip().startswith("#")]
        for ln in lines:
            self.assertNotIn("| python3 -\n", ln + "\n")

    def test_script_sources_notify_sh(self):
        self.assertIn("notify.sh", self.content)

    def test_script_has_set_euo_pipefail(self):
        # V37.9.61 升级到 set -eEuo pipefail (加 -E errtrace, V37.9.58-hotfix4 教训)
        # alternation 接受 set -euo / set -eEuo 两种形式 (向后兼容)
        has_set_e = any(marker in self.content for marker in
                        ["set -euo pipefail", "set -eEuo pipefail"])
        self.assertTrue(has_set_e,
            "kb_deep_dive.sh 必须含 set -euo pipefail 或 set -eEuo pipefail (V37.9.61 升级)")

    def test_script_writes_status_file(self):
        self.assertIn("last_run_deep_dive.json", self.content)

    def test_script_uses_deep_dive_topic(self):
        self.assertIn("--topic deep_dive", self.content)

    def test_script_has_rsync_with_forensics(self):
        """V37.9.14 INV-BACKUP-001 check 4: rsync+MOVESPEED 必须调 incident capture
        V37.9.27 升级: site 用 movespeed_rsync_helper.sh, helper 内部接管 capture"""
        # V37.9.14 旧契约 OR V37.9.27 新契约 任一满足即合规
        has_legacy = "movespeed_incident_capture.sh" in self.content
        has_v37_9_27_helper = "movespeed_rsync_helper.sh" in self.content
        self.assertTrue(has_legacy or has_v37_9_27_helper,
            "V37.9.14: site 需调 movespeed_incident_capture.sh OR "
            "V37.9.27: 通过 movespeed_rsync_helper.sh 间接调用 (helper 内部接管)")

    def test_script_fail_fast_on_llm_failed(self):
        """llm_failed 分支必须 exit 1（fail-fast 契约）"""
        # 简单 heuristic: llm_failed 块内容里必须 exit 1
        idx = self.content.find("llm_failed")
        self.assertGreater(idx, 0)
        # 从该处到后面 500 字符内必须有 exit 1
        region = self.content[idx : idx + 800]
        self.assertIn("exit 1", region)

    # ── V37.9.21 multi-part WA shell guards ──────────────────────────────

    def test_v37_9_21_marker_in_script(self):
        self.assertIn("V37.9.21", self.content,
            "kb_deep_dive.sh must mark V37.9.21 multi-part section")

    def test_script_extracts_wa_parts_via_env_var_heredoc(self):
        """V37.5.1 反模式防御：必须用 env-var heredoc, 不能 echo | python3 -."""
        self.assertIn('COLLECTOR_OUTPUT="$COLLECTOR_OUTPUT"', self.content)
        self.assertIn("wa_parts", self.content,
            "Script must consume wa_parts (V37.9.21 list)")

    def test_script_writes_chunks_to_tempdir(self):
        """V37.9.21: chunks written to mktemp dir, cleaned up on EXIT."""
        self.assertIn("mktemp -d", self.content)
        self.assertIn("WA_CHUNK_DIR", self.content)
        # trap EXIT for cleanup
        self.assertIn("trap", self.content)
        self.assertIn('rm -rf "$WA_CHUNK_DIR"', self.content)

    def test_script_loops_over_chunk_files_with_sleep(self):
        """V37.9.21: send loop iterates *.txt with 1s sleep between (Dream pattern)."""
        # Loop body
        self.assertIn('for chunk_file in "$WA_CHUNK_DIR"/*.txt', self.content)
        # Sleep between segments to avoid WhatsApp out-of-order
        self.assertIn("sleep 1", self.content)
        # WA_PART_IDX counter
        self.assertIn("WA_PART_IDX", self.content)
        # Total parts variable
        self.assertIn("WA_PARTS_TOTAL", self.content)

    def test_script_has_separate_wa_and_discord_send(self):
        """Discord stays single-send (per V37.9.21 design decision 1: only WA splits)."""
        # Both notify --channel whatsapp and --channel discord must exist
        self.assertIn("--channel whatsapp --topic deep_dive", self.content)
        self.assertIn("--channel discord --topic deep_dive", self.content)
        # Discord NOT in WA chunk loop — separate notify call
        # Check that DISCORD_MSG is sent ONCE (not in loop)
        # heuristic: count occurrences of DISCORD_MSG sends
        discord_send_count = self.content.count('"$DISCORD_MSG"')
        self.assertGreaterEqual(discord_send_count, 1)

    def test_script_logs_part_count_on_success(self):
        """ops visibility: log message includes 'X/N 段' so failures are visible."""
        self.assertIn("$WA_SEND_OK/$WA_PARTS_TOTAL", self.content)

    # ── V37.9.22: deployment inconsistency window detection ────────────────

    def test_v37_9_22_marker_in_script(self):
        self.assertIn("V37.9.22", self.content,
            "kb_deep_dive.sh must mark V37.9.22 wa_parts missing detection section")

    def test_v37_9_22_warns_when_wa_parts_field_missing(self):
        """Source-level guard: heredoc must check 'wa_parts' not in data and emit WARN to stderr.
        Detects the V37.9.21 deployment-inconsistency-window silent fail
        (new shell + old py temporarily mixed during 2-min auto_deploy polling)."""
        # The check itself
        self.assertIn('"wa_parts" not in data', self.content,
            "Must explicitly check key existence (not None/empty) to distinguish 'old py' from 'picker returned empty'")
        # Must write to stderr (so $(...) capture doesn't swallow it)
        self.assertIn("file=sys.stderr", self.content,
            "WARN must go to stderr to avoid contaminating $(...) command substitution capture")
        # Specific marker mentioning the deployment inconsistency cause
        self.assertIn("部署不一致", self.content,
            "WARN message must explain root cause for ops visibility")


class TestV37_9_22_WaPartsMissingWarn(unittest.TestCase):
    """Behavior-level: extract heredoc and run it to verify stderr WARN behavior.
    Mirrors V37.9.13 test_restart_launchd subprocess-runtime pattern."""

    @classmethod
    def setUpClass(cls):
        cls.script_path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "kb_deep_dive.sh"
        )
        with open(cls.script_path, "r", encoding="utf-8") as f:
            cls.content = f.read()

    def _extract_heredoc_python(self):
        """Extract the wa_parts heredoc (between PYEOF markers right after WA_PARTS_TOTAL=)."""
        marker = "WA_PARTS_TOTAL=$(COLLECTOR_OUTPUT="
        start = self.content.find(marker)
        self.assertGreater(start, 0, "WA_PARTS_TOTAL heredoc not found")
        # Find the heredoc body between << 'PYEOF' and PYEOF
        heredoc_open = self.content.find("<< 'PYEOF'", start)
        body_start = self.content.find("\n", heredoc_open) + 1
        body_end = self.content.find("\nPYEOF", body_start)
        return self.content[body_start:body_end]

    def _run_heredoc(self, collector_json):
        import subprocess, tempfile
        body = self._extract_heredoc_python()
        with tempfile.TemporaryDirectory() as td:
            env = {"COLLECTOR_OUTPUT": collector_json, "CHUNK_DIR": td, "PATH": os.environ.get("PATH", "")}
            proc = subprocess.run(
                ["python3", "-c", body],
                env=env, capture_output=True, text=True, timeout=10,
            )
            return proc

    def test_warn_emitted_when_wa_parts_field_missing(self):
        """Old py simulation: collector JSON without wa_parts field → stderr WARN."""
        old_py_output = json.dumps({
            "status": "ok",
            "wa_message": "single-segment fallback",
            "discord_message": "discord version",
        })
        proc = self._run_heredoc(old_py_output)
        self.assertEqual(proc.returncode, 0, f"heredoc failed: {proc.stderr}")
        self.assertIn("WARN", proc.stderr)
        self.assertIn("wa_parts", proc.stderr)
        self.assertIn("部署不一致", proc.stderr)
        # stdout still produces parts count for $(...) capture
        self.assertEqual(proc.stdout.strip(), "1")

    def test_no_warn_when_wa_parts_field_present(self):
        """New py: collector JSON with wa_parts → no WARN."""
        new_py_output = json.dumps({
            "status": "ok",
            "wa_parts": ["part 1 of 2", "part 2 of 2"],
            "wa_message": "part 1 of 2",
            "discord_message": "discord version",
        })
        proc = self._run_heredoc(new_py_output)
        self.assertEqual(proc.returncode, 0)
        self.assertNotIn("WARN", proc.stderr)
        self.assertEqual(proc.stdout.strip(), "2")

    def test_no_warn_when_wa_parts_present_but_empty(self):
        """Edge case: wa_parts=[] should NOT trigger WARN (key exists, just empty).
        Empty list is a valid 'no candidates' state already handled upstream."""
        new_py_empty = json.dumps({
            "status": "ok",
            "wa_parts": [],
            "wa_message": "fallback message",
            "discord_message": "discord",
        })
        proc = self._run_heredoc(new_py_empty)
        self.assertEqual(proc.returncode, 0)
        self.assertNotIn("WARN", proc.stderr,
            "Empty wa_parts is valid (key exists), only missing key should warn")
        # Falls back to wa_message (existing behavior)
        self.assertEqual(proc.stdout.strip(), "1")

    def test_warn_does_not_pollute_stdout_capture(self):
        """Critical contract: WARN must go to stderr, not stdout.
        If WARN went to stdout, WA_PARTS_TOTAL=$(...) would capture 'WARN: ...\\n1'
        and downstream arithmetic would break."""
        old_py_output = json.dumps({"status": "ok", "wa_message": "x"})
        proc = self._run_heredoc(old_py_output)
        # stdout must be ONLY the integer count
        self.assertEqual(proc.stdout.strip(), "1")
        self.assertNotIn("WARN", proc.stdout)


class TestV37960Hotfix3SilentAbortFix(unittest.TestCase):
    """V37.9.60-hotfix3 反向验证守卫:
    send_wa_parts_via_notify / send_wa_parts_via_openclaw 函数末尾必须 `return 0`,
    防 V37.9.21 引入的 bash quirk: `[ X -lt Y ] && sleep 1` 单段时短路返回 1,
    函数 implicit return 1 → set -e 杀 caller → write_status 不跑 → last_run 停滞.

    5/8-5/12 5 天血案: kb_deep_dive cron 每天 LLM ok → markdown 写盘 → 推送成功,
    但 last_run.json 停在 5/7 22:30 status:llm_failed (前一次失败) 因为
    write_status "ok" 在 send_wa_parts 后, 被 silent abort 杀.
    """

    def setUp(self):
        self.script_path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "kb_deep_dive.sh"
        )
        with open(self.script_path, "r", encoding="utf-8") as f:
            self.content = f.read()

    def test_send_wa_parts_via_notify_ends_with_return_0(self):
        """送 notify 函数末尾必须显式 `return 0` 防 bash quirk."""
        # 找函数定义到下一个 `}` 之间的内容
        import re
        m = re.search(
            r'send_wa_parts_via_notify\(\)\s*\{(.+?)^\}',
            self.content, re.DOTALL | re.MULTILINE
        )
        self.assertIsNotNone(m, "找不到 send_wa_parts_via_notify 函数定义")
        body = m.group(1)
        self.assertIn("return 0", body,
            "V37.9.60-hotfix3: send_wa_parts_via_notify 末尾必须 `return 0` 防 bash quirk 单段时 "
            "`[ X -lt Y ] && sleep 1` 短路返回 1 → set -e 杀 caller. "
            "血案 5/8-5/12 5 天 kb_deep_dive last_run 不更新.")

    def test_send_wa_parts_via_openclaw_ends_with_return_0(self):
        """送 openclaw 函数末尾同款保护."""
        import re
        m = re.search(
            r'send_wa_parts_via_openclaw\(\)\s*\{(.+?)^\}',
            self.content, re.DOTALL | re.MULTILINE
        )
        self.assertIsNotNone(m, "找不到 send_wa_parts_via_openclaw 函数定义")
        body = m.group(1)
        self.assertIn("return 0", body,
            "V37.9.60-hotfix3: send_wa_parts_via_openclaw 末尾必须 `return 0` 防 bash quirk")

    def test_v37_9_60_hotfix3_marker_present(self):
        """V37.9.60-hotfix3 marker 必须在源码中可追溯."""
        self.assertIn("V37.9.60-hotfix3", self.content,
            "V37.9.60-hotfix3 marker 必须在 kb_deep_dive.sh 注释中追溯血案")

    def test_blood_lesson_references_silent_abort(self):
        """注释必须引用 5/8-5/12 silent 血案 + V37.9.21 quirk 来源."""
        self.assertIn("bash quirk", self.content,
            "V37.9.60-hotfix3: 必须引用 bash quirk 作为根因")
        # 引用 silent + last_run 或 5/8-5/12 之一作为血案证据
        has_blood_marker = any(
            keyword in self.content for keyword in
            ["silent", "5/8-5/12", "set -e 杀 caller", "last_run 停滞", "implicit return"]
        )
        self.assertTrue(has_blood_marker,
            "V37.9.60-hotfix3: 必须含血案锚点 (silent / 5/8-5/12 / set -e 杀 / etc)")


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