from __future__ import annotations

import json
import uuid

import pytest

from investibook.core import derive_workspace as dw
from investibook.db import queries as dbq


@pytest.fixture
def db(monkeypatch, tmp_path):
    p = tmp_path / "investibook.db"
    monkeypatch.setattr(dbq, "DB_PATH", p)
    monkeypatch.setattr(dw, "DB_PATH", p, raising=False)
    dbq._db_init()
    return p


def _insert_workspace(conn, wid: str, created_at: str) -> None:
    conn.execute(
        """
        INSERT INTO workspaces (id, user_id, name, description, is_example, created_at, updated_at)
        VALUES (?, 'u', ?, '', 0, ?, ?)
        """,
        (wid, wid, created_at, created_at),
    )


def _insert_jira(
    conn,
    issue_key: str,
    *,
    cluster_id: str = "",
    summary: str = "",
    raw_payload_json: str = "{}",
    created_e: int | None = None,
) -> None:
    now_e = created_e or dbq._utc_epoch_now()
    conn.execute(
        """
        INSERT INTO jira_tickets (
          issue_key, project_key, summary, status, issue_type, cluster_id,
          jira_created_at, jira_updated_at, raw_payload_json, first_seen_at, last_jira_sync_at
        ) VALUES (?, 'TCOC', ?, 'Open', 'Incident', ?, ?, ?, ?, ?, ?)
        """,
        (issue_key, summary, cluster_id, now_e, now_e, raw_payload_json, now_e, now_e),
    )


def test_incident_time_merge_eligible_only_content_sources():
    assert dw.incident_time_is_merge_eligible("description")
    assert dw.incident_time_is_merge_eligible("comment")
    assert dw.incident_time_is_merge_eligible("alert_field")
    assert not dw.incident_time_is_merge_eligible("jira_created_at")
    assert not dw.incident_time_is_merge_eligible("first_seen_at")
    assert not dw.incident_time_is_merge_eligible("now_fallback")


def test_merge_candidates_require_problem_occurrence_time(db):
    """Same cluster but only jira_created_at — must not be merge candidates."""
    src_id = str(uuid.uuid4())
    tgt_id = str(uuid.uuid4())
    cluster = "c-merge-test"
    base_e = 1_800_000_000
    tgt_e = base_e + 8 * 86400
    jira_only = '{"fields":{"description":"","comment":{"comments":[]}}}'
    with dbq._db_lock:
        conn = dbq._db_connect()
        for did, key, title, created_e in [
            (src_id, "TCOC-SRC", "source", base_e),
            (tgt_id, "TCOC-TGT", "target", tgt_e),
        ]:
            conn.execute(
                """
                INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
                VALUES (?, ?, 'active', NULL, ?, ?)
                """,
                (did, title, created_e, created_e),
            )
            _insert_jira(
                conn,
                key,
                cluster_id=cluster,
                summary=title,
                raw_payload_json=jira_only,
                created_e=base_e,
            )
            conn.execute(
                """
                INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
                VALUES (?, ?, 'primary', 'auto', ?)
                """,
                (did, key, created_e),
            )
        conn.commit()
        conn.close()

    dossier = {
        "dossier_id": src_id,
        "issues": ["TCOC-SRC"],
        "cluster_id": cluster,
        "effective_incident_at": base_e,
        "dossier_created_at": base_e,
        "incident_time_source": "jira_created_at",
        "incident_time_merge_eligible": False,
    }
    assert dw._find_merge_candidates_for_dossier(dossier) == []


