from __future__ import annotations

import asyncio
import importlib
import json
from contextlib import contextmanager
from datetime import datetime
from decimal import Decimal
from unittest.mock import patch

import pytest
from sqlalchemy import create_engine, event, select
from sqlalchemy.orm import Session, sessionmaker

from agents.find_agent import create_find_agent
from agents.find_agent.async_runner import arun_find_agent
from agents.find_agent.demand_run import (
    FindDemandContext,
    FindDemandPoint,
    FindDemandVideo,
    build_find_agent_user_input,
    prepare_video_discovery_run,
)
from agents.find_agent.tools import video_discovery_store
from agents.find_agent.tools.batch_search_and_record import batch_search_and_record
from agents.find_agent.support.search_persistence import _candidate_from_search_result
from agents.find_agent.support.douyin_search import (
    DEFAULT_MIN_DURATION_SECONDS as INTERNAL_SEARCH_MIN_DURATION,
)
from agents.find_agent.support.douyin_search_tikhub import (
    DEFAULT_MIN_DURATION_SECONDS as TIKHUB_SEARCH_MIN_DURATION,
)
from agents.find_agent.support.douyin_user_videos import (
    DEFAULT_MIN_DURATION_SECONDS as AUTHOR_SEARCH_MIN_DURATION,
)
from supply_infra.video_discovery_gates import evaluate_candidate_gate
from supply_infra.db.models.video_discovery import (
    VideoDiscoveryCandidate,
    VideoDiscoveryEvidence,
    VideoDiscoveryGateEvaluation,
    VideoDiscoveryRun,
    VideoDiscoverySearch,
)
from supply_infra.db.repositories.video_discovery_repo import (
    VideoDiscoveryRepository,
)
from supply_infra.scheduler.jobs.discover_videos_from_demands import (
    discover_videos_from_demands,
)


def _expire_on_commit_session_factory() -> sessionmaker[Session]:
    """Match production get_session: commit expires ORM instances."""
    engine = create_engine("sqlite+pysqlite:///:memory:")
    # SQLite 对 BigInteger PK 不会自增，测试里显式写入 id。
    VideoDiscoveryRun.__table__.create(engine)
    VideoDiscoverySearch.__table__.create(engine)
    VideoDiscoveryCandidate.__table__.create(engine)
    VideoDiscoveryEvidence.__table__.create(engine)
    VideoDiscoveryGateEvaluation.__table__.create(engine)
    return sessionmaker(bind=engine, autoflush=False, autocommit=False)


def _patch_service_session(
    monkeypatch: pytest.MonkeyPatch,
    factory: sessionmaker[Session],
) -> None:
    @contextmanager
    def get_test_session():
        session = factory()
        try:
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

    monkeypatch.setattr(
        "supply_infra.services.video_discovery_service.get_session",
        get_test_session,
    )
    import supply_infra.services.video_discovery_service as service_module

    service_module._default_service = None


def _seed_run(
    factory: sessionmaker[Session],
    *,
    run_id: str,
    demand_grade_id: int,
    status: str = "running",
    row_id: int = 1,
) -> None:
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=row_id,
                run_id=run_id,
                biz_dt="20260728",
                demand_grade_id=demand_grade_id,
                demand_word="广场舞",
                seed_video_id="vid-1",
                seed_video_title="参考标题",
                relevant_points_json="[]",
                status=status,
            )
        )
        session.commit()


