feat: refine topic insights and learner experience

ai_knowledge_base_v2/apps/backend/tests/test_content_generation_configs.py

@@ -14,7 +14,9 @@ from app.models.ai_config import ContentGenerationConfig
 from app.services.content_generation_config_service import (
     SAMPLE_VALUES,
     ContentGenerationConfigService,
+    _parse_json_object,
 )
+from app.services.content_generation_variables import default_variables
 from app.services.tracked_generation_service import TrackedGenerationService
 
 
@@ -89,9 +91,12 @@ def test_ai_generation_uses_configured_instruction_and_only_accepts_allowed_fiel
         return SimpleNamespace(
             answer=json.dumps(
                 {
+                    "topic_title": "面对判断时的当下观察",
                     "issue": "整理后的问题",
                     "summary": "整理后的摘要",
                     "current_focus": "整理后的关注",
+                    "next_observation": "可以继续留意当下的感受。",
+                    "teacher_question": "请老师帮我确认暂停的时机。",
                     "unknown": "不能进入卡片",
                 },
                 ensure_ascii=False,
@@ -197,3 +202,173 @@ def test_custom_variable_versions_are_saved_and_restored_together():
         restored_variables_json = restored.variables_json
 
     assert json.loads(restored_variables_json)[0]["name"] == "custom_summary"
+
+
+def test_default_topic_title_is_extracted_by_ai_and_test_material_has_no_fake_title():
+    variables = default_variables("help_card")
+    topic_title = next(item for item in variables if item["name"] == "topic_title")
+
+    values = ContentGenerationConfigService.build_test_values(
+        "help_card",
+        variables,
+        "用户：我在面对领导时会紧张，想看看当下的身体感受。",
+    )
+
+    assert topic_title["valueSource"] == "ai"
+    assert topic_title["sourceKey"] is None
+    assert "topic_title" not in values
+    assert values["student_name"] == SAMPLE_VALUES["student_name"]
+    assert "面对领导时会紧张" in values["source_material"]
+
+
+def test_content_generation_parser_uses_final_json_after_reasoning_and_draft():
+    raw = (
+        '<think>{"topic_title":"思考草稿"}</think>\n'
+        '中间草稿：{"topic_title":"不完整标题"}\n'
+        '```json\n{"topic_title":"面对领导时的紧张觉察"}\n```'
+    )
+
+    assert _parse_json_object(raw) == {"topic_title": "面对领导时的紧张觉察"}
+
+
+def test_generation_prompt_keeps_late_conversation_material(monkeypatch):
+    variables = [
+        {
+            "name": "topic_title",
+            "label": "主题标题",
+            "description": "根据用户原话提炼主题",
+            "valueSource": "ai",
+            "sourceKey": None,
+            "sampleValue": "示例标题",
+        }
+    ]
+    captured: dict[str, str] = {}
+
+    def fake_generate(db, *, prompt, scenario, user_id):
+        captured["prompt"] = prompt
+        return SimpleNamespace(answer='{"topic_title":"最后的真实主题"}')
+
+    monkeypatch.setattr(TrackedGenerationService, "generate", fake_generate)
+    material = "用户：前置内容\n" + ("中间内容" * 1800) + "\n用户：最后我真正想讨论的是面对领导时的紧张。"
+
+    with _db() as db:
+        generated, used_fallback = ContentGenerationConfigService.generate_values(
+            db,
+            config_type="help_card",
+            instruction_content="忠实提炼",
+            variables=variables,
+            values={"source_material": material},
+            user_id=1,
+        )
+
+    assert used_fallback is False
+    assert generated["topic_title"] == "最后的真实主题"
+    assert "最后我真正想讨论的" in captured["prompt"]
+
+
+def test_preview_sample_never_leaks_into_formal_generation_fallback(monkeypatch):
+    variables = [
+        {
+            "name": "topic_title",
+            "label": "主题标题",
+            "description": "提炼主题",
+            "valueSource": "ai",
+            "sourceKey": None,
+            "sampleValue": "这只是排版预览示例",
+        }
+    ]
+
+    def fake_generate(db, *, prompt, scenario, user_id):
+        return SimpleNamespace(answer='{"other":"模型漏掉了主题标题"}')
+
+    monkeypatch.setattr(TrackedGenerationService, "generate", fake_generate)
+    with _db() as db:
+        generated, used_fallback = ContentGenerationConfigService.generate_values(
+            db,
+            config_type="help_card",
+            instruction_content="忠实提炼",
+            variables=variables,
+            values={"source_material": "用户：我最近面对领导时会紧张。"},
+            user_id=1,
+        )
+
+    assert used_fallback is True
+    assert generated["topic_title"] == "（请补充）"
+
+
+def test_missing_custom_fields_are_retried_and_list_values_are_renderable(monkeypatch):
+    variables = [
+        {
+            "name": "scene",
+            "label": "发生场景",
+            "description": "提炼具体场景",
+            "valueSource": "ai",
+            "sourceKey": None,
+            "sampleValue": "示例场景",
+        },
+        {
+            "name": "body_signals",
+            "label": "身体信号",
+            "description": "提炼用户明确提到的身体感受",
+            "valueSource": "ai",
+            "sourceKey": None,
+            "sampleValue": "示例感受",
+        },
+    ]
+    calls: list[str] = []
+
+    def fake_generate(db, *, prompt, scenario, user_id):
+        calls.append(prompt)
+        if len(calls) == 1:
+            return SimpleNamespace(answer='{"scene":"部门会议汇报"}')
+        return SimpleNamespace(answer='{"body_signals":["肩膀紧绷","呼吸很浅"]}')
+
+    monkeypatch.setattr(TrackedGenerationService, "generate", fake_generate)
+    with _db() as db:
+        generated, used_fallback = ContentGenerationConfigService.generate_values(
+            db,
+            config_type="help_card",
+            instruction_content="忠实提炼",
+            variables=variables,
+            values={"source_material": "用户：我在部门会议汇报时肩膀紧绷，呼吸很浅。"},
+            user_id=1,
+        )
+
+    assert used_fallback is False
+    assert generated == {"scene": "部门会议汇报", "body_signals": "肩膀紧绷；呼吸很浅"}
+    assert len(calls) == 2
+    assert "定向重试" in calls[1]
+    assert '"body_signals"' in calls[1]
+    assert '"scene"' not in calls[1]
+
+
+def test_all_thirty_custom_ai_variables_are_processed_without_hardcoded_field_names(monkeypatch):
+    variables = [
+        {
+            "name": f"custom_field_{index}",
+            "label": f"自定义字段 {index}",
+            "description": f"根据用户原话提炼第 {index} 个指定内容",
+            "valueSource": "ai",
+            "sourceKey": None,
+            "sampleValue": f"示例 {index}",
+        }
+        for index in range(1, 31)
+    ]
+    answer = {item["name"]: f"提炼结果 {index}" for index, item in enumerate(variables, start=1)}
+
+    def fake_generate(db, *, prompt, scenario, user_id):
+        return SimpleNamespace(answer=json.dumps(answer, ensure_ascii=False))
+
+    monkeypatch.setattr(TrackedGenerationService, "generate", fake_generate)
+    with _db() as db:
+        generated, used_fallback = ContentGenerationConfigService.generate_values(
+            db,
+            config_type="help_card",
+            instruction_content="忠实提炼",
+            variables=variables,
+            values={"source_material": "用户：这是用于验证动态变量的对话材料。"},
+            user_id=1,
+        )
+
+    assert used_fallback is False
+    assert generated == answer

ai_knowledge_base_v2/apps/backend/tests/test_entitlements_topics.py

@@ -36,6 +36,40 @@ def _seed_user_session(db: Session) -> tuple[User, ChatSession]:
     return user, session
 
 
+def test_monthly_topic_count_uses_shanghai_calendar_boundary():
+    with _db() as db:
+        user, session = _seed_user_session(db)
+        db.add_all(
+            [
+                TopicSession(
+                    user_id=user.id,
+                    chat_session_id=session.id,
+                    title="七月主题",
+                    core_question="七月",
+                    quota_deducted=1,
+                    started_at=datetime(2026, 7, 31, 15, 59, 59),
+                ),
+                TopicSession(
+                    user_id=user.id,
+                    chat_session_id=session.id,
+                    title="八月主题",
+                    core_question="八月",
+                    quota_deducted=1,
+                    started_at=datetime(2026, 7, 31, 16, 0, 0),
+                ),
+            ]
+        )
+        db.commit()
+
+        used = TopicSessionService.monthly_used_count(
+            db,
+            user.id,
+            at=datetime(2026, 8, 15, 12, 0, tzinfo=UTC),
+        )
+
+        assert used == 1
+
+
 def test_default_entitlement_uses_basic_plan_when_user_has_no_assignment():
     with _db() as db:
         user, _session = _seed_user_session(db)

ai_knowledge_base_v2/apps/backend/tests/test_help_cards.py

@@ -15,6 +15,7 @@ from app.models.entitlement import EntitlementPlan
 from app.models.growth import TeacherHelpCard
 from app.models.user import User
 from app.services.help_card_service import HelpCardService
+from app.services.help_card_service import _format_time
 
 
 def _db() -> Session:
@@ -87,3 +88,7 @@ def test_generate_help_card_from_topic_summary_and_mark_copied():
 
         HelpCardService.delete(db, user=user, card_id=card.id)
         assert db.get(TeacherHelpCard, card.id) is None
+
+
+def test_topic_time_is_rendered_in_china_local_timezone():
+    assert _format_time(datetime(2026, 8, 3, 3, 24, 7)) == "2026-08-03 11:24"

ai_knowledge_base_v2/apps/backend/tests/test_knowledge_agent.py

@@ -11,6 +11,7 @@ from sqlalchemy.orm import Session
 from sqlalchemy.pool import StaticPool
 
 from app.models import Base
+from app.models.ai_config import ModelConfig
 from app.models.knowledge import (
     Knowledge,
     KnowledgeChunk,
@@ -293,6 +294,34 @@ def test_long_follow_up_reference_is_detected_and_keeps_multiple_user_questions(
     assert question in rewritten
 
 
+def test_retrieval_model_is_deterministic_without_losing_runtime_limits():
+    model = ModelConfig(
+        provider="openai",
+        display_name="测试模型",
+        api_type="openai_compatible",
+        model_name="test-model",
+        api_url="https://example.com/v1/chat/completions",
+        api_key="secret",
+        temperature=0.8,
+        top_p=0.9,
+        top_k=40,
+        presence_penalty=0.2,
+        frequency_penalty=0.1,
+        max_token=4096,
+        stream_enabled=1,
+    )
+    model.id = 7
+
+    deterministic = KnowledgeAgentService._deterministic_retrieval_model(model)
+
+    assert deterministic.id == model.id
+    assert deterministic.temperature == 0
+    assert deterministic.top_p == model.top_p
+    assert deterministic.top_k == model.top_k
+    assert deterministic.max_token == model.max_token
+    assert deterministic.stream_enabled == 0
+
+
 def test_homework_overview_expands_practice_terms_and_section_limit():
     terms = KnowledgeAgentService._query_terms("合一的作业是什么？")
 

ai_knowledge_base_v2/apps/backend/tests/test_topic_auto_settlement.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.models import Base
+from app.models.ai_config import SystemConfig
+from app.models.chat import ChatMessage, ChatSession, TopicSession
+from app.models.growth import TopicSummary
+from app.models.user import User
+from app.core.auth_context import ChatAccessScope
+from app.services.chat_service import ChatService
+from app.services.topic_auto_settlement_service import TopicAutoSettlementService
+from app.services.topic_session_service import TopicSessionService
+
+
+def _db() -> Session:
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    return Session(engine)
+
+
+def _now() -> datetime:
+    return datetime.now(UTC).replace(tzinfo=None)
+
+
+def _seed(db: Session) -> tuple[User, ChatSession, TopicSession]:
+    user = User(id=1, phone="13800000001", name="学员", daily_chat_limit=100, daily_chat_used=0)
+    session = ChatSession(id=1, user_id=1, title="测试", message_count=0, last_message_at=_now(), is_deleted=0)
+    topic = TopicSession(
+        id=1,
+        user_id=1,
+        chat_session_id=1,
+        title="原始问题",
+        core_question="原始问题",
+        status="active",
+        message_count=0,
+        quota_deducted=1,
+        started_at=_now(),
+    )
+    db.add_all([user, session, topic])
+    db.commit()
+    return user, session, topic
+
+
+def _add_round(db: Session, topic: TopicSession, round_number: int, *, status: str = "FINISHED") -> None:
+    user_message = ChatMessage(
+        id=round_number * 2 - 1,
+        session_id=topic.chat_session_id,
+        topic_session_id=topic.id,
+        user_id=topic.user_id,
+        role="user",
+        content=f"问题{round_number}",
+        message_status="FINISHED",
+        created_at=_now(),
+    )
+    assistant_message = ChatMessage(
+        id=round_number * 2,
+        session_id=topic.chat_session_id,
+        topic_session_id=topic.id,
+        user_id=topic.user_id,
+        role="assistant",
+        content=f"回答{round_number}",
+        message_status=status,
+        created_at=_now(),
+    )
+    db.add_all([user_message, assistant_message])
+    TopicSessionService.attach_user_message(user_message, topic)
+    TopicSessionService.attach_assistant_message(assistant_message, topic, token_input=1, token_output=1)
+    db.flush()
+
+
+def test_default_second_successful_round_creates_snapshot_without_consuming_another_topic():
+    with _db() as db:
+        user, session, topic = _seed(db)
+
+        _add_round(db, topic, 1)
+        assert TopicAutoSettlementService.queue_if_due(db, user=user, topic=topic) is None
+        assert topic.status == "active"
+
+        _add_round(db, topic, 2)
+        summary = TopicAutoSettlementService.queue_if_due(db, user=user, topic=topic)
+        db.commit()
+
+        assert summary is not None
+        assert summary.status == "pending"
+        assert topic.status == "active"
+        assert topic.ended_at is None
+
+        same_topic = TopicSessionService.get_or_create_active(
+            db,
+            user=user,
+            session=session,
+            question="继续聊另一个问题",
+            deduct_quota=True,
+        )
+        assert same_topic.id == topic.id
+
+        new_session = ChatService.create_session(
+            db,
+            user,
+            ChatAccessScope.direct(),
+            current_session_id=session.id,
+        )
+        db.refresh(topic)
+        assert topic.status == "completed"
+        assert topic.ended_at is not None
+
+        next_topic = TopicSessionService.get_or_create_active(
+            db,
+            user=user,
+            session=new_session,
+            question="真正的新议题",
+            deduct_quota=True,
+        )
+        assert next_topic.id != topic.id
+        assert next_topic.quota_deducted == 1
+
+
+def test_configured_round_limit_only_counts_finished_assistant_messages():
+    with _db() as db:
+        user, _session, topic = _seed(db)
+        db.add(SystemConfig(config_key="topic_auto_settle_successful_rounds", config_value="3"))
+        db.flush()
+
+        _add_round(db, topic, 1)
+        _add_round(db, topic, 2, status="FAILED")
+        _add_round(db, topic, 3)
+        assert TopicAutoSettlementService.queue_if_due(db, user=user, topic=topic) is None
+
+        _add_round(db, topic, 4)
+        summary = TopicAutoSettlementService.queue_if_due(db, user=user, topic=topic)
+
+        assert summary is not None
+        assert topic.status == "active"
+        assert db.query(TopicSummary).filter_by(topic_session_id=topic.id).count() == 1