life-echo/api/tests/test_interview_turn_plan.py

"""interview_turn_plan：轮次模式与主槽选择（服务端硬编排）。"""

from app.agents.chat.interview_turn_plan import (
    InterviewTurnPlan,
    apply_safe_mode_override,
    extract_anchor_snippet,
    format_interview_turn_directive_block,
    plan_interview_turn,
    primary_empty_slot,
)


def test_primary_empty_slot_order():
    assert primary_empty_slot("childhood", ["emotion", "place"]) == "place"
    assert primary_empty_slot("childhood", ["emotion"]) == "emotion"


def test_extract_anchor_snippet_prefers_user_when_long_enough():
    mem = "摘录的一段记忆\n\n[场景氛围提示"
    um = "用户说很长一句" * 5
    sn = extract_anchor_snippet(memory_evidence_text=mem, user_message=um)
    assert sn.startswith("用户说")
    assert "摘录" not in sn


def test_extract_anchor_snippet_prefers_first_m_line():
    mem = (
        "【相关记忆摘录·聊天专用】\n"
        "说明行……\n"
        "[M1] 你在校园演出里饰演罗密欧。\n"
    )
    sn = extract_anchor_snippet(memory_evidence_text=mem, user_message="短")
    assert "校园演出" in sn
    assert "【相关" not in sn


def test_plan_sets_memory_usage_when_evidence_present():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place"],
        normalized_user_message="嗯。",
        memory_evidence_text="【头】\n[M1] 你提过河边。",
        stage_switched_this_turn=False,
    )
    assert p.memory_usage == "allowed_with_attribution"


def test_plan_marks_assistant_identity_question() -> None:
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place"],
        normalized_user_message="你是哪里人，你的童年是什么样的？",
        memory_evidence_text="用户曾说：「我小时候在上海长大。」",
        stage_switched_this_turn=False,
    )
    assert p.assistant_identity_question is True