def test_merge_candidates_fallback_uses_dossier_created_at_within_window(db):
    """Reimported DB without Jira description — same cluster + dossier created_at within ±7d."""
    src_id = str(uuid.uuid4())
    tgt_id = str(uuid.uuid4())
    cluster = "c-merge-fallback"
    base_e = 1_900_000_000
    jira_only = '{"fields":{"description":"","comment":{"comments":[]}}}'
    with dbq._db_lock:
        conn = dbq._db_connect()
        for did, key, title, created_e in [
            (src_id, "TCOC-FB-S", "source", base_e),
            (tgt_id, "TCOC-FB-T", "target", base_e + 2 * 86400),
        ]:
            conn.execute(
                """
                INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
                VALUES (?, ?, 'active', NULL, ?, ?)
                """,
                (did, title, created_e, created_e),
            )
            _insert_jira(
                conn,
                key,
                cluster_id=cluster,
                summary=title,
                raw_payload_json=jira_only,
                created_e=base_e,
            )
            conn.execute(
                """
                INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
                VALUES (?, ?, 'primary', 'auto', ?)
                """,
                (did, key, created_e),
            )
        conn.commit()
        conn.close()

    dossier = {
        "dossier_id": src_id,
        "issues": ["TCOC-FB-S"],
        "cluster_id": cluster,
        "effective_incident_at": base_e,
        "dossier_created_at": base_e,
        "incident_time_source": "jira_created_at",
        "incident_time_merge_eligible": False,
    }
    out = dw._find_merge_candidates_for_dossier(dossier)
    assert len(out) == 1
    assert out[0]["candidate_dossier_id"] == tgt_id
    assert out[0]["incident_time_approximate"] is True


def test_merge_candidates_filtered_by_problem_date_not_jira_created(db):
    src_id = str(uuid.uuid4())
    tgt_id = str(uuid.uuid4())
    cluster = "c-merge-date"
    src_desc = '{"fields":{"description":"problem on 2026-05-10","comment":{"comments":[]}}}'
    tgt_desc = '{"fields":{"description":"problem on 2026-05-20","comment":{"comments":[]}}}'
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'src', 'active', NULL, 1778900749, 1778900749)
            """,
            (src_id,),
        )
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'tgt', 'active', NULL, 1778900749, 1778900749)
            """,
            (tgt_id,),
        )
        _insert_jira(conn, "TCOC-SRC2", cluster_id=cluster, raw_payload_json=src_desc, created_e=1778900749)
        _insert_jira(conn, "TCOC-TGT2", cluster_id=cluster, raw_payload_json=tgt_desc, created_e=1778900749)
        conn.execute(
            "INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at) VALUES (?, 'TCOC-SRC2', 'primary', 'auto', ?)",
            (src_id, 1778900749),
        )
        conn.execute(
            "INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at) VALUES (?, 'TCOC-TGT2', 'primary', 'auto', ?)",
            (tgt_id, 1778900749),
        )
        conn.commit()
        conn.close()

    src_rows = [{"raw_payload_json": src_desc, "jira_created_at": 1778900749, "first_seen_at": 1778900749}]
    eff, src, eligible = dw.dossier_incident_time_from_issues(src_rows)
    assert eligible and src == "description"
    dossier = {
        "dossier_id": src_id,
        "issues": ["TCOC-SRC2"],
        "cluster_id": cluster,
        "effective_incident_at": eff,
        "incident_time_merge_eligible": eligible,
    }
    # Jira created_at identical, but problem dates 10 days apart → no merge candidate
    assert dw._find_merge_candidates_for_dossier(dossier) == []


def test_filter_workspace_excludes_cluster_only_candidate():
    """Workspace bound to TCOC-5240 must not be offered for dossier TCOC-5243 on cluster match alone."""
    dossier = {
        "issues": ["TCOC-5243"],
        "primary_issue_key": "TCOC-5243",
    }
    candidates = [
        {
            "workspace_id": "ws-5240",
            "workspace_name": "TCOC-5240_SaaS01_SlowQuery",
            "signals": ["message_cluster_reference", "source_cluster_reference"],
            "source_ticket_keys": ["TCOC-5240", "ST-2685"],
            "snippet": "slow query investigation for saas01",
            "assistant_msg_count": 3,
        }
    ]
    filtered = dw._filter_workspace_candidates_for_dossier(dossier, candidates)
    assert filtered == []


def test_parse_unsupported_non_runtime_tidb_category():
    parsed = {
        "actions": [
            {
                "type": "mark_unsupported_incident",
                "category": "non_runtime_tidb",
                "reason": "Cannot select 64 vCPU tier is a product capability fault, not TiDB runtime diagnosis",
            }
        ]
    }
    actions = dw._parse_llm_derive_actions(
        parsed,
        dossier_id="d1",
        allowed_workspace_ids=set(),
        allowed_merge_targets=set(),
    )
    assert actions[0]["type"] == "mark_unsupported_incident"
    assert actions[0]["category"] == "non_runtime_tidb"


