""" Unit tests for memory-focused simulation logic (T3.6 Part 1 / 2). Covers: - Memory scenario YAML loads correctly into SimulationCase objects - scripted_user_input() injects the recall question on the configured turn - check_memory_recall() returns hit/miss correctly for single- and multi-detail recall scenarios - check_continuity() flags obvious continuity-break phrasing To run: PYTHONPATH=src pytest tests/test_memory_simulation.py -q -s """ from __future__ import annotations import importlib.util import sys from pathlib import Path from textwrap import dedent import pytest # The data_sources/, schemas/, and scripts/ directories live at the project # root, not under src/. Ensure project root is importable regardless of how # pytest is invoked. PROJECT_ROOT = Path(__file__).resolve().parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from data_sources.memory_scenario_loader import load_memory_scenarios # noqa: E402 from schemas.simulation_case import SimulationCase # noqa: E402 def _load_simulation_module(): """ Load scripts/run_agent_simulation.py as an importable module so we can test its functions directly (the file is a script, not a package). """ script_path = PROJECT_ROOT / "scripts" / "run_agent_simulation.py" spec = importlib.util.spec_from_file_location( "run_agent_simulation_under_test", script_path ) module = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(module) return module SIM = _load_simulation_module() # --------------------------------------------------------------------------- # Test #1: YAML scenarios load correctly # --------------------------------------------------------------------------- def test_default_memory_scenarios_load() -> None: """The shipped memory_scenarios.yaml loads into SimulationCase objects with all memory fields populated.""" scenarios = load_memory_scenarios() assert len(scenarios) >= 6, "expected at least 6 default memory scenarios" for scn in scenarios: assert isinstance(scn, SimulationCase) assert scn.memory_seed, f"{scn.persona_id} missing memory_seed" assert scn.memory_recall_turn is not None, ( f"{scn.persona_id} missing memory_recall_turn" ) assert scn.expected_recall, f"{scn.persona_id} missing expected_recall" assert scn.recall_question, f"{scn.persona_id} missing recall_question" assert scn.distractor_turns, f"{scn.persona_id} missing distractor_turns" def test_multi_detail_scenario_uses_list_expected_recall() -> None: """Scenario 6 was upgraded to a true multi-detail recall test, so its expected_recall must be a list with both seeded facts.""" scenarios = load_memory_scenarios() multi = next(s for s in scenarios if s.persona_id == "MEM-P06") assert isinstance(multi.expected_recall, list) expected_lower = [t.lower() for t in multi.expected_recall] assert "seoul" in expected_lower assert "three months" in expected_lower def test_load_custom_memory_yaml_roundtrip(tmp_path) -> None: """A user-supplied YAML path is honored and parsed into SimulationCases.""" yaml_text = dedent( """ scenarios: - persona_id: "T-P01" module_id: "T-M01" prompt: "test prompt" emotion_before: "calm" ai_behavior: "be supportive" tone: "warm" memory_seed: "I have a cat named Mochi." memory_recall_turn: 2 expected_recall: "Mochi" recall_question: "What was my cat's name?" qa_check_type: "contains_keyword" distractor_turns: - "filler 1" """ ).strip() custom = tmp_path / "custom_mem.yaml" custom.write_text(yaml_text, encoding="utf-8") scenarios = load_memory_scenarios(custom) assert len(scenarios) == 1 s = scenarios[0] assert s.persona_id == "T-P01" assert s.expected_recall == "Mochi" assert s.memory_recall_turn == 2 # --------------------------------------------------------------------------- # Test #2: recall question is injected on the expected turn # --------------------------------------------------------------------------- def _make_scenario(**overrides) -> SimulationCase: base = dict( persona_id="MEM-TEST", module_id="MEM-MOD", prompt="Walking through the park.", emotion_before="calm", ai_behavior="listen", tone="warm", memory_seed="My dog's name is Benny.", memory_recall_turn=3, expected_recall="Benny", recall_question="What was my dog's name?", qa_check_type="contains_keyword", distractor_turns=[ "Distractor one.", "Distractor two.", ], ) base.update(overrides) return SimulationCase(**base) def test_scripted_user_input_seeds_memory_on_turn_zero() -> None: scn = _make_scenario() text = SIM.scripted_user_input(0, scn) assert scn.memory_seed in text assert scn.prompt.split(".")[0] in text # prompt body is included def test_scripted_user_input_returns_distractors_between_seed_and_recall() -> None: scn = _make_scenario() assert SIM.scripted_user_input(1, scn) == scn.distractor_turns[0] assert SIM.scripted_user_input(2, scn) == scn.distractor_turns[1] def test_scripted_user_input_injects_recall_question_on_recall_turn() -> None: """The recall_question must surface on the configured recall turn.""" scn = _make_scenario(memory_recall_turn=3) assert SIM.scripted_user_input(3, scn) == scn.recall_question def test_scripted_user_input_injects_recall_question_when_recall_turn_changes() -> None: """Sanity check: changing memory_recall_turn moves where the question fires.""" scn = _make_scenario(memory_recall_turn=4, distractor_turns=["a", "b", "c"]) assert SIM.scripted_user_input(3, scn) == "c" # last distractor assert SIM.scripted_user_input(4, scn) == scn.recall_question # --------------------------------------------------------------------------- # Test #3: check_memory_recall hit / miss behavior # --------------------------------------------------------------------------- def test_check_memory_recall_hit_single_keyword() -> None: scn = _make_scenario(expected_recall="Benny", memory_recall_turn=3) result = SIM.check_memory_recall( "Yes — you mentioned Benny earlier, your dog.", scn, turn=3 ) assert result["is_recall_turn"] is True assert result["recall_hit"] == 1 assert result["matched_terms"] == ["benny"] assert result["missing_terms"] == [] def test_check_memory_recall_miss_single_keyword() -> None: scn = _make_scenario(expected_recall="Benny", memory_recall_turn=3) result = SIM.check_memory_recall( "I don't think we've discussed any pets yet.", scn, turn=3 ) assert result["is_recall_turn"] is True assert result["recall_hit"] == 0 assert "benny" in result["missing_terms"] def test_check_memory_recall_returns_none_off_recall_turn() -> None: scn = _make_scenario(memory_recall_turn=3) result = SIM.check_memory_recall("any response", scn, turn=2) assert result["is_recall_turn"] is False assert result["recall_hit"] is None def test_check_memory_recall_skips_when_not_a_memory_scenario() -> None: scn = SimulationCase( persona_id="NOMEM", module_id="X", prompt="p", emotion_before="calm", ) result = SIM.check_memory_recall("anything", scn, turn=3) assert result["is_recall_turn"] is False assert result["recall_hit"] is None def test_check_memory_recall_multi_detail_hit_requires_all_terms() -> None: """Multi-detail recall must require *every* listed keyword.""" scn = _make_scenario( expected_recall=["Seoul", "three months"], memory_recall_turn=3, ) response = "You mentioned moving from Seoul about three months ago." result = SIM.check_memory_recall(response, scn, turn=3) assert result["recall_hit"] == 1 assert set(result["matched_terms"]) == {"seoul", "three months"} assert result["missing_terms"] == [] def test_check_memory_recall_multi_detail_miss_when_any_term_absent() -> None: """Reproducing only one of two seeded facts must be a MISS.""" scn = _make_scenario( expected_recall=["Seoul", "three months"], memory_recall_turn=3, ) response = "You mentioned moving from Seoul, right?" result = SIM.check_memory_recall(response, scn, turn=3) assert result["recall_hit"] == 0 assert "three months" in result["missing_terms"] assert "seoul" in result["matched_terms"] # --------------------------------------------------------------------------- # Test #4: check_continuity catches obvious breaks # --------------------------------------------------------------------------- @pytest.mark.parametrize( "response", [ "I don't know what you're referring to.", "We haven't discussed that before.", "I'm not sure what you mean by earlier.", "This is our first conversation, so I have no context.", "I don't have any previous context for this chat.", ], ) def test_check_continuity_flags_obvious_breaks(response: str) -> None: scn = _make_scenario() result = SIM.check_continuity(response, turn=2, scenario=scn) assert result["continuity_break"] == 1, ( f"expected continuity break for response: {response!r}" ) def test_check_continuity_passes_normal_response() -> None: scn = _make_scenario() result = SIM.check_continuity( "Of course — you mentioned earlier that Benny is your dog.", turn=2, scenario=scn, ) assert result["continuity_break"] == 0 def test_check_continuity_skips_turn_zero_and_non_memory_scenarios() -> None: scn = _make_scenario() # turn 0 is the opening, before memory is seeded — not evaluated assert SIM.check_continuity("any text", turn=0, scenario=scn)["continuity_break"] is None # non-memory scenarios skip the continuity check entirely plain = SimulationCase( persona_id="NOMEM", module_id="X", prompt="p", emotion_before="calm", ) assert SIM.check_continuity( "I don't know what you're referring to.", turn=2, scenario=plain )["continuity_break"] is None