from datetime import UTC, date, datetime from pathlib import Path from beaver_gateway.conversations.texts import UserSaid from beaver_gateway.conversations.envelope import RecallContext from beaver_agent.hands.calendars import Calendars, Event, Feed from beaver_agent.memory.recall import ( LABEL, Recall, ReplyLog, frontmatter_aliases, mentions, notes_head, routine_names, ) def _card(path: Path, aliases: list[str] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) fm = "aliases:\n" + "".join(f" - {a}\n" for a in aliases or []) path.write_text( f"---\n{fm}tags:\n - люди/друг\n---\n# 👤 {path.stem}\n", encoding="utf-8" ) def _vault(tmp_path: Path, calendar: Calendars | None = None) -> Recall: people = tmp_path / "👤 люди" _card(people / "личное" / "Зина.md", ["Зина"]) _card(people / "личное" / "Прохор.md", ["Прохор Тестов"]) _card( people / "профессиональное" / "программирование" / "Фёдор Плюшкин.md", ["Фёдор Плюшкин", "Фёдор"], ) _card(people / "личное" / "архив" / "Глафира.md") _card(people / "личное" / "рандомы" / "Устинья Пример.md") diary = tmp_path / "📅 дни" diary.mkdir() (diary / "2026-08-31.md").write_text("с [[Зина|Зиной]] гуляли\n", encoding="utf-8") (diary / "2026-08-26.md").write_text( "[[Зина]] звонила, [[Фёдор Плюшкин|Фёдор]] рассказывал\n", encoding="utf-8" ) (diary / "2026-08-13.md").write_text("обед с [[Зина|Зины]]\n", encoding="utf-8") (diary / "2026-08-01.md").write_text("[[Зина]] ещё раз\n", encoding="utf-8") notes = tmp_path / "мета" / "бобер" / "наблюдения" notes.mkdir(parents=True) (notes / "Зина - наблюдения.md").write_text( "# Зина - наблюдения\n\n## сейчас\n- просила перезвонить вечером 26.08\n" "- ждёт ответа про поездку до четверга\n\n## паттерны\n" "- переносит встречи: [[2026-08-10]], [[2026-08-26]]\n\n## факты\n\n## открытое\n\n" "## лента\n- 2026-08-26 · перенесла встречу на час · [[2026-08-26]]\n" "- 2026-08-31 · прислала список покупок · транскрипт\n", encoding="utf-8", ) (notes / "Прохор - наблюдения.md").write_text( "# Прохор - наблюдения\n\n## сейчас\n\n## лента\n- 2026-08-07 · приехал · [[2026-08-07]]\n", encoding="utf-8", ) boards = tmp_path / "📆 доски" boards.mkdir() (boards / "организация.md").write_text( "## повседнев\n\n- [ ] оплатить счёт за свет 📅 2026-09-01\n- [ ] купить лампочки 📅 2026-09-02\n" "- [ ] далеко 📅 2026-09-20\n- [x] сделано 📅 2026-08-30\n\n***\n\n## Archive\n\n- [ ] старое 📅 2025-01-01\n", encoding="utf-8", ) return Recall( vault=tmp_path, people_dir=people, notes_dir=notes, diary_dir=diary, boards_dir=boards, tz="Europe/Warsaw", calendar=calendar, ) def test_frontmatter_aliases_handles_empty_and_lists() -> None: assert frontmatter_aliases("---\naliases:\ntags:\n---\n# x") == () assert frontmatter_aliases( "---\naliases:\n - Фёдор Плюшкин\n - Фёдор\ntags:\n---\n" ) == ("Фёдор Плюшкин", "Фёдор") assert frontmatter_aliases("нет фронтматтера") == () def test_mentions_by_case_and_link_but_archive_only_by_link(tmp_path: Path) -> None: recall = _vault(tmp_path) people = recall.people.all() names = [p.name for p in mentions("созвон с Фёдором, потом к Зине", people)] assert names == ["Фёдор Плюшкин", "Зина"] assert [p.name for p in mentions("Глафира опять написала", people)] == [] assert [ p.name for p in mentions("глянь [[Глафира]] и [[Устинья Пример]]", people) ] == ["Глафира", "Устинья Пример"] assert [p.name for p in mentions("Прохором обсуждали", people)] == ["Прохор"] # "Ван"+2 буквы ловит "ванна" - цена падежей; на голое имя короче 4 букв не смотрим assert [ p.name for p in mentions("машина без людей, эля тут ни при чём", people) ] == [] def test_block_points_to_card_notes_diary_and_deadlines_once(tmp_path: Path) -> None: recall = _vault(tmp_path) now = datetime(2026, 9, 1, 12, 0, tzinfo=UTC) block = recall.block(RecallContext(text="что там у Зины", kind="master", now=now)) assert block is not None lines = block.splitlines() assert lines[0] == LABEL assert lines[1] == ( "👤 Зина - карточка `👤 люди/личное/Зина.md`; записки " "`мета/бобер/наблюдения/Зина - наблюдения.md`: лента 2 стр., паттернов 1; " "дневник: [[2026-08-31]], [[2026-08-26]], [[2026-08-13]]; сейчас:" ) assert lines[2:4] == [ " - просила перезвонить вечером 26.08", " - ждёт ответа про поездку до четверга", ] assert lines[4:] == [ "📆 09-01 организация: оплатить счёт за свет", "📆 09-02 организация: купить лампочки", ] artem = recall.person_line( next(p for p in recall.people.all() if p.name == "Прохор") ) assert "лента 1 стр., последняя: 2026-08-07 · приехал" in artem assert "сейчас:" not in artem again = recall.block(RecallContext(text="и ещё про Зину", kind="master", now=now)) assert again is not None and "📆" not in again branch = recall.block(RecallContext(text="Фёдор звонил", kind="branch", now=now)) assert branch is not None assert ( "записок нет (`мета/бобер/наблюдения/Фёдор Плюшкин - наблюдения.md`)" in branch ) assert "дневник: [[2026-08-26]]" in branch assert ( recall.block(RecallContext(text="просто привет", kind="branch", now=now)) is None ) def test_reply_log_appends_one_line_per_message_and_strips_seed(tmp_path: Path) -> None: log = ReplyLog(tmp_path / "реплики", tz="Europe/Warsaw") at = datetime(2026, 9, 1, 12, 4, tzinfo=UTC) log.write( UserSaid( conversation_id="c1", kind="master", title=None, text="привет\n\nты меня любишь", at=at, ) ) log.write( UserSaid( conversation_id="c2", kind="branch", title="крипто-карта", text="[сид: clean] branch, 2026-09-01.\n\nсравни Trustee и RedotPay", at=at, ) ) log.write( UserSaid( conversation_id="c3", kind="branch", title="x", text="[сид: brief] только сид", at=at, ) ) text = (tmp_path / "реплики" / "2026-09.md").read_text(encoding="utf-8") assert text.startswith("# реплики Бобра, 2026-09\n") assert text.endswith( "- 2026-09-01 14:04 · мастер · привет ⏎ ты меня любишь\n" "- 2026-09-01 14:04 · ветка «крипто-карта» · сравни Trustee и RedotPay\n" ) assert date.fromisoformat("2026-09-01") # tz shift stays inside the month here def test_notes_head_picks_sections_for_the_seed(tmp_path: Path) -> None: recall = _vault(tmp_path) path = recall.notes_dir / "Зина - наблюдения.md" head = notes_head(path, ("сейчас", "паттерны")) assert head == ( "## сейчас\n- просила перезвонить вечером 26.08\n" "- ждёт ответа про поездку до четверга\n" "## паттерны\n- переносит встречи: [[2026-08-10]], [[2026-08-26]]" ) assert notes_head(path, ("факты",)) is None assert notes_head(recall.notes_dir / "нет.md", ("сейчас",)) is None short = notes_head(path, ("сейчас", "паттерны"), cap=2) assert short is not None and short.endswith("… ещё 3 строк в файле") def _ics(*events: str) -> str: body = "".join(f"BEGIN:VEVENT\n{e}\nEND:VEVENT\n" for e in events) return f"BEGIN:VCALENDAR\nVERSION:2.0\n{body}END:VCALENDAR\n" def _calendar(text: str | None) -> Calendars: """Фид, который отдаёт готовый ical; None - фид, который падает.""" class Fake(Calendars): def fetch(self, url: str) -> str: if text is None: raise OSError("фид не ответил") return text return Fake((Feed("личный", "https://cal.example/x.ics"),), tz="Europe/Warsaw") def _daily(title: str, clock: str) -> str: return f"SUMMARY:{title}\nDTSTART;TZID=Europe/Warsaw:20260902T{clock}\nRRULE:FREQ=DAILY" SINGLES = ( "SUMMARY:Созвон\nDTSTART:20260902T160000Z", "SUMMARY:Врач\nDTSTART:20260903T090000Z", "SUMMARY:Analiza matematyczna II - Egzamin\nDTSTART;VALUE=DATE-TIME:20260905T110000\n" "DESCRIPTION:Sala: 3\\nA23 - CW\\n\nLOCATION:ul. Piotrowo 2\\, Poznań", "SUMMARY:за окном\nDTSTART:20260925T080000Z", ) ROUTINE = ( _daily("Завтрак", "080000"), _daily("Обед", "150000"), _daily("Ужин", "200000"), _daily("Перекус", "180000"), _daily("Wind down", "223000"), ) BUSY = _ics(*SINGLES, *ROUTINE) def _section(recall: Recall, now: datetime, text: str = "привет") -> list[str]: block = recall.block(RecallContext(text=text, kind="master", now=now)) assert block is not None return [ln for ln in block.splitlines() if "календар" in ln or ln.startswith(" -")] def test_calendar_shows_every_single_event_and_folds_the_daily_routine( tmp_path: Path, ) -> None: recall = _vault(tmp_path, _calendar(BUSY)) now = datetime(2026, 9, 1, 12, 0, tzinfo=UTC) lines = _section(recall, now) assert lines == [ "🗓 календарь (7 дней), разовых 3, повседневных 31, данные на 2026-09-01 14:00:", " - 09-02 18:00 Созвон", " - 09-03 11:00 Врач", " - 09-05 11:00 Analiza matematyczna II - Egzamin · Sala: 3, ul. Piotrowo 2, Poznań", " - повседневных повторов: 31 (Завтрак, Обед, Перекус, Ужин, Wind down)", ] # секция дневная, как и сроки досок: второй конверт за те же сутки её не несёт again = recall.block(RecallContext(text="и ещё", kind="master", now=now)) assert again is None or "календарь" not in again def test_no_amount_of_routine_can_push_out_a_single_event(tmp_path: Path) -> None: noise = tuple(_daily(f"быт {i}", f"{i:02d}0000") for i in range(9, 24)) recall = _vault(tmp_path, _calendar(_ics(*SINGLES, *noise))) now = datetime(2026, 9, 1, 12, 0, tzinfo=UTC) lines = _section(recall, now) assert lines[0].startswith("🗓 календарь (7 дней), разовых 3, повседневных 96") assert [ln for ln in lines if "Egzamin" in ln] assert ( len([ln for ln in lines if ln.startswith(" - ")]) == 4 ) # 3 разовых + свёртка assert lines[-1].startswith(" - повседневных повторов: 96") assert "и др." in lines[-1] def test_routine_line_counts_and_names_up_to_five(tmp_path: Path) -> None: events = [ Event( start=datetime(2026, 9, 2, 8, tzinfo=UTC), title=t, place="", all_day=False, routine=True, ) for t in ("Завтрак", "Обед", "Завтрак", "Ужин") ] assert routine_names(events) == "Завтрак, Обед, Ужин" assert routine_names(events, limit=2) == "Завтрак, Обед и др." def test_calendar_section_is_printed_even_when_the_week_is_empty( tmp_path: Path, ) -> None: recall = _vault( tmp_path, _calendar(_ics("SUMMARY:потом\nDTSTART:20261101T080000Z")) ) now = datetime(2026, 9, 1, 12, 0, tzinfo=UTC) block = recall.block(RecallContext(text="просто привет", kind="master", now=now)) assert block is not None assert ( "🗓 календарь (7 дней): событий нет, данные на 2026-09-01 14:00" in block.splitlines() ) def test_calendar_failure_shows_up_in_the_envelope_instead_of_disappearing( tmp_path: Path, ) -> None: recall = _vault(tmp_path, _calendar(None)) now = datetime(2026, 9, 1, 12, 0, tzinfo=UTC) block = recall.block(RecallContext(text="просто привет", kind="master", now=now)) assert block is not None assert "🗓 календарь: ОШИБКА личный: OSError: фид не ответил" in block.splitlines() # ветке справка календарь не носит, как и сроки досок branch = recall.block(RecallContext(text="просто привет", kind="branch", now=now)) assert branch is None def test_without_configured_feeds_there_is_no_calendar_section(tmp_path: Path) -> None: recall = _vault(tmp_path) now = datetime(2026, 9, 1, 12, 0, tzinfo=UTC) block = recall.block(RecallContext(text="просто привет", kind="master", now=now)) assert block is not None and "календарь" not in block