def test_no_analysis_needed_blocked_when_workspace_linked():
    dossier = {
        "jira_issues": [{"status": "Resolved"}],
        "linked_workspace_ids": ["ws-1"],
    }
    actions = [
        {"type": "mark_no_analysis_needed", "category": "already_analyzed", "reason": "has RCA"},
    ]
    out = dw._filter_classification_actions(dossier, actions, linked_workspace_ids=["ws-1"])
    assert out == []


def test_dossier_has_settled_classification():
    assert dw._dossier_has_settled_classification({"manual_not_incident": True}) is True
    assert (
        dw._dossier_has_settled_classification(
            {"derive_mark": {"mark_type": "unsupported_incident", "category": "non_runtime_tidb"}}
        )
        is True
    )
    assert (
        dw._dossier_has_settled_classification(
            {"linked_workspace_ids_manual": ["ws-1"], "linked_workspace_ids_derived": []}
        )
        is False
    )
    assert dw._dossier_has_settled_classification({}) is False


def test_dossier_signals_runtime_diagnosis_oom():
    dossier = {
        "title": "Query failed because of OOM",
        "jira_summary": "customer query OOM on TiDB",
        "jira_issues": [],
    }
    assert dw._dossier_signals_runtime_diagnosis(dossier) is True


def test_filter_unsupported_stripped_for_runtime_oom():
    dossier = {"title": "Query failed because of OOM", "manual_not_incident": False}
    actions = [
        {
            "type": "mark_unsupported_incident",
            "category": "non_runtime_tidb",
            "reason": "LLM wrongly said unsupported",
        },
    ]
    out = dw._filter_classification_actions(dossier, actions, linked_workspace_ids=[])
    assert out == []


def test_dossier_signals_runtime_diagnosis_optimizer_wrong_index():
    title = (
        "v8.5.4 optimizer picks wrong index on taggings table inside correlated subquery "
        "(Apply), causing massive row estimate blowup vs v8.1.1"
    )
    dossier = {"title": title, "jira_issues": [{"summary": title, "description": ""}]}
    assert dw._dossier_signals_runtime_diagnosis(dossier) is True


def test_dossier_signals_runtime_diagnosis_chinese_query_timeout():
    title = "SAAS活动派发相关查询执行超时问题咨询"
    dossier = {"title": title, "jira_issues": [{"summary": title, "description": ""}]}
    assert dw._dossier_signals_runtime_diagnosis(dossier) is True


def test_dossier_signals_runtime_diagnosis_japanese_slow_query():
    title = "クエリ実行に時間がかかる事象が断続的に発生したため、原因調査のご依頼"
    dossier = {"title": title, "jira_issues": [{"summary": title, "description": ""}]}
    assert dw._dossier_signals_runtime_diagnosis(dossier) is True


def test_filter_unsupported_stripped_for_optimizer_wrong_index():
    title = (
        "v8.5.4 optimizer picks wrong index on taggings table inside correlated subquery "
        "(Apply), causing massive row estimate blowup vs v8.1.1"
    )
    dossier = {"title": title, "manual_not_incident": False}
    actions = [
        {
            "type": "mark_unsupported_incident",
            "category": "non_runtime_tidb",
            "reason": "product capability issue",
        },
    ]
    out = dw._filter_classification_actions(dossier, actions, linked_workspace_ids=[])
    assert out == []


def test_filter_keeps_unsupported_for_datadog_metrics_integration():
    title = (
        "DataDog - Inconsistencies in TiDB integration data are resulting in false alerts."
    )
    dossier = {"title": title, "jira_issues": [{"summary": title, "description": ""}]}
    assert dw._dossier_signals_platform_integration(dossier)
    assert dw._dossier_signals_runtime_diagnosis(dossier)
    actions = [
        {
            "type": "mark_unsupported_incident",
            "category": "other_non_tidb",
            "reason": "Datadog metrics integration false alerts",
        },
    ]
    out = dw._filter_classification_actions(dossier, actions, linked_workspace_ids=[])
    assert len(out) == 1
    assert out[0]["category"] == "other_non_tidb"