def test_create_video_discovery_run_reuses_precreated_run(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """定时任务预创建 run 后，Agent 复用 run_id 不得触发 DetachedInstanceError。"""
    factory = _expire_on_commit_session_factory()
    _patch_service_session(monkeypatch, factory)
    _seed_run(factory, run_id="precreated-run", demand_grade_id=101)

    payload = json.loads(
        video_discovery_store.create_video_discovery_run(
            demand_word="广场舞",
            seed_video_title="参考标题",
            relevant_points=[],
            run_id="precreated-run",
            demand_grade_id=101,
        )
    )

    assert "error" not in payload
    assert payload["run_id"] == "precreated-run"
    assert payload["status"] == "running"
    assert payload["pre_created"] is True


def test_prepare_then_reuse_run_id_scheduled_flow(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """完整调度衔接：prepare 得到 run_id → create 复用，全程无 DetachedInstanceError。"""
    factory = _expire_on_commit_session_factory()
    _patch_service_session(monkeypatch, factory)
    # 先写入一条 failed 记录，prepare 会原地重置为 running 并复用 run_id。
    _seed_run(
        factory,
        run_id="scheduled-run",
        demand_grade_id=202,
        status="failed",
    )
    ctx = FindDemandContext(
        biz_dt="20260728",
        demand_grade_id=202,
        demand_name="广场舞",
        grade="S",
        videos=[
            FindDemandVideo(
                video_id="vid-1",
                title="参考标题",
                points=[
                    FindDemandPoint(
                        point="动作简单",
                        point_type="key",
                    )
                ],
            )
        ],
    )

    run_id, skip_reason = prepare_video_discovery_run(ctx)
    assert skip_reason is None
    assert run_id == "scheduled-run"

    reuse_id, reuse_skip = prepare_video_discovery_run(ctx)
    assert reuse_id is None
    assert reuse_skip is not None
    assert "attempt_count=1" in reuse_skip

    forced_id, forced_skip = prepare_video_discovery_run(ctx, force=True)
    assert forced_id == "scheduled-run"
    assert forced_skip is None
    from supply_infra.services.video_discovery_service import (
        get_video_discovery_service,
    )

    forced_run = get_video_discovery_service().lookup_run("scheduled-run")
    assert forced_run is not None
    assert forced_run["attempt_count"] == 2

    payload = json.loads(
        video_discovery_store.create_video_discovery_run(
            demand_word=ctx.demand_name,
            seed_video_title="参考标题",
            relevant_points=[{"point": "动作简单", "point_type": "key"}],
            run_id=run_id,
            demand_grade_id=ctx.demand_grade_id,
            seed_video_id="vid-1",
        )
    )
    assert "error" not in payload
    assert payload["pre_created"] is True
    assert payload["run_id"] == run_id


def test_prepare_skips_when_run_already_finished(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    factory = _expire_on_commit_session_factory()
    _patch_service_session(monkeypatch, factory)
    _seed_run(
        factory,
        run_id="finished-run",
        demand_grade_id=303,
        status="finished",
    )
    ctx = FindDemandContext(
        biz_dt="20260728",
        demand_grade_id=303,
        demand_name="广场舞",
        grade="S",
        videos=[
            FindDemandVideo(
                video_id="vid-1",
                title="参考标题",
                points=[FindDemandPoint(point="动作简单", point_type="key")],
            )
        ],
    )

    run_id, skip_reason = prepare_video_discovery_run(ctx)
    assert run_id is None
    assert skip_reason is not None
    assert "finished" in skip_reason


def test_video_discovery_models_exclude_unused_columns() -> None:
    run_columns = set(VideoDiscoveryRun.__table__.columns.keys())
    candidate_columns = set(VideoDiscoveryCandidate.__table__.columns.keys())

    assert "backup_count" not in run_columns
    assert {
        "video_url",
        "content_analysis",
        "content_analysis_verified",
        "hit_points_json",
        "publish_timestamp",
        "detail_verified",
        "content_portrait_attempted",
        "account_portrait_attempted",
        "age_portraits_normalized",
        "expansion_worthy_tags_json",
        "confidence",
        "relevance_reason",
        "elder_reason",
        "share_reason",
        "manual_review_note",
        "manual_review_status",
    }.isdisjoint(candidate_columns)
    assert "search_id" in candidate_columns
    candidate_constraints = {
        constraint.name
        for constraint in VideoDiscoveryCandidate.__table__.constraints
    }
    search_constraints = {
        constraint.name
        for constraint in VideoDiscoverySearch.__table__.constraints
    }
    assert "uk_video_discovery_candidate_run_aweme" not in candidate_constraints
    assert "fk_video_discovery_candidate_search" in candidate_constraints
    assert "uk_video_discovery_search_key" not in search_constraints


def test_create_find_agent_registers_discovery_tools() -> None:
    agent = create_find_agent()

    assert agent.name == "find_agent"
    assert "batch_search_and_record" in agent.tools.list_tools()
    assert "batch_update_video_discovery_candidates" in agent.tools.list_tools()
    assert "update_video_discovery_run_status" in agent.tools.list_tools()
    assert "create_video_discovery_run" not in agent.tools.list_tools()
    assert "batch_save_video_candidate_evaluations" not in agent.tools.list_tools()
    assert "audit_video_discovery_run" not in agent.tools.list_tools()
    assert "query_video_discovery_state" in agent.tools.list_tools()


def test_find_agent_input_uses_reference_videos_without_seed_fields() -> None:
    ctx = FindDemandContext(
        biz_dt="20260729",
        demand_grade_id=101,
        demand_name="广场舞",
        grade="S",
        videos=[
            FindDemandVideo(
                video_id="vid-1",
                title="参考标题",
                points=[FindDemandPoint(point="动作简单", point_type="key")],
            )
        ],
    )

    user_input = build_find_agent_user_input(ctx, "scheduled-run")

    assert "seed_video_id：" not in user_input
    assert "seed_video_title：" not in user_input
    assert "reference_videos：" in user_input
    assert '"video_id": "vid-1"' in user_input
    assert '"title": "参考标题"' in user_input
    assert "create_video_discovery_run" not in user_input
    assert "relevant_points" not in user_input
    assert "current_datetime：" in user_input
    assert "timezone：Asia/Shanghai" in user_input
    assert "quality_gate_rules：" in user_input


_P0_RULES = {
    "rule_version": "test-p0",
    "timezone": "Asia/Shanghai",
    "current_datetime": "2026-07-31T12:00:00+08:00",
    "current_date": "2026-07-31",
    "min_duration_seconds": 30,
    "min_share_count": 1000,
    "min_content_50_plus_ratio": 0.20,
    "min_account_50_plus_ratio": 0.20,
    "festival_lead_days": 7,
    "event_max_age_days": 7,
    "seasonal_max_age_days": 180,
}


def _p0_candidate(**overrides):
    candidate = {
        "title": "适合家庭分享的生活技巧",
        "publish_at": "2026-07-31T09:00:00+08:00",
        "duration_seconds": 30,
        "share_count": 1000,
        "content_50_plus_ratio": 0.20,
        "account_50_plus_ratio": 0.30,
    }
    candidate.update(overrides)
    return candidate


def test_p0_gate_enforces_boundaries_and_time_context() -> None:
    assert evaluate_candidate_gate(_p0_candidate(), _P0_RULES)["primary_eligible"]

    short = evaluate_candidate_gate(
        _p0_candidate(duration_seconds=Decimal("29.999")),
        _P0_RULES,
    )
    assert "DURATION_TOO_SHORT" in short["failed_reason_codes"]

    low_share = evaluate_candidate_gate(_p0_candidate(share_count=999), _P0_RULES)
    assert "SHARE_COUNT_TOO_LOW" in low_share["failed_reason_codes"]

    low_elder = evaluate_candidate_gate(
        _p0_candidate(content_50_plus_ratio=0.199, account_50_plus_ratio=0.19),
        _P0_RULES,
    )
    assert "PORTRAIT_50_PLUS_TOO_LOW" in low_elder["failed_reason_codes"]

    morning = evaluate_candidate_gate(
        _p0_candidate(title="早上好，送给家人的祝福"),
        _P0_RULES,
    )
    assert "DAYPART_EXPIRED" in morning["failed_reason_codes"]

    festival = evaluate_candidate_gate(
        _p0_candidate(title="春节祝福送给全家"),
        _P0_RULES,
    )
    assert "FESTIVAL_OUT_OF_WINDOW" in festival["failed_reason_codes"]


def test_p0_gate_keeps_content_and_account_portraits_separate() -> None:
    conflict = evaluate_candidate_gate(
        _p0_candidate(
            content_50_plus_ratio=0.28,
            account_50_plus_ratio=0.08,
        ),
        _P0_RULES,
    )
    assert conflict["content_portrait_status"] == "pass"
    assert conflict["account_portrait_status"] == "fail"
    assert conflict["portrait_conflict"] is True
    assert conflict["primary_eligible"] is True

    account_only = evaluate_candidate_gate(
        _p0_candidate(
            content_50_plus_ratio=None,
            account_50_plus_ratio=0.55,
        ),
        _P0_RULES,
    )
    assert account_only["primary_eligible"] is True
    assert "CONTENT_PORTRAIT_MISSING" not in account_only["failed_reason_codes"]

    content_only = evaluate_candidate_gate(
        _p0_candidate(
            content_50_plus_ratio=0.28,
            account_50_plus_ratio=None,
        ),
        _P0_RULES,
    )
    assert content_only["primary_eligible"] is True

    one_side_low = evaluate_candidate_gate(
        _p0_candidate(
            content_50_plus_ratio=None,
            account_50_plus_ratio=0.08,
        ),
        _P0_RULES,
    )
    assert one_side_low["primary_eligible"] is False
    assert "PORTRAIT_50_PLUS_TOO_LOW" in one_side_low["failed_reason_codes"]
    assert "CONTENT_PORTRAIT_MISSING" not in one_side_low["failed_reason_codes"]

    both_low = evaluate_candidate_gate(
        _p0_candidate(
            content_50_plus_ratio=0.10,
            account_50_plus_ratio=0.08,
        ),
        _P0_RULES,
    )
    assert both_low["primary_eligible"] is False
    assert "PORTRAIT_50_PLUS_TOO_LOW" in both_low["failed_reason_codes"]


def test_p0_gate_compensates_missing_fields_with_strong_signals() -> None:
    missing_publish = evaluate_candidate_gate(
        _p0_candidate(publish_at=None, share_count=2000),
        _P0_RULES,
    )
    assert missing_publish["primary_eligible"] is True

    missing_duration = evaluate_candidate_gate(
        _p0_candidate(duration_seconds=None, share_count=2000),
        _P0_RULES,
    )
    assert missing_duration["primary_eligible"] is True

    missing_shares = evaluate_candidate_gate(
        _p0_candidate(share_count=None, like_count=8000),
        _P0_RULES,
    )
    assert missing_shares["primary_eligible"] is True

    strong_scores = evaluate_candidate_gate(
        _p0_candidate(
            content_50_plus_ratio=None,
            account_50_plus_ratio=0.22,
            relevance_score=0.82,
            elder_score=0.78,
            share_score=0.72,
            value_score=0.70,
        ),
        _P0_RULES,
    )
    assert strong_scores["primary_eligible"] is True


def test_search_candidate_persists_publish_time_duration_and_shares() -> None:
    candidate = _candidate_from_search_result(
        {
            "aweme_id": "video-p0",
            "desc": "测试视频",
            "duration_ms": 65000,
            "publish_at": "2026-07-31T08:30:00+08:00",
            "statistics": {"share_count": 45},
        },
        "测试关键词",
    )
    assert candidate is not None
    assert candidate["duration_seconds"] == Decimal("65.000")
    assert candidate["publish_at"] == datetime(2026, 7, 31, 8, 30)
    assert candidate["share_count"] == 45


def test_all_search_sources_default_to_thirty_seconds() -> None:
    assert INTERNAL_SEARCH_MIN_DURATION == 30
    assert TIKHUB_SEARCH_MIN_DURATION == 30
    assert AUTHOR_SEARCH_MIN_DURATION == 30


def test_repository_reclassifies_primary_that_fails_p0_gate() -> None:
    factory = _expire_on_commit_session_factory()
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="p0-gate-run",
                demand_word="生活技巧",
                relevant_points_json="[]",
                status="running",
                rule_version="test-p0",
                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
            )
        )
        session.add(
            VideoDiscoveryCandidate(
                id=2,
                run_id="p0-gate-run",
                aweme_id="video-p0",
                title="适合家庭分享的生活技巧",
                publish_at=datetime(2026, 7, 31, 9, 0),
                duration_seconds=Decimal("30.000"),
                share_count=999,
                content_50_plus_ratio=Decimal("0.280000"),
                account_50_plus_ratio=Decimal("0.080000"),
                decision_bucket="pending_evaluation",
            )
        )
        session.commit()

    with factory() as session:
        repo = VideoDiscoveryRepository(session)
        batch = repo.update_candidates(
            "p0-gate-run",
            [{"candidate_id": 2, "decision_bucket": "primary"}],
        )
        session.commit()

        assert batch.candidates[0].decision_bucket == "rejected"
        assert batch.candidates[0].reject_reason_code == "SHARE_COUNT_TOO_LOW"
        assert batch.reclassified == [
            {
                "candidate_id": 2,
                "requested_bucket": "primary",
                "saved_bucket": "rejected",
                "failed_reason_codes": ["SHARE_COUNT_TOO_LOW"],
            }
        ]


def test_repository_gate_uses_persisted_like_and_play_compensation() -> None:
    factory = _expire_on_commit_session_factory()
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="complete-gate-input",
                demand_word="生活技巧",
                relevant_points_json="[]",
                status="running",
                rule_version="test-p0",
                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
            )
        )
        session.add(
            VideoDiscoveryCandidate(
                id=2,
                run_id="complete-gate-input",
                aweme_id="video-like-compensation",
                title="适合家庭分享的生活技巧",
                publish_at=datetime(2026, 7, 31, 9, 0),
                duration_seconds=Decimal("30.000"),
                share_count=None,
                like_count=8000,
                content_50_plus_ratio=Decimal("0.280000"),
                detail_fetch_status="success",
                content_portrait_fetch_status="success",
                decision_bucket="pending_evaluation",
            )
        )
        session.commit()

    with factory() as session:
        batch = VideoDiscoveryRepository(session).update_candidates(
            "complete-gate-input",
            [{"candidate_id": 2, "decision_bucket": "primary"}],
        )
        session.commit()
        assert batch.candidates[0].decision_bucket == "primary"
        assert batch.candidates[0].gate_status == "pass"
        evaluation = session.get(
            VideoDiscoveryGateEvaluation,
            batch.candidates[0].latest_gate_evaluation_id,
        )
        assert evaluation is not None
        assert json.loads(evaluation.gate_input_json)["like_count"] == 8000


