"""JSON 载荷解析、证据格式化、Story 批量规划校验(纯函数)。""" from app.agents.chat.reply_limits import truncate_chat_segments from app.agents.memoir.classification_agent import _normalize_llm_category from app.agents.memoir.story_route_agent import ( StoryBatchPlan, StoryBatchPlanUnit, validate_story_batch_plan, ) from app.core.json_utils import extract_json_payload from app.features.memory.evidence_format import format_evidence_chunks_for_prompt def test_extract_json_payload_strips_markdown_fence() -> None: raw = """```json {"a": 1} ```""" assert '"a"' in extract_json_payload(raw) def test_extract_json_payload_balanced_nested_braces() -> None: raw = 'noise {"outer": {"inner": 1}} trailing' assert extract_json_payload(raw) == '{"outer": {"inner": 1}}' def test_normalize_llm_category_strips_quotes() -> None: assert _normalize_llm_category('"childhood"') == "childhood" assert _normalize_llm_category("`beliefs`") == "beliefs" def test_format_evidence_chunks_uses_memory_formatter_without_timeline() -> None: ev = { "relevant_chunks": [{"content": "chunk1"}], "relevant_facts": [ {"subject": "我", "predicate": "生于", "object_json": "1950"} ], "relevant_summaries": [], "relevant_stories": [], } out = format_evidence_chunks_for_prompt(ev) assert "chunk1" in out assert "1950" in out or "生于" in out assert "恢复高考" not in out def test_validate_story_batch_plan_ok() -> None: ordered = ["s1", "s2"] plan = StoryBatchPlan( units=[ StoryBatchPlanUnit( segment_ids=["s1", "s2"], decision="new_story", target_story_id=None, new_story_title="标题", reason=None, ) ] ) ok, err = validate_story_batch_plan(ordered, plan, valid_story_ids=set()) assert ok is True assert err is None def test_truncate_chat_segments() -> None: out = truncate_chat_segments( ["a" * 300, "b"], max_segments=2, max_chars_per_segment=220, ) assert out[0] == "a" * 219 + "…" assert len(out[0]) == 220 assert out[1] == "b" def test_validate_story_batch_plan_duplicate_segment() -> None: plan = StoryBatchPlan( units=[ StoryBatchPlanUnit( segment_ids=["s1"], decision="new_story", target_story_id=None, new_story_title="A", reason=None, ), StoryBatchPlanUnit( segment_ids=["s1"], decision="new_story", target_story_id=None, new_story_title="B", reason=None, ), ] ) ok, err = validate_story_batch_plan(["s1", "s1"], plan, valid_story_ids=set()) assert ok is False assert err == "duplicate_segment"