def test_dossier_signals_platform_integration_chinese_metrics_integration():
    title = "TiDB监控集成指标数据不一致导致误告警"
    dossier = {"title": title, "jira_issues": [{"summary": title, "description": ""}]}
    assert dw._dossier_signals_platform_integration(dossier) is True


def test_filter_classification_strips_when_already_analyzed():
    dossier = {"manual_not_incident": False}
    actions = [
        {"type": "mark_unsupported_incident", "category": "non_runtime_tidb", "reason": "x"},
        {"type": "link_workspaces", "workspace_ids": ["ws-2"], "reason": "more"},
    ]
    out = dw._filter_classification_actions(dossier, actions, linked_workspace_ids=["ws-1"])
    assert len(out) == 1
    assert out[0]["type"] == "link_workspaces"


@pytest.mark.asyncio
async def test_analyze_does_not_skip_dossier_with_linked_workspace(db, monkeypatch):
    """Linked workspace blocks re-classification but analyze still runs for more links."""
    did = str(uuid.uuid4())
    now_e = dbq._utc_epoch_now()
    ws_id = f"ws-{uuid.uuid4().hex[:8]}"
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'Analyzed inc', 'active', NULL, ?, ?)
            """,
            (did, now_e, now_e),
        )
        _insert_jira(conn, "TCOC-AN", summary="already done")
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, 'TCOC-AN', 'primary', 'auto', ?)
            """,
            (did, now_e),
        )
        _insert_workspace(conn, ws_id, dbq._utc_now())
        conn.execute(
            """
            INSERT INTO incident_dossier_workspaces (dossier_id, workspace_id, created_at, created_by)
            VALUES (?, ?, ?, 'u')
            """,
            (did, ws_id, now_e),
        )
        conn.commit()
        conn.close()

    async def fake_llm(*_a, **_k):
        return (
            [{"type": "mark_unsupported_incident", "category": "non_runtime_tidb", "reason": "wrong"}],
            None,
            {"input_tokens": 0, "output_tokens": 0},
        )

    monkeypatch.setattr(dw, "_llm_propose_actions_for_dossier", fake_llm)
    scope = dw.AnalyzeScope(
        dossier_ids=[did],
        metric="",
        day="",
        project="ALL",
        granularity="week",
        range_from=None,
        range_to=None,
    )
    out = await dw.analyze(scope)
    assert out["summary"]["dossiers_skipped_prior_state"] == 0
    assert out["proposals"] == []


@pytest.mark.asyncio
async def test_analyze_skips_dossier_with_derive_mark_unsupported(db, monkeypatch):
    """TCOC-5275-style: already marked unsupported — omit from next analyze."""
    did = str(uuid.uuid4())
    now_e = dbq._utc_epoch_now()
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'Need scale out PD', 'active', NULL, ?, ?)
            """,
            (did, now_e, now_e),
        )
        _insert_jira(conn, "TCOC-5275", summary="Need scale out and in PD node")
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, 'TCOC-5275', 'primary', 'auto', ?)
            """,
            (did, now_e),
        )
        conn.commit()
        conn.close()
    dbq._db_upsert_incident_dossier_derive_mark(
        did,
        mark_type="unsupported_incident",
        category="non_runtime_tidb",
        reason="product operation",
        actor_user_id="u1",
    )

    async def fake_llm(*_a, **_k):
        return (
            [
                {
                    "type": "mark_unsupported_incident",
                    "category": "non_runtime_tidb",
                    "reason": "duplicate",
                }
            ],
            None,
            {"input_tokens": 0, "output_tokens": 0},
        )

    monkeypatch.setattr(dw, "_llm_propose_actions_for_dossier", fake_llm)

    scope = dw.AnalyzeScope(
        dossier_ids=[did],
        metric="",
        day="",
        project="ALL",
        granularity="week",
        range_from=None,
        range_to=None,
    )
    out = await dw.analyze(scope)
    assert out["proposals"] == []
    assert out["summary"]["dossiers_skipped_prior_state"] == 1