def test_publishable_candidates_require_current_gate_version() -> None:
    factory = _expire_on_commit_session_factory()
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="stale-gate-run",
                demand_word="生活技巧",
                relevant_points_json="[]",
                status="finished",
            )
        )
        session.add_all(
            [
                VideoDiscoveryCandidate(
                    id=2,
                    run_id="stale-gate-run",
                    aweme_id="current",
                    decision_bucket="primary",
                    gate_status="pass",
                    evidence_version=2,
                    gate_evidence_version=2,
                ),
                VideoDiscoveryCandidate(
                    id=3,
                    run_id="stale-gate-run",
                    aweme_id="stale",
                    decision_bucket="primary",
                    gate_status="pass",
                    evidence_version=2,
                    gate_evidence_version=1,
                ),
            ]
        )
        session.commit()

    with factory() as session:
        rows = VideoDiscoveryRepository(session).list_publishable_candidates(
            run_id="stale-gate-run"
        )
        assert [item.aweme_id for item in rows] == ["current"]


def test_candidate_evidence_is_applied_and_gate_is_recomputed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from supply_infra.services.video_discovery_service import (
        build_evidence_values,
        get_video_discovery_service,
    )

    factory = _expire_on_commit_session_factory()
    _patch_service_session(monkeypatch, factory)
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="evidence-apply-run",
                demand_word="生活技巧",
                relevant_points_json="[]",
                status="running",
                rule_version="test-p0",
                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
            )
        )
        session.add(
            VideoDiscoveryCandidate(
                id=2,
                run_id="evidence-apply-run",
                aweme_id="evidence-video",
                title="适合家庭分享的生活技巧",
                decision_bucket="pending_evaluation",
            )
        )
        session.commit()

    detail = {
        "publish_at": datetime(2026, 7, 31, 9, 0),
        "duration_seconds": Decimal("30.000"),
        "share_count": 1000,
        "like_count": 8000,
    }
    detail_evidence = build_evidence_values(
        run_id="evidence-apply-run",
        evidence_type="detail",
        provider="crawler",
        subject_key="evidence-video",
        raw_response={"data": {"share_count": 1000}},
        normalized=detail,
        fetch_status="success",
        aweme_id="evidence-video",
    )
    first = get_video_discovery_service().record_candidate_evidence(
        run_id="evidence-apply-run",
        evidence_values=detail_evidence,
        aweme_id="evidence-video",
        normalized=detail,
    )
    assert first["candidates"][0]["gate_status"] == "pending"

    portrait = {
        "content_50_plus_ratio": Decimal("0.280000"),
        "content_50_plus_tgi": Decimal("130.0000"),
    }
    portrait_evidence = build_evidence_values(
        run_id="evidence-apply-run",
        evidence_type="content_portrait",
        provider="douhot",
        subject_key="evidence-video",
        raw_response={"data": {"年龄": {"50-": {"percentage": 0.28}}}},
        normalized=portrait,
        fetch_status="success",
        aweme_id="evidence-video",
    )
    second = get_video_discovery_service().record_candidate_evidence(
        run_id="evidence-apply-run",
        evidence_values=portrait_evidence,
        aweme_id="evidence-video",
        normalized=portrait,
    )
    candidate = second["candidates"][0]
    assert candidate["gate_status"] == "pass"
    assert candidate["evidence_version"] == 2
    assert candidate["gate_evidence_version"] == 2

    with factory() as session:
        assert len(session.scalars(select(VideoDiscoveryEvidence)).all()) == 2
        assert len(session.scalars(select(VideoDiscoveryGateEvaluation)).all()) == 2