def test_plan_reply_shape_ack_then_question_on_her_story_followup():
    p = plan_interview_turn(
        current_stage="education",
        empty_slots=["school"],
        normalized_user_message="那你讲讲她的故事吧。",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.reply_shape == "ack_then_question"


def test_directive_includes_attribution_when_memory_allowed():
    plan = InterviewTurnPlan(
        mode="memoir_push",
        anchor_slot_key="place",
        anchor_slot_readable="成长的地方",
        anchor_snippet="挂钩",
        memory_usage="allowed_with_attribution",
        memory_reference_style="你之前提过",
        reply_shape="ack_then_question",
    )
    block = format_interview_turn_directive_block(plan)
    assert "线索" in block
    assert "你之前提过" in block
    assert "真实人生传记" in block
    assert "ack" not in block.lower()
    assert "承接" in block
    assert (
        "本轮追问" in block
        or "承接角度" in block
        or "本轮承接重点" in block
    )


def test_directive_includes_focus_summary_when_set():
    plan = InterviewTurnPlan(
        mode="memoir_push",
        anchor_slot_key="place",
        anchor_slot_readable="成长的地方",
        anchor_snippet="挂钩",
        focus_summary="先接住「芳芳」与怕丢脸这条线",
        focus_source="llm",
        primary_focus="relationship",
    )
    block = format_interview_turn_directive_block(plan)
    assert "芳芳" in block
    assert "本轮追问" in block or "承接角度" in block


def test_directive_marks_user_message_anchor_source_correctly():
    plan = InterviewTurnPlan(
        mode="memoir_push",
        anchor_slot_key="place",
        anchor_slot_readable="成长的地方",
        anchor_snippet="那年夏天我总往河边跑",
        anchor_source_kind="user_message",
    )
    block = format_interview_turn_directive_block(plan)
    assert "来自用户本轮原话摘录" in block
    assert "不是用户本轮新说的内容" not in block


def test_directive_marks_memory_anchor_source_correctly():
    plan = InterviewTurnPlan(
        mode="memoir_push",
        anchor_slot_key="place",
        anchor_slot_readable="成长的地方",
        anchor_snippet="你小时候常在河边玩",
        anchor_source_kind="memory",
    )
    block = format_interview_turn_directive_block(plan)
    assert "来自检索到的用户过往口述/摘要" in block
    assert "不是用户本轮新说的内容" in block


def test_directive_adds_boundary_for_assistant_identity_question():
    plan = InterviewTurnPlan(
        mode="memoir_push",
        anchor_slot_key="place",
        anchor_slot_readable="成长的地方",
        anchor_snippet="用户曾说：「我小时候在上海长大。」",
        assistant_identity_question=True,
    )
    block = format_interview_turn_directive_block(plan)
    assert "本轮用户在问助手本人" in block
    assert "你刚提到上海" in block


def test_apply_safe_mode_override_blocks_emotion_downgrade():
    assert apply_safe_mode_override("emotion_first", "memoir_push", primary_focus="emotion") is None


def test_apply_safe_mode_override_allows_memoir_to_emotion():
    assert (
        apply_safe_mode_override("memoir_push", "emotion_first", primary_focus="relationship")
        == "emotion_first"
    )


def test_apply_safe_mode_override_clarify_blocks_memoir():
    assert apply_safe_mode_override("clarify_first", "memoir_push", primary_focus="emotion") is None


def test_apply_safe_mode_override_clarify_to_emotion():
    assert (
        apply_safe_mode_override("clarify_first", "emotion_first", primary_focus="emotion")
        == "emotion_first"
    )


def test_plan_clarify_first_when_ambiguous():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place", "people"],
        normalized_user_message="那种喜欢我也说不清，好像有又好像没有。",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "clarify_first"


def test_plan_clarify_first_when_very_short():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place"],
        normalized_user_message="有点说不清",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "clarify_first"


def test_plan_low_information_reply_pushes_next_topic_when_slots_remain():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place", "people"],
        normalized_user_message="嗯。",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "memoir_push"
    assert p.anchor_slot_key == "place"
    assert p.reply_shape == "ack_then_question"
    assert p.low_information_reply is True


def test_low_information_directive_asks_for_proactive_topic_not_clarification():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place"],
        normalized_user_message="对",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    block = format_interview_turn_directive_block(p)
    assert "低信息短回复处理" in block
    assert "不要把这个短答本身当成需要澄清的内容" in block
    assert "恰好一个" in block
    assert "主动" in block


def test_plan_low_information_reply_uses_follow_when_no_slots_remain():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=[],
        normalized_user_message="是的",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "follow_user_only"
    assert p.reply_shape == "ack_then_question"
    assert p.low_information_reply is True


def test_short_substantive_reply_is_not_treated_as_low_information():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place", "people"],
        normalized_user_message="上海",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.low_information_reply is False


def test_plan_memoir_push():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place", "people"],
        normalized_user_message="我小时候住在河边，夏天常去玩水。",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "memoir_push"
    assert p.anchor_slot_key == "place"
    assert p.anchor_snippet


def test_plan_emotion_first():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=["place"],
        normalized_user_message="想起来还是很难受，忍不住想哭。",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "emotion_first"


def test_plan_follow_on_stage_switch():
    p = plan_interview_turn(
        current_stage="education",
        empty_slots=["school", "city"],
        normalized_user_message="后来我去省城读中学了。",
        memory_evidence_text="",
        stage_switched_this_turn=True,
    )
    assert p.mode == "follow_user_only"


def test_plan_follow_when_no_empty_slots():
    p = plan_interview_turn(
        current_stage="childhood",
        empty_slots=[],
        normalized_user_message="嗯。",
        memory_evidence_text="",
        stage_switched_this_turn=False,
    )
    assert p.mode == "follow_user_only"
    assert p.low_information_reply is True


def test_build_topic_chips_english_uses_slot_name_map_en():
    from app.agents.chat.prompts_conversation import build_topic_chips

    chips = build_topic_chips("childhood", ["place"], max_chips=2, language="en")
    assert len(chips) >= 1
    place_chip = next(c for c in chips if c["id"] == "place")
    assert place_chip["label"] == "where you grew up"
    assert place_chip["text"].startswith("I'd like to talk about")


def test_build_topic_chips_belief_stage_not_empty():
    from app.agents.chat.prompts_conversation import build_topic_chips

    chips = build_topic_chips("belief", [], max_chips=4, language="en")
    assert len(chips) == 4
    assert chips[0]["label"]