def test_no_analysis_needed_never_proposed_from_jira_status():
    """Resolved/closed Jira must not auto-mark no_analysis_needed (e.g. TCOC-5279)."""
    for status in ("PENDING FOR SUPPORT", "Resolved", "Closed", "Done"):
        dossier = {"jira_issues": [{"status": status, "summary": "incident"}]}
        assert dw._may_propose_no_analysis_needed(dossier, linked_workspace_ids=[]) is False
        actions = [{"type": "mark_no_analysis_needed", "category": "already_analyzed", "reason": "rca in jira"}]
        assert dw._filter_classification_actions(dossier, actions, linked_workspace_ids=[]) == []


def test_dedupe_prefers_unsupported_over_not_incident():
    actions = [
        {"type": "mark_not_incident", "reason": "configuration issue"},
        {"type": "mark_unsupported_incident", "category": "non_runtime_tidb", "reason": "real fault but not runtime"},
    ]
    out = dw._dedupe_classification(actions)
    assert len(out) == 1
    assert out[0]["type"] == "mark_unsupported_incident"


def test_filter_workspace_keeps_direct_issue_match():
    dossier = {"issues": ["TCOC-5243"], "primary_issue_key": "TCOC-5243"}
    candidates = [
        {
            "workspace_id": "ws-5243",
            "signals": ["source_issue_reference"],
            "source_ticket_keys": ["TCOC-5243"],
        }
    ]
    filtered = dw._filter_workspace_candidates_for_dossier(dossier, candidates)
    assert len(filtered) == 1
    assert filtered[0]["link_strength"] == "direct_issue"


def test_resolve_effective_incident_time_prefers_description_date(db):
    row = {
        "jira_created_at": 1_800_000_000,
        "first_seen_at": 1_800_000_100,
        "raw_payload_json": '{"fields":{"description":"incident started at 2026-05-12 01:02"}}',
    }
    e, src = dw.resolve_effective_incident_time(row)
    assert src == "description"
    assert e < 1_800_000_000


@pytest.mark.asyncio
async def test_apply_payload_hash_async(db):
    did = str(uuid.uuid4())
    now_e = dbq._utc_epoch_now()
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'd', 'active', NULL, ?, ?)
            """,
            (did, now_e, now_e),
        )
        conn.commit()
        conn.close()
    action = {
        "type": "mark_not_incident",
        "reason": "x",
        "action_id": "act-1",
        "action_hash": "bad-hash",
    }
    out = await dw.apply_selected(selections=[{"dossier_id": did, "actions": [action]}], actor_user_id="u")
    assert out["applied"] == []
    assert out["skipped"]
    row = dbq._db_get_incident_dossier_row(did)
    assert row is not None
    assert int(row["manual_not_incident"] or 0) == 0


@pytest.mark.asyncio
async def test_apply_hash_survives_pydantic_model_dump(db):
    did = str(uuid.uuid4())
    now_e = dbq._utc_epoch_now()
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'd', 'active', NULL, ?, ?)
            """,
            (did, now_e, now_e),
        )
        conn.commit()
        conn.close()
    from investibook.models.requests import DeriveAction

    normalized = dw._normalize_action(
        did, {"type": "mark_not_incident", "category": "", "reason": "consultation"}
    )
    action = DeriveAction.model_validate(normalized).model_dump()
    out = await dw.apply_selected(
        selections=[{"dossier_id": did, "actions": [action]}],
        actor_user_id="u",
    )
    assert len(out["applied"]) == 1
    row = dbq._db_get_incident_dossier_row(did)
    assert int(row["manual_not_incident"] or 0) == 1