def test_repository_keeps_valid_items_when_same_batch_contains_gate_failure() -> None:
    factory = _expire_on_commit_session_factory()
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="mixed-gate-run",
                demand_word="生活技巧",
                relevant_points_json="[]",
                status="running",
                rule_version="test-p0",
                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
            )
        )
        session.add_all(
            [
                VideoDiscoveryCandidate(
                    id=2,
                    run_id="mixed-gate-run",
                    aweme_id="valid-video",
                    decision_bucket="pending_evaluation",
                ),
                VideoDiscoveryCandidate(
                    id=3,
                    run_id="mixed-gate-run",
                    aweme_id="low-share-video",
                    decision_bucket="pending_evaluation",
                ),
            ]
        )
        session.commit()

    base = {
        "publish_at": datetime(2026, 7, 31, 9, 0),
        "duration_seconds": Decimal("30.000"),
        "content_50_plus_ratio": Decimal("0.280000"),
        "account_50_plus_ratio": Decimal("0.080000"),
        "decision_bucket": "primary",
    }
    with factory() as session:
        batch = VideoDiscoveryRepository(session).update_candidates(
            "mixed-gate-run",
            [
                {"candidate_id": 2, "share_count": 1000, **base},
                {"candidate_id": 3, "share_count": 999, **base},
            ],
        )
        session.commit()

        assert [item.decision_bucket for item in batch.candidates] == [
            "primary",
            "rejected",
        ]
        assert [item["candidate_id"] for item in batch.reclassified] == [3]


def test_finish_run_computes_partial_from_distinct_valid_primary_and_archives_pending() -> None:
    factory = _expire_on_commit_session_factory()
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="finish-partial",
                demand_word="生活技巧",
                relevant_points_json="[]",
                status="running",
            )
        )
        session.add_all(
            [
                VideoDiscoveryCandidate(
                    id=2,
                    run_id="finish-partial",
                    aweme_id="same-video",
                    decision_bucket="primary",
                    gate_status="pass",
                    evidence_version=1,
                    gate_evidence_version=1,
                ),
                VideoDiscoveryCandidate(
                    id=3,
                    run_id="finish-partial",
                    aweme_id="same-video",
                    decision_bucket="primary",
                    gate_status="pass",
                    evidence_version=1,
                    gate_evidence_version=1,
                ),
                VideoDiscoveryCandidate(
                    id=4,
                    run_id="finish-partial",
                    aweme_id="not-selected",
                    decision_bucket="pending_evaluation",
                ),
            ]
        )
        session.commit()

    with factory() as session:
        repo = VideoDiscoveryRepository(session)
        run = repo.finish_run("finish-partial", status="finished")
        session.commit()

        assert run.primary_count == 2
        assert run.valid_primary_count == 1
        assert run.outcome_status == "partial"
        pending = session.scalar(
            select(VideoDiscoveryCandidate).where(VideoDiscoveryCandidate.id == 4)
        )
        assert pending is not None
        assert pending.decision_bucket == "rejected"
        assert pending.reject_reason_code == "NOT_SELECTED_AFTER_EVALUATION"