def test_batch_workspace_candidates_and_lark_message_signal(db):
    did = str(uuid.uuid4())
    issue_key = "TCOC-123"
    eff = 1_800_000_000
    created_iso = "2027-01-15T00:00:00+00:00"
    ws1 = f"lark-ws-{uuid.uuid4().hex[:8]}"
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'd', 'active', NULL, ?, ?)
            """,
            (did, eff, eff),
        )
        _insert_jira(conn, issue_key, cluster_id="c1", summary="s", created_e=eff)
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, ?, 'primary', 'auto', ?)
            """,
            (did, issue_key, eff),
        )
        _insert_workspace(conn, ws1, created_iso)
        conn.execute(
            "INSERT INTO workspace_sources(workspace_id, data) VALUES (?, ?)",
            (
                ws1,
                '[{"type":"jira_issue","issueId":"TCOC-123"},{"type":"cluster_meta","meta":{"cluster_id":"c1"}}]',
            ),
        )
        conn.execute(
            """
            INSERT INTO messages_v2 (id, workspace_id, role, content, parts, status, issue_info, attachments, created_at, updated_at)
            VALUES ('m1', ?, 'assistant', 'related to TCOC-123 and cluster c1', '', 'completed', '', '', ?, ?)
            """,
            (ws1, created_iso, created_iso),
        )
        conn.commit()
        conn.close()

    out = dbq._db_batch_find_workspace_candidates_for_worklist(
        [
            {
                "dossier_id": did,
                "issue_keys": [issue_key],
                "cluster_id": "c1",
                "effective_incident_at": eff,
            }
        ]
    )
    assert did in out
    assert out[did]
    assert out[did][0]["workspace_id"] == ws1
    assert "source_issue_reference" in out[did][0]["signals"] or "message_issue_reference" in out[did][0]["signals"]


def test_message_jira_browse_url_without_jira_source(db):
    """User pastes Jira browse URL in chat — no workspace_jira_links or jira source."""
    did = str(uuid.uuid4())
    issue_key = "TCOC-456"
    eff = 1_800_000_000
    created_iso = "2027-01-15T00:00:00+00:00"
    ws = f"msg-url-ws-{uuid.uuid4().hex[:8]}"
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'd', 'active', NULL, ?, ?)
            """,
            (did, eff, eff),
        )
        _insert_jira(conn, issue_key, cluster_id="c-url", summary="OOM on cluster", created_e=eff)
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, ?, 'primary', 'auto', ?)
            """,
            (did, issue_key, eff),
        )
        _insert_workspace(conn, ws, created_iso)
        conn.execute(
            """
            INSERT INTO messages_v2 (id, workspace_id, role, content, parts, status, issue_info, attachments, created_at, updated_at)
            VALUES ('u1', ?, 'user', ?, '', 'completed', '', '', ?, ?)
            """,
            (
                ws,
                f"Please check https://jira.example.com/browse/{issue_key} for cluster c-url",
                created_iso,
                created_iso,
            ),
        )
        conn.commit()
        conn.close()

    out = dbq._db_batch_find_workspace_candidates_for_worklist(
        [
            {
                "dossier_id": did,
                "issue_keys": [issue_key],
                "cluster_id": "c-url",
                "effective_incident_at": eff,
                "problem_texts": ["OOM on cluster"],
            }
        ]
    )
    assert out[did]
    assert out[did][0]["workspace_id"] == ws
    assert "message_issue_reference" in out[did][0]["signals"]
    assert out[did][0]["time_gate_rule"] == "china_same_day_message"


def test_message_same_problem_same_cluster_same_china_day(db):
    did = str(uuid.uuid4())
    issue_key = "TCOC-789"
    eff = 1_800_000_000
    created_iso = "2027-01-15T00:00:00+00:00"
    ws = f"msg-prob-ws-{uuid.uuid4().hex[:8]}"
    problem = "TiKV high CPU usage on cluster prod-cluster-99"
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, ?, 'active', NULL, ?, ?)
            """,
            (did, problem, eff, eff),
        )
        _insert_jira(conn, issue_key, cluster_id="prod-cluster-99", summary=problem, created_e=eff)
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, ?, 'primary', 'auto', ?)
            """,
            (did, issue_key, eff),
        )
        _insert_workspace(conn, ws, created_iso)
        conn.execute(
            """
            INSERT INTO messages_v2 (id, workspace_id, role, content, parts, status, issue_info, attachments, created_at, updated_at)
            VALUES ('u2', ?, 'user', ?, '', 'completed', '', '', ?, ?)
            """,
            (
                ws,
                "Diagnose TiKV high CPU usage on cluster prod-cluster-99 today",
                created_iso,
                created_iso,
            ),
        )
        conn.commit()
        conn.close()

    out = dbq._db_batch_find_workspace_candidates_for_worklist(
        [
            {
                "dossier_id": did,
                "issue_keys": [issue_key],
                "cluster_id": "prod-cluster-99",
                "effective_incident_at": eff,
                "problem_texts": [problem],
            }
        ]
    )
    assert out[did]
    assert out[did][0]["workspace_id"] == ws
    assert "message_same_problem_same_day" in out[did][0]["signals"]


def test_message_same_problem_different_china_day_no_match(db):
    did = str(uuid.uuid4())
    issue_key = "TCOC-790"
    eff = 1_800_000_000
    msg_iso = "2027-01-18T00:00:00+00:00"
    ws = f"msg-day-ws-{uuid.uuid4().hex[:8]}"
    problem = "TiKV high CPU usage on cluster prod-cluster-99"
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, ?, 'active', NULL, ?, ?)
            """,
            (did, problem, eff, eff),
        )
        _insert_jira(conn, issue_key, cluster_id="prod-cluster-99", summary=problem, created_e=eff)
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, ?, 'primary', 'auto', ?)
            """,
            (did, issue_key, eff),
        )
        _insert_workspace(conn, ws, msg_iso)
        conn.execute(
            """
            INSERT INTO messages_v2 (id, workspace_id, role, content, parts, status, issue_info, attachments, created_at, updated_at)
            VALUES ('u3', ?, 'user', ?, '', 'completed', '', '', ?, ?)
            """,
            (
                ws,
                "Diagnose TiKV high CPU usage on cluster prod-cluster-99 today",
                msg_iso,
                msg_iso,
            ),
        )
        conn.commit()
        conn.close()

    out = dbq._db_batch_find_workspace_candidates_for_worklist(
        [
            {
                "dossier_id": did,
                "issue_keys": [issue_key],
                "cluster_id": "prod-cluster-99",
                "effective_incident_at": eff,
                "problem_texts": [problem],
            }
        ]
    )
    assert did in out
    assert out[did] == []


def test_extract_tcoc_keys_from_text():
    keys = dbq._extract_tcoc_keys_from_text(
        "see https://jira.example.com/browse/TCOC-100 and also TCOC-200 in text"
    )
    assert keys == {"TCOC-100", "TCOC-200"}


def test_message_same_instance_same_cluster_matches_lark_style(db):
    """Lark chat: cluster + tikv-79 in message, dossier title tikv-79 CPU 高 — no TCOC in chat."""
    did = str(uuid.uuid4())
    issue_key = "TCOC-5243"
    eff = 1778562952
    created_iso = "2026-05-12T07:39:57+00:00"
    ws = "lark-ws-test-tikv79"
    cluster = "10989049060142230334"
    problem = "[wetech]tikv-79 CPU 高，需要协助分析根因"
    user_msg = (
        f"查找 saas01 {cluster} 里 5/11 15:30-15:40 造成 tikv-79 cpu 热点的 sql，"
        "并给出 sql 的执行计划，执行频次。"
    )
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, ?, 'active', NULL, ?, ?)
            """,
            (did, problem, eff, eff),
        )
        _insert_jira(conn, issue_key, cluster_id=cluster, summary=problem, created_e=eff)
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, ?, 'primary', 'auto', ?)
            """,
            (did, issue_key, eff),
        )
        _insert_workspace(conn, ws, created_iso)
        conn.execute(
            """
            INSERT INTO messages_v2 (id, workspace_id, role, content, parts, status, issue_info, attachments, created_at, updated_at)
            VALUES ('u-lark', ?, 'user', ?, '', 'completed', '', '', ?, ?)
            """,
            (ws, user_msg, created_iso, created_iso),
        )
        conn.execute(
            """
            INSERT INTO messages_v2 (id, workspace_id, role, content, parts, status, issue_info, attachments, created_at, updated_at)
            VALUES ('a-lark', ?, 'assistant', 'analysis done', '', 'completed', '', '', ?, ?)
            """,
            (ws, created_iso, created_iso),
        )
        conn.commit()
        conn.close()

    out = dbq._db_batch_find_workspace_candidates_for_worklist(
        [
            {
                "dossier_id": did,
                "issue_keys": [issue_key],
                "cluster_id": cluster,
                "effective_incident_at": eff,
                "problem_texts": [problem],
            }
        ]
    )
    assert out[did]
    assert out[did][0]["workspace_id"] == ws
    assert "message_same_problem_same_day" in out[did][0]["signals"]