def test_new_run_lifecycle_accepts_one_portrait_side_and_finishes_partial(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """贯穿工具、Service、P0、结束归档和运行结果判定的完整新流程。"""
    factory = _expire_on_commit_session_factory()
    _patch_service_session(monkeypatch, factory)
    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="lifecycle-one-side",
                demand_word="生活技巧",
                relevant_points_json="[]",
                status="running",
                attempt_count=1,
                rule_version="test-p0",
                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
            )
        )
        session.add_all(
            [
                VideoDiscoveryCandidate(
                    id=2,
                    run_id="lifecycle-one-side",
                    aweme_id="account-portrait-only",
                    title="适合家庭分享的生活技巧",
                    decision_bucket="pending_evaluation",
                ),
                VideoDiscoveryCandidate(
                    id=3,
                    run_id="lifecycle-one-side",
                    aweme_id="not-evaluated",
                    decision_bucket="pending_evaluation",
                ),
            ]
        )
        session.commit()

    updated = json.loads(
        video_discovery_store.batch_update_video_discovery_candidates(
            "lifecycle-one-side",
            [
                {
                    "candidate_id": 2,
                    "publish_at": "2026-07-31T09:00:00+08:00",
                    "duration_seconds": 30,
                    "share_count": 1000,
                    "content_50_plus_ratio": None,
                    "account_50_plus_ratio": None,
                    "age_normalization": {
                        "content": {
                            "has_age_portrait": False,
                            "older_ratio": 0,
                            "strength": "missing",
                        },
                        "account": {
                            "has_age_portrait": True,
                            "older_ratio": 0.55,
                            "strength": "strong",
                        },
                    },
                    "decision_bucket": "primary",
                }
            ],
        )
    )
    assert "error" not in updated
    assert updated["reclassified_count"] == 0
    assert updated["candidates"][0]["decision_bucket"] == "primary"
    assert updated["candidates"][0]["gate_status"] == "pass"
    with factory() as session:
        saved = session.get(VideoDiscoveryCandidate, 2)
        assert saved is not None
        assert saved.content_50_plus_ratio is None
        assert saved.account_50_plus_ratio == Decimal("0.550000")

    finished = json.loads(
        video_discovery_store.update_video_discovery_run_status(
            "lifecycle-one-side",
            "finished",
            stop_reason="合理搜索前沿已耗尽",
        )
    )
    assert "error" not in finished
    assert finished["run"]["valid_primary_count"] == 1
    assert finished["run"]["outcome_status"] == "partial"

    from agents.find_agent.run_outcome import evaluate_find_agent_run

    outcome = evaluate_find_agent_run("lifecycle-one-side")
    assert outcome.succeeded is True
    assert outcome.business_outcome == "partial"
    assert outcome.goal_met is False
    assert outcome.valid_primary_count == 1

    with factory() as session:
        pending = session.scalar(
            select(VideoDiscoveryCandidate).where(VideoDiscoveryCandidate.id == 3)
        )
        assert pending is not None
        assert pending.decision_bucket == "rejected"
        assert pending.reject_reason_code == "NOT_SELECTED_AFTER_EVALUATION"


@pytest.mark.asyncio
async def test_douyin_search_automatically_persists_page(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    search_module = importlib.import_module(
        "agents.find_agent.tools.douyin_search"
    )

    async def fake_raw_search(**_kwargs):
        return json.dumps(
            {
                "results_count": 1,
                "has_more": False,
                "search_results": [{"aweme_id": "auto-saved"}],
            }
        )

    persisted: dict[str, object] = {}

    def fake_persist(payload_json: str, **kwargs):
        persisted.update(kwargs)
        payload = json.loads(payload_json)
        payload.update(
            {
                "persisted": True,
                "search_id": 11,
                "new_candidate_count": 1,
                "candidates": [
                    {
                        "candidate_id": 21,
                        "search_id": 11,
                        "aweme_id": "auto-saved",
                        "title": "自动保存",
                        "decision_bucket": "pending_evaluation",
                    }
                ],
            }
        )
        return json.dumps(payload)

    monkeypatch.setattr(search_module, "_douyin_search_raw", fake_raw_search)
    monkeypatch.setattr(search_module, "persist_search_payload", fake_persist)

    result = json.loads(
        await search_module.douyin_search(
            run_id="run-auto-save",
            keyword="广场舞",
            query_reason="验证需求根搜索",
            source_type="demand",
        )
    )

    assert result["persisted"] is True
    assert result["search_id"] == 11
    assert result["candidates"][0]["candidate_id"] == 21
    assert persisted["run_id"] == "run-auto-save"
    assert persisted["keyword"] == "广场舞"
    assert persisted["provider"] == "internal_keyword"


def test_search_persistence_keeps_complete_raw_response(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    persistence = importlib.import_module(
        "agents.find_agent.support.search_persistence"
    )
    captured: dict[str, object] = {}

    class FakeService:
        def save_search_page(
            self, run_id, search_values, candidate_rows, evidence_values
        ):
            captured["run_id"] = run_id
            captured["evidence_values"] = evidence_values
            return {
                "search_id": 11,
                "new_candidate_count": 1,
                "candidates": [
                    {
                        "candidate_id": 21,
                        "search_id": 11,
                        "aweme_id": "raw-video",
                        "title": "原始响应测试",
                    }
                ],
            }

    monkeypatch.setattr(
        persistence,
        "get_video_discovery_service",
        lambda: FakeService(),
    )
    raw_payload = {
        "results_count": 1,
        "has_more": False,
        "provider_extension": {"must_keep": [1, 2, 3]},
        "_raw_response": {
            "provider_extension": {"must_keep": [1, 2, 3]},
            "unparsed_provider_field": "kept",
        },
        "search_results": [
            {
                "aweme_id": "raw-video",
                "desc": "原始响应测试",
                "statistics": {"share_count": 0},
            }
        ],
    }
    result = json.loads(
        persistence.persist_search_payload(
            json.dumps(raw_payload, ensure_ascii=False),
            run_id="raw-run",
            keyword="原始响应",
            query_reason="验证完整落库",
            source_type="demand",
            cursor="0",
            page_no=1,
            provider="internal_keyword",
        )
    )

    evidence_values = captured["evidence_values"]
    assert isinstance(evidence_values, dict)
    saved_raw = json.loads(str(evidence_values["raw_response_json"]))
    assert saved_raw["provider_extension"] == {"must_keep": [1, 2, 3]}
    assert saved_raw["unparsed_provider_field"] == "kept"
    assert "_raw_response" not in result
    assert result["candidates"][0]["candidate_id"] == 21


@pytest.mark.asyncio
async def test_batch_search_records_each_page_and_carries_parent(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    batch_module = importlib.import_module(
        "agents.find_agent.tools.batch_search_and_record"
    )
    calls: list[dict[str, object]] = []

    async def fake_search(**kwargs):
        calls.append(kwargs)
        page_no = int(kwargs["page_no"])
        return json.dumps(
            {
                "results_count": 1,
                "has_more": page_no == 1,
                "next_cursor": "next-page" if page_no == 1 else None,
                "persisted": True,
                "search_id": 100 + page_no,
                "new_candidate_count": 1,
                "candidates": [
                    {
                        "candidate_id": 200 + page_no,
                        "search_id": 100 + page_no,
                        "aweme_id": "same-video",
                        "title": f"第 {page_no} 页",
                        "decision_bucket": "pending_evaluation",
                    }
                ],
            }
        )

    monkeypatch.setattr(batch_module, "douyin_search", fake_search)

    result = json.loads(
        await batch_search_and_record(
            run_id="run-batch",
            searches=[
                {
                    "keyword": "广场舞",
                    "query_reason": "验证需求根搜索",
                    "source_type": "demand",
                    "max_pages": 2,
                }
            ],
        )
    )

    assert result["saved_page_count"] == 2
    assert result["new_candidate_count"] == 2
    assert result["tasks"][0]["pages"][0]["candidates"][0]["candidate_id"] == 201
    assert result["tasks"][0]["pages"][1]["candidates"][0]["candidate_id"] == 202
    assert calls[0]["parent_search_id"] is None
    assert calls[1]["parent_search_id"] == 101
    assert calls[1]["cursor"] == "next-page"


def test_batch_update_candidates_uses_database_candidate_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured: dict[str, object] = {}

    class FakeService:
        def update_candidates(self, run_id, rows):
            captured["run_id"] = run_id
            captured["rows"] = rows
            return {
                "updated_count": 1,
                "candidates": [
                    {
                        "candidate_id": 901,
                        "search_id": 801,
                        "aweme_id": "same-video",
                        "decision_bucket": "primary",
                    }
                ],
            }

    monkeypatch.setattr(
        video_discovery_store,
        "get_video_discovery_service",
        lambda: FakeService(),
    )

    result = json.loads(
        video_discovery_store.batch_update_video_discovery_candidates(
            run_id="run-update",
            items=[
                {
                    "candidate_id": 901,
                    "decision_bucket": "primary",
                    "relevance_score": 0.8,
                    "elder_score": 0.7,
                    "share_score": 0.6,
                    "age_normalization": {
                        "age_normalization_result": {
                            "content": {"older_ratio": 0.62},
                            "account": {"older_ratio": 0.60},
                        }
                    },
                }
            ],
        )
    )

    assert result["updated_count"] == 1
    assert captured["run_id"] == "run-update"
    assert captured["rows"][0]["candidate_id"] == 901
    assert captured["rows"][0]["content_50_plus_ratio"] == Decimal("0.620000")
    assert captured["rows"][0]["account_50_plus_ratio"] == Decimal("0.600000")
    assert "aweme_id" not in captured["rows"][0]


@pytest.mark.parametrize(
    ("age_normalization", "expected_content", "expected_account"),
    [
        (
            {
                "content": {
                    "has_age_portrait": False,
                    "older_ratio": 0.0,
                    "strength": "missing",
                },
                "account": {
                    "has_age_portrait": True,
                    "older_ratio": 0.55,
                    "strength": "strong",
                },
            },
            None,
            Decimal("0.550000"),
        ),
        (
            {
                "content": {
                    "has_age_portrait": True,
                    "older_ratio": 0.42,
                    "strength": "strong",
                },
                "account": {
                    "has_age_portrait": False,
                    "older_ratio": 0.0,
                    "strength": "missing",
                },
            },
            Decimal("0.420000"),
            None,
        ),
    ],
)
def test_candidate_update_keeps_unavailable_portrait_side_missing(
    age_normalization,
    expected_content,
    expected_account,
) -> None:
    from agents.find_agent.support.video_discovery import (
        _normalize_candidate_update,
    )

    row = _normalize_candidate_update(
        {
            "candidate_id": 1,
            "decision_bucket": "primary",
            "age_normalization": age_normalization,
        }
    )

    assert row["content_50_plus_ratio"] == expected_content
    assert row["account_50_plus_ratio"] == expected_account


def test_each_search_inserts_new_candidate_occurrences() -> None:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    VideoDiscoveryRun.__table__.create(engine)
    VideoDiscoverySearch.__table__.create(engine)
    VideoDiscoveryCandidate.__table__.create(engine)
    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    ids = {"search": 100, "candidate": 1000}

    @event.listens_for(factory.class_, "before_flush")
    def assign_sqlite_bigint_ids(session, _flush_context, _instances):
        for entity in session.new:
            if isinstance(entity, VideoDiscoverySearch) and entity.id is None:
                ids["search"] += 1
                entity.id = ids["search"]
            elif isinstance(entity, VideoDiscoveryCandidate) and entity.id is None:
                ids["candidate"] += 1
                entity.id = ids["candidate"]

    with factory() as session:
        session.add(
            VideoDiscoveryRun(
                id=1,
                run_id="run-occurrences",
                demand_word="广场舞",
                relevant_points_json="[]",
                status="running",
            )
        )
        session.commit()

    search_values = {
        "run_id": "run-occurrences",
        "search_key": "same-search-key",
        "keyword": "广场舞",
        "query_reason": "验证相同搜索也生成新记录",
        "source_type": "demand",
        "provider": "internal_keyword",
        "content_type": "视频",
        "sort_type": "综合排序",
        "publish_time": "不限",
        "cursor": "0",
        "page_no": 1,
        "results_count": 1,
        "new_candidate_count": 0,
        "has_more": 0,
        "status": "success",
    }
    candidate_rows = [
        {
            "aweme_id": "same-video",
            "title": "同一个视频",
            "_source_keyword": "广场舞",
        }
    ]

    with factory() as session:
        repo = VideoDiscoveryRepository(session)
        first_search, first_candidates = repo.save_search_page(
            dict(search_values),
            candidate_rows,
        )
        second_search, second_candidates = repo.save_search_page(
            dict(search_values),
            candidate_rows,
        )
        session.commit()

        assert first_search.id != second_search.id
        assert first_candidates[0].id != second_candidates[0].id
        assert first_candidates[0].search_id == first_search.id
        assert second_candidates[0].search_id == second_search.id

    with factory() as session:
        searches = session.scalars(select(VideoDiscoverySearch)).all()
        candidates = session.scalars(select(VideoDiscoveryCandidate)).all()
        assert len(searches) == 2
        assert len(candidates) == 2
        assert {candidate.aweme_id for candidate in candidates} == {"same-video"}


def _seed_candidate(
    factory: sessionmaker[Session],
    *,
    run_id: str,
    aweme_id: str = "7631830155522179258",
    row_id: int = 1,
) -> None:
    with factory() as session:
        session.add(
            VideoDiscoveryCandidate(
                id=row_id,
                run_id=run_id,
                aweme_id=aweme_id,
                decision_bucket="pending_evaluation",
            )
        )
        session.commit()


def test_list_skip_grade_ids_uses_finished_or_attempt_limit(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    factory = _expire_on_commit_session_factory()
    _patch_service_session(monkeypatch, factory)
    _seed_run(factory, run_id="finished", demand_grade_id=301, status="finished")
    _seed_run(
        factory,
        run_id="attempted",
        demand_grade_id=302,
        status="running",
        row_id=2,
    )
    with factory() as session:
        attempted = session.scalar(
            select(VideoDiscoveryRun).where(VideoDiscoveryRun.run_id == "attempted")
        )
        assert attempted is not None
        attempted.attempt_count = 1
        session.commit()

    from supply_infra.services.video_discovery_service import get_video_discovery_service

    skip_ids = get_video_discovery_service().list_skip_grade_ids("20260728")
    assert skip_ids == {301, 302}


def test_evaluate_find_agent_run_accepts_partial_as_completed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from agents.find_agent.run_outcome import evaluate_find_agent_run
    from supply_agent.types import AgentResult

    monkeypatch.setattr(
        "agents.find_agent.run_outcome.get_video_discovery_service",
        lambda: type(
            "Svc",
            (),
            {
                "lookup_run": staticmethod(
                    lambda _run_id: {
                        "status": "finished",
                        "outcome_status": "partial",
                        "valid_primary_count": 3,
                    }
                )
            },
        )(),
    )

    outcome = evaluate_find_agent_run(
        "run-1",
        AgentResult(content="任意文案", messages=[], iterations=3, tool_calls_made=0),
    )
    assert outcome.succeeded is True
    assert outcome.business_outcome == "partial"
    assert outcome.goal_met is False
    assert outcome.valid_primary_count == 3
    assert outcome.failure_reason is None


def test_evaluate_find_agent_run_fails_for_technical_failure(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from agents.find_agent.run_outcome import evaluate_find_agent_run
    from supply_agent.types import AgentResult

    monkeypatch.setattr(
        "agents.find_agent.run_outcome.get_video_discovery_service",
        lambda: type(
            "Svc",
            (),
            {
                "lookup_run": staticmethod(
                    lambda _run_id: {
                        "status": "failed",
                        "outcome_status": "failed",
                        "valid_primary_count": 0,
                    }
                )
            },
        )(),
    )

    outcome = evaluate_find_agent_run(
        "run-1",
        AgentResult(
            content="任务未完成（工具故障）",
            messages=[],
            iterations=1,
            tool_calls_made=0,
        ),
    )
    assert outcome.succeeded is False
    assert outcome.business_outcome == "failed"
    assert outcome.failure_reason == "run_failed"


@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.process_single_discover"
)
@patch("supply_infra.scheduler.jobs.discover_videos_from_demands._count_passed_videos")
@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.filter_pending_contexts"
)
@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.list_find_demand_contexts"
)
def test_stops_discovery_after_daily_passed_video_limit(
    mock_list_contexts,
    mock_filter_contexts,
    mock_count_passed,
    mock_process,
) -> None:
    from supply_infra.scheduler.jobs.discover_videos_from_demands import (
        _DAILY_PASSED_VIDEO_LIMIT,
    )

    contexts = [
        FindDemandContext(
            biz_dt="20260727",
            demand_grade_id=1,
            demand_name="需求A",
            grade="S",
        ),
        FindDemandContext(
            biz_dt="20260727",
            demand_grade_id=2,
            demand_name="需求B",
            grade="A",
        ),
    ]
    mock_list_contexts.return_value = ("20260727", contexts)
    mock_filter_contexts.return_value = (
        contexts,
        {"total_loaded": 2, "skipped_already_done": 0},
    )
    almost_full = _DAILY_PASSED_VIDEO_LIMIT - 1
    full = _DAILY_PASSED_VIDEO_LIMIT
    # 初始检查 / 执行前检查 / 执行后检查 / 收尾检查
    mock_count_passed.side_effect = [almost_full, almost_full, full, full]
    mock_process.return_value = {
        "success": True,
        "skipped": False,
        "demand_grade_id": 1,
        "demand_name": "需求A",
        "business_outcome": "goal_met",
    }

    result = discover_videos_from_demands("20260727", workers=1)

    assert mock_process.call_count == 1
    assert result["processed"] == 1
    assert result["passed_videos"] == full
    assert result["stopped_by_passed_video_limit"] is True
    assert result["passed_video_limit"] == _DAILY_PASSED_VIDEO_LIMIT


@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.process_single_discover"
)
@patch("supply_infra.scheduler.jobs.discover_videos_from_demands._count_passed_videos")
@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.filter_pending_contexts"
)
@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.list_find_demand_contexts"
)
def test_queue_workers_process_all_pending_tasks(
    mock_list_contexts,
    mock_filter_contexts,
    mock_count_passed,
    mock_process,
) -> None:
    contexts = [
        FindDemandContext(
            biz_dt="20260727",
            demand_grade_id=i,
            demand_name=f"需求{i}",
            grade="S",
        )
        for i in range(1, 8)
    ]
    mock_list_contexts.return_value = ("20260727", contexts)
    mock_filter_contexts.return_value = (
        contexts,
        {"total_loaded": 7, "skipped_already_done": 0},
    )
    mock_count_passed.return_value = 0

    def _fake_process(ctx, *, force=False):
        return {
            "success": True,
            "skipped": False,
            "demand_grade_id": ctx.demand_grade_id,
            "demand_name": ctx.demand_name,
            "business_outcome": "partial",
        }

    mock_process.side_effect = _fake_process

    result = discover_videos_from_demands("20260727", workers=5)

    assert mock_process.call_count == 7
    assert result["processed"] == 7
    assert result["succeeded"] == 7
    assert result["partial"] == 7
    assert result["workers"] == 5
    assert result["stopped_by_passed_video_limit"] is False