def test_jira_issue_detail_includes_full_description_and_comments(db):
    with dbq._db_lock:
        conn = dbq._db_connect()
        _insert_jira(
            conn,
            "TCOC-999",
            summary="quota increase",
            raw_payload_json=json.dumps(
                {
                    "fields": {
                        "description": "Please add whitelist for tenant foo",
                        "comment": {
                            "comments": [
                                {"body": "This is a billing question not an outage"},
                            ]
                        },
                        "labels": ["billing"],
                    }
                }
            ),
        )
        row = conn.execute(
            "SELECT issue_key, summary, status, issue_type, cluster_id, tenant_id, "
            "jira_created_at, jira_updated_at, raw_payload_json FROM jira_tickets WHERE issue_key = ?",
            ("TCOC-999",),
        ).fetchone()
        conn.commit()
        conn.close()
    detail = dw._jira_issue_detail_for_llm(dict(row))
    assert "whitelist" in detail["description"]
    assert detail["comments"] == ["This is a billing question not an outage"]
    assert detail["labels"] == ["billing"]


def test_parse_llm_derive_actions_filters_to_allowed_candidates():
    parsed = {
        "actions": [
            {"type": "mark_not_incident", "reason": "consultation only"},
            {"type": "link_workspaces", "workspace_ids": ["ws-ok", "ws-bad"], "reason": "related"},
            {
                "type": "merge_into",
                "target_dossier_id": "d-target",
                "reason": "duplicate tracking",
                "same_problem_confidence": 0.91,
            },
        ]
    }
    actions = dw._parse_llm_derive_actions(
        parsed,
        dossier_id="d-src",
        allowed_workspace_ids={"ws-ok"},
        allowed_merge_targets={"d-target"},
    )
    types = [a["type"] for a in actions]
    assert types == ["mark_not_incident", "link_workspaces", "merge_into"]
    assert actions[1]["workspace_ids"] == ["ws-ok"]
    assert actions[2]["target_dossier_id"] == "d-target"


@pytest.mark.asyncio
async def test_analyze_uses_llm_not_keywords(db, monkeypatch):
    did = str(uuid.uuid4())
    now_e = dbq._utc_epoch_now()
    raw = json.dumps(
        {
            "fields": {
                "description": "请加白名单",
                "comment": {"comments": []},
            }
        }
    )
    with dbq._db_lock:
        conn = dbq._db_connect()
        conn.execute(
            """
            INSERT INTO incident_dossiers (id, title, status, merged_into_dossier_id, created_at, updated_at)
            VALUES (?, 'd', 'active', NULL, ?, ?)
            """,
            (did, now_e, now_e),
        )
        _insert_jira(conn, "TCOC-WL", summary="whitelist request", raw_payload_json=raw, created_e=now_e)
        conn.execute(
            """
            INSERT INTO incident_dossier_issues (dossier_id, issue_key, link_role, link_confidence, created_at)
            VALUES (?, 'TCOC-WL', 'primary', 'auto', ?)
            """,
            (did, now_e),
        )
        conn.commit()
        conn.close()

    llm_calls: list[str] = []

    async def _fake_llm(prompt: str):
        llm_calls.append(prompt)
        return (
            {
                "actions": [
                    {"type": "mark_not_incident", "reason": "Whitelist request is not an incident"},
                ]
            },
            None,
            {"input_tokens": 1200, "output_tokens": 80},
        )

    monkeypatch.setattr(dw, "_call_derive_llm", _fake_llm)
    out = await dw.analyze(
        dw.AnalyzeScope(
            dossier_ids=[did],
            metric="",
            day="",
            project="ALL",
            granularity="week",
            range_from=None,
            range_to=None,
        )
    )
    assert len(llm_calls) == 1
    assert "请加白名单" in llm_calls[0]
    assert "jira_issues" in llm_calls[0]
    assert out["proposals"]
    assert out["proposals"][0]["actions"][0]["type"] == "mark_not_incident"
    assert out["summary"]["llm"]["input_tokens"] == 1200
    assert out["summary"]["llm"]["output_tokens"] == 80