@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands._run_task_with_timeout"
)
@patch("supply_infra.scheduler.jobs.discover_videos_from_demands._count_passed_videos")
@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.filter_pending_contexts"
)
@patch(
    "supply_infra.scheduler.jobs.discover_videos_from_demands.list_find_demand_contexts"
)
def test_worker_timeout_marks_failed_and_continues_queue(
    mock_list_contexts,
    mock_filter_contexts,
    mock_count_passed,
    mock_run_timeout,
) -> None:
    from concurrent.futures import TimeoutError as FuturesTimeoutError

    contexts = [
        FindDemandContext(
            biz_dt="20260727",
            demand_grade_id=1,
            demand_name="超时任务",
            grade="S",
        ),
        FindDemandContext(
            biz_dt="20260727",
            demand_grade_id=2,
            demand_name="后续任务",
            grade="A",
        ),
    ]
    mock_list_contexts.return_value = ("20260727", contexts)
    mock_filter_contexts.return_value = (
        contexts,
        {"total_loaded": 2, "skipped_already_done": 0},
    )
    mock_count_passed.return_value = 0

    def _side_effect(ctx, *, force=False, timeout_seconds=600.0):
        if ctx.demand_grade_id == 1:
            raise FuturesTimeoutError()
        return {
            "success": True,
            "skipped": False,
            "demand_grade_id": ctx.demand_grade_id,
            "demand_name": ctx.demand_name,
            "business_outcome": "goal_met",
        }

    mock_run_timeout.side_effect = _side_effect

    result = discover_videos_from_demands("20260727", workers=1)

    assert mock_run_timeout.call_count == 2
    assert result["processed"] == 2
    assert result["failed"] == 1
    assert result["succeeded"] == 1
    assert result["goal_met"] == 1
    assert any("timed out" in str(err.get("error", "")) for err in result["errors"])


class _AsyncClient:
    def __init__(self) -> None:
        self.closed = False

    async def close(self) -> None:
        self.closed = True


class _SlowAgent:
    def __init__(self) -> None:
        self.llm = type("LLM", (), {"_async_client": _AsyncClient()})()

    async def arun_core(self, _user_input: str) -> None:
        await asyncio.sleep(60)


@pytest.mark.asyncio
async def test_find_agent_timeout_closes_async_client() -> None:
    agent = _SlowAgent()

    with pytest.raises(TimeoutError, match="find_agent timed out"):
        await arun_find_agent(
            agent,  # type: ignore[arg-type]
            "test",
            run_id="test-run",
            timeout_seconds=0.01,
        )

    assert agent.llm._async_client.closed is True


@pytest.mark.asyncio
async def test_detail_candidate_mode_persists_raw_and_returns_gate_summary(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    detail_module = importlib.import_module(
        "agents.find_agent.support.douyin_detail"
    )
    recorded: list[dict[str, object]] = []

    class FakeService:
        def get_candidates_by_ids(self, _run_id, _candidate_ids):
            return [
                {
                    "candidate_id": 21,
                    "aweme_id": "7641118685977614586",
                    "gate_status": "pending" if not recorded else "pass",
                    "gate_results": {"failed_reason_codes": []},
                    "evidence_version": len(recorded),
                }
            ]

        def record_candidate_evidence(self, **kwargs):
            recorded.append(kwargs)
            return {"evidence_id": 1, "updated_count": 1, "candidates": []}

    async def fake_fetch(_client, content_id):
        return (
            {
                "content_id": content_id,
                "title": "详情自动落库",
                "publish_at": "2026-07-31T09:00:00",
                "duration_seconds": 60,
                "share_count": 2000,
            },
            {"code": 0, "data": {"unparsed": "raw-detail"}},
        )

    service = FakeService()
    monkeypatch.setattr(detail_module, "get_video_discovery_service", lambda: service)
    monkeypatch.setattr(detail_module, "_fetch_one_detail", fake_fetch)

    result = json.loads(
        await detail_module.douyin_detail(
            run_id="detail-run",
            candidate_ids=[21],
        )
    )

    assert len(recorded) == 1
    evidence = recorded[0]["evidence_values"]
    assert isinstance(evidence, dict)
    assert "raw-detail" in str(evidence["raw_response_json"])
    assert "details" not in result
    assert "detail" not in result
    assert result["candidates"][0]["candidate_id"] == 21


@pytest.mark.asyncio
async def test_portrait_candidate_mode_persists_both_sides_and_returns_summary(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    portrait_module = importlib.import_module(
        "agents.find_agent.support.portrait"
    )
    recorded: list[dict[str, object]] = []

    class FakeService:
        def get_candidates_by_ids(self, _run_id, _candidate_ids):
            return [
                {
                    "candidate_id": 31,
                    "aweme_id": "7641118685977614586",
                    "author_sec_uid": "MS4wLjABAAAA-test",
                    "gate_status": "pass",
                    "gate_results": {"failed_reason_codes": []},
                    "evidence_version": len(recorded),
                }
            ]

        def record_candidate_evidence(self, **kwargs):
            recorded.append(kwargs)
            return {"evidence_id": len(recorded), "updated_count": 1, "candidates": []}

    async def fake_content(_client, _content_id, _flags):
        portrait = {"年龄": {"50-": {"percentage": 0.3, "preference": 130}}}
        return None, {
            "has_portrait": True,
            "portrait_data": portrait,
            "output": "content",
            "raw_data": {"raw": "content-portrait"},
        }

    async def fake_account(_client, _account_id, _flags):
        portrait = {"年龄": {"50-": {"percentage": 0.4, "preference": 140}}}
        return None, {
            "has_portrait": True,
            "portrait_data": portrait,
            "output": "account",
            "raw_data": {"raw": "account-portrait"},
        }

    service = FakeService()
    monkeypatch.setattr(portrait_module, "get_video_discovery_service", lambda: service)
    monkeypatch.setattr(portrait_module, "_fetch_content_portrait", fake_content)
    monkeypatch.setattr(portrait_module, "_fetch_account_portrait", fake_account)

    result = json.loads(
        await portrait_module.batch_fetch_portraits(
            run_id="portrait-run",
            candidate_ids=[31],
            fetch_account_portrait=True,
        )
    )

    assert [
        item["evidence_values"]["evidence_type"] for item in recorded
    ] == ["content_portrait", "account_portrait"]
    assert "results" not in result
    assert result["candidates"][0]["candidate_id"] == 31
