beaver-agent/scripts/seed_memory.py

#!/usr/bin/env python3
# ruff: noqa: E501, EXE001, S603, S108, PLR0915 - одноразовый инструмент, не сервис
"""Одноразовый засев памяти агента из истории vault (решение h, 2026-09-01).

Гоняет `claude -p` на маке (подписка, как бенч) по трём источникам и пишет в
локальный vault - Sync донесёт до dell:

- `💬 чаты/` (не тесты, ещё не в индексе) → выжимка в `мета/бобер/выжимки/`
  в формате дистиллятора + строка в `индекс.md`, плюс наблюдения по людям;
- `📅 дни/` (заполненные, помесячно) → наблюдения по людям и о Бобре;
- `🧠 мысли/` (по файлу) → наблюдения, в основном о Бобре.

Наблюдения - `мета/бобер/наблюдения/<Имя> - наблюдения.md`, строка =
`- дата · факт · [[источник]]`; имена только из карточек `👤 люди/` плюс
«Бобёр». В карточку добавляется строка `наблюдения: [[Имя - наблюдения]]`.
Сырые ответы модели лежат в `scripts/seed_out/` - повторный запуск их
переиспользует (resume), файлы vault дописываются без дублей.

    python3 scripts/seed_memory.py --dry-run
    python3 scripts/seed_memory.py --only chats --limit 2
    python3 scripts/seed_memory.py            # всё
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import date
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent))
from beaver_gateway.agents.prompts import assemble  # noqa: E402

from beaver_agent.memory.recall import People  # noqa: E402

VAULT = Path.home() / "Documents/obsidian/hh"
BEAVER = VAULT / "мета/бобер"
CHATS = VAULT / "💬 чаты"
DIARY = VAULT / "📅 дни"
THOUGHTS = VAULT / "🧠 мысли"
PEOPLE = VAULT / "👤 люди"
DIGESTS = BEAVER / "выжимки"
INDEX = BEAVER / "индекс.md"
NOTES = BEAVER / "наблюдения"
OUT = HERE / "seed_out"
MODEL = "claude-opus-5"
MAX_CHARS = 350_000
BOBR = "Бобёр"

PROMPTS = BEAVER / "промпты"
GRANULES = PROMPTS / "гранулы"
VOICE = tuple(
    (tag, GRANULES / "голос" / f"{tag}.md")
    for tag in (
        "role",
        "philosophy",
        "user_profile",
        "operating_modes",
        "how_you_operate",
        "interaction_guidelines",
    )
)
# та же сборка, что у мастера в config.py, только окружение - джоб
SEED_BUILD = (
    *VOICE,
    ("profile", PROMPTS / "профиль.md"),
    ("corrections", PROMPTS / "поправки.md"),
    ("vault_map", GRANULES / "карта-vault.md"),
    ("environment", GRANULES / "окружения" / "джоб.md"),
    ("dispatcher", GRANULES / "диспетчер.md"),
)

DIGEST_MARK = "=== ВЫЖИМКА ==="
NOTES_MARK = "=== НАБЛЮДЕНИЯ ==="
END_MARK = "=== КОНЕЦ ==="


@dataclass(frozen=True)
class Source:
    kind: str  # chats | days | thoughts
    id: str  # имя для seed_out и [[ссылки]]
    title: str
    text: str
    when: str  # дата источника YYYY-MM-DD (для наблюдений без даты в тексте)
    link: str  # что ставить в [[...]] у наблюдений


# ---------------------------------------------------------------- prompts --


def system_prompt(names: list[str]) -> str:
    """Настоящие гранулы Менеджера Бобрения + блок задания на засев."""
    distiller = (GRANULES / "дистиллятор.md").read_text(encoding="utf-8")
    keep = distiller[distiller.find("Что сохранять:") :].strip()
    known = ", ".join(names)
    seed = f"""Это засев памяти, одноразовый джоб (решение Бобра, 2026-09-01): тебе по одному отдают
документы из истории vault - старые чаты с тобой, месяцы дневника, заметки-мысли, - и ты
отдаёшь то, что из них должно остаться у тебя в памяти. Тулз нет, файлы запишет gateway из
твоего ответа. Ответ - строго в формате ниже, ничего вне маркеров.

{DIGEST_MARK}
(только для чата; для дневника и мысли блок пропускается целиком)
# тема
## контекст
2-4 строки: зачем открывали.
## решили
- решения и их причины; цифры, имена, даты, [[ссылки]] дословно
## открытые вопросы
- что не закрыто
## ссылки
- [[люди]], [[проекты]], соседние чаты - только реально упомянутые
{NOTES_MARK}
Имя :: YYYY-MM-DD · факт
{END_MARK}

Выжимка - по правилам дистиллятора: нейтральный язык, третье лицо («Бобёр решил»), без
мнения, которого в разговоре не было. {keep}

Наблюдения - твои записки из блока «Память» оверлея: то, чего нет в vault и что понадобится,
чтобы заметить повтор - реакция человека, отговорка, обещание, оценка, состояние, паттерн,
событие в отношениях. Не биография и не пересказ карточки (возраст, где живёт), не общие слова.
Имя - ровно одно из списка ниже, как написано, или «{BOBR}» для самого Бобра; людей не из
списка не записывай. Дата - из текста, если она там есть, иначе дата документа. Одна строка,
до 200 знаков, коротко и по факту; на человека до 6 строк с документа, на {BOBR} - до 12.
Нет наблюдений - блок пустой.

Известные имена: {known}"""
    parts = list(SEED_BUILD)
    text = assemble(parts)
    return f"{text}\n<seed>\n{seed}\n</seed>\n"


def user_prompt(src: Source) -> str:
    what = {
        "chats": "чат с агентом",
        "days": "месяц дневника",
        "thoughts": "заметка-мысль",
    }[src.kind]
    return f"Документ: {what}, «{src.title}», дата {src.when}.\n\n{src.text}"


# ---------------------------------------------------------------- sources --

CALLOUT_RE = re.compile(r"^> \[!(tool|thinking)\]")


def strip_chat(text: str) -> str:
    """Без фронтматтера, тул-коллов и thinking: только реплики."""
    if text.startswith("---"):
        end = text.find("\n---", 3)
        if end > 0:
            text = text[end + 4 :]
    out: list[str] = []
    skipping = False
    for line in text.splitlines():
        if CALLOUT_RE.match(line):
            skipping = True
            continue
        if skipping and line.startswith(">"):
            continue
        skipping = False
        out.append(line)
    cleaned = re.sub(r"\n{3,}", "\n\n", "\n".join(out)).strip()
    return cleaned[:MAX_CHARS]


def indexed_chats() -> set[str]:
    if not INDEX.exists():
        return set()
    return set(re.findall(r"\[\[([^\]]+)\]\] →", INDEX.read_text(encoding="utf-8")))


def chat_sources() -> list[Source]:
    done = indexed_chats()
    out: list[Source] = []
    for path in sorted(CHATS.rglob("*.md")):
        if (
            "тест" in path.stem.lower()
            or path.stat().st_size < 5_000
            or path.stem in done
        ):
            continue
        text = strip_chat(path.read_text(encoding="utf-8", errors="ignore"))
        if len(text) < 1_500:
            continue
        when = (
            path.stem[:10]
            if re.match(r"\d{4}-\d{2}-\d{2}", path.stem)
            else date.today().isoformat()
        )
        out.append(
            Source("chats", f"chat_{path.stem}", path.stem, text, when, path.stem)
        )
    return out


def day_sources() -> list[Source]:
    months: dict[str, list[Path]] = {}
    for path in sorted(DIARY.glob("????-??-??.md")):
        if path.stat().st_size > 700:
            months.setdefault(path.stem[:7], []).append(path)
    out: list[Source] = []
    for month, files in sorted(months.items()):
        parts = []
        for f in files:
            body = f.read_text(encoding="utf-8", errors="ignore")
            body = body.split("\n# планы", 1)[0]  # хвост с tasks-запросами
            parts.append(f"### [[{f.stem}]]\n{body.strip()}")
        text = "\n\n".join(parts)[:MAX_CHARS]
        out.append(
            Source(
                "days",
                f"days_{month}",
                f"дневник {month}",
                text,
                f"{month}-01",
                "дневник",
            )
        )
    return out


def thought_sources() -> list[Source]:
    out: list[Source] = []
    for path in sorted(THOUGHTS.rglob("*.md")):
        if path.stat().st_size < 400:
            continue
        m = re.match(r"(\d{4}-\d{2}-\d{2})", path.stem)
        when = m.group(1) if m else date.today().isoformat()
        text = path.read_text(encoding="utf-8", errors="ignore")[:MAX_CHARS]
        out.append(
            Source("thoughts", f"thought_{path.stem}", path.stem, text, when, path.stem)
        )
    return out


# ------------------------------------------------------------------ model --


def _result(payload: str) -> dict:
    """`claude -p --output-format json` отдаёт либо объект, либо массив сообщений."""
    data = json.loads(payload)
    if isinstance(data, list):
        results = [x for x in data if isinstance(x, dict) and x.get("type") == "result"]
        return results[-1] if results else {}
    return data if isinstance(data, dict) else {}


def ask(src: Source, system: str) -> str:
    raw = OUT / f"{src.id}.json"
    if raw.exists():
        try:
            data = _result(raw.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            data = {}
        if data.get("result") and not data.get("is_error"):
            return data["result"]
    cmd = [
        "claude",
        "-p",
        user_prompt(src),
        "--system-prompt",
        system,
        "--model",
        MODEL,
        "--effort",
        "medium",
        "--tools",
        "",
        "--setting-sources",
        "",
        "--strict-mcp-config",
        "--disable-slash-commands",
        "--no-session-persistence",
        "--output-format",
        "json",
    ]
    proc = subprocess.run(
        cmd, capture_output=True, text=True, timeout=1200, cwd="/tmp", check=False
    )
    OUT.mkdir(exist_ok=True)
    raw.write_text(
        proc.stdout or json.dumps({"error": proc.stderr[-2000:]}), encoding="utf-8"
    )
    if proc.returncode != 0 or not proc.stdout:
        msg = f"{src.id}: claude exit {proc.returncode}: {proc.stderr[-500:]}"
        raise RuntimeError(msg)
    data = _result(proc.stdout)
    if data.get("is_error") or not data.get("result"):
        msg = f"{src.id}: {data.get('result') or data}"
        raise RuntimeError(msg)
    return data["result"]


def parse(answer: str) -> tuple[str | None, list[tuple[str, str, str]]]:
    digest = None
    if DIGEST_MARK in answer and NOTES_MARK in answer:
        digest = answer.split(DIGEST_MARK, 1)[1].split(NOTES_MARK, 1)[0].strip() or None
        if digest and digest.startswith("("):  # «(только для чата ...)» пропуск
            digest = None
    notes_part = answer.split(NOTES_MARK, 1)[1] if NOTES_MARK in answer else ""
    notes_part = notes_part.split(END_MARK, 1)[0]
    notes: list[tuple[str, str, str]] = []
    for line in notes_part.splitlines():
        m = re.match(
            r"^\s*-?\s*(.+?)\s*::\s*(\d{4}-\d{2}-\d{2})\s*[·\-]\s*(.+?)\s*$", line
        )
        if m:
            notes.append((m.group(1).strip(), m.group(2), m.group(3).strip()))
    return digest, notes


# ------------------------------------------------------------------ write --

SLUG_BAD = re.compile(r"[^\w\s-]", re.UNICODE)


def slugify(text: str, maxlen: int = 40) -> str:
    cleaned = re.sub(r"[\s-]+", "-", SLUG_BAD.sub(" ", text).strip()).strip("-")
    return cleaned[:maxlen].rstrip("-") or "untitled"


def write_digest(src: Source, body: str, today: str) -> str:
    DIGESTS.mkdir(parents=True, exist_ok=True)
    stem = f"{today} - {slugify(src.title)}"
    path = DIGESTS / f"{stem}.md"
    n = 2
    while path.exists() and f"[[{src.title}]]" not in path.read_text(encoding="utf-8"):
        path = DIGESTS / f"{stem} ({n}).md"
        n += 1
    path.write_text(
        f'---\ntype: выжимка\nsource: "[[{src.title}]]"\ndate: {today}\n---\n\n{body.strip()}\n',
        encoding="utf-8",
    )
    line = f"- {today} [[{src.title}]] → [[{path.stem}]]"
    text = (
        INDEX.read_text(encoding="utf-8")
        if INDEX.exists()
        else "# индекс\n\nстрока на выжимку: чат → его выжимка.\n"
    )
    if line not in text.splitlines():
        INDEX.write_text(text.rstrip("\n") + "\n" + line + "\n", encoding="utf-8")
    return path.stem


def note_file(name: str) -> Path:
    return NOTES / f"{name} - наблюдения.md"


def append_notes(name: str, lines: list[str]) -> int:
    NOTES.mkdir(parents=True, exist_ok=True)
    path = note_file(name)
    if not path.exists():
        path.write_text(
            f"# {name} - наблюдения\n\n> папка терапевта: пишет менеджер бобрения. "
            "Строка = дата · факт · источник. Бобёр по умолчанию не читает, но может.\n\n",
            encoding="utf-8",
        )
    existing = set(path.read_text(encoding="utf-8").splitlines())
    fresh = [ln for ln in lines if ln not in existing]
    if fresh:
        with path.open("a", encoding="utf-8") as fh:
            fh.write("\n".join(fresh) + "\n")
    return len(fresh)


def sort_notes() -> None:
    """Сортирует и дедупит только `## лента`; шапку (сейчас/паттерны/…) не трогает."""
    for path in NOTES.glob("* - наблюдения.md"):
        text = path.read_text(encoding="utf-8")
        head_text, sep, feed = text.partition("## лента\n")
        bullets = [ln for ln in feed.splitlines() if ln.startswith("- ")]
        bullets = sorted(dict.fromkeys(bullets), key=lambda ln: ln[2:12])
        if not sep:
            head_text = head_text.rstrip("\n") + "\n\n"
            head_text = head_text.replace("\n- ", "\n").rstrip() + "\n\n"
        path.write_text(
            head_text.rstrip("\n") + "\n\n## лента\n" + "\n".join(bullets) + "\n",
            encoding="utf-8",
        )


def link_card(person_path: Path, name: str) -> bool:
    text = person_path.read_text(encoding="utf-8")
    link = f"[[{name} - наблюдения]]"
    if link in text:
        return False
    person_path.write_text(
        text.rstrip("\n") + f"\n\nнаблюдения: {link}\n", encoding="utf-8"
    )
    return True


# ------------------------------------------------------------------- main --


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--only", choices=["chats", "days", "thoughts"])
    ap.add_argument("--limit", type=int)
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--jobs", type=int, default=4)
    ap.add_argument("--pick", help="подстрока в названии источника")
    args = ap.parse_args()

    # хабы групп («бывшие», «рандомы») - не люди, наблюдения им не пишем
    people = [
        p for p in People(PEOPLE, VAULT, ttl=10_000).all() if "/группы/" not in p.path
    ]
    by_name = {p.name: p for p in people}
    for p in people:  # алиасы тоже принимаем, но пишем под именем карточки
        for a in p.aliases:
            by_name.setdefault(a, p)
    names = sorted(p.name for p in people)
    system = system_prompt(names)

    sources: list[Source] = []
    if args.only in (None, "chats"):
        sources += chat_sources()
    if args.only in (None, "days"):
        sources += day_sources()
    if args.only in (None, "thoughts"):
        sources += thought_sources()
    if args.pick:
        sources = [s for s in sources if args.pick.lower() in s.title.lower()]
    if args.limit:
        sources = sources[: args.limit]
    total_chars = sum(len(s.text) for s in sources)
    print(
        f"источников: {len(sources)}, символов: {total_chars:,} (~{total_chars // 3.5:,.0f} токенов)"
    )
    for s in sources:
        print(f"  {s.kind:8} {s.title[:60]:60} {len(s.text):>8}")
    if args.dry_run:
        return

    today = date.today().isoformat()
    written: dict[str, int] = {}
    linked: list[str] = []
    failures: list[str] = []
    with ThreadPoolExecutor(max_workers=args.jobs) as pool:
        futs = {pool.submit(ask, s, system): s for s in sources}
        for fut in as_completed(futs):
            src = futs[fut]
            try:
                answer = fut.result()
            except Exception as exc:  # noqa: BLE001
                failures.append(str(exc))
                print(f"FAIL {src.id}: {exc}"[:300])
                continue
            digest, notes = parse(answer)
            digest_stem = (
                write_digest(src, digest, today)
                if src.kind == "chats" and digest
                else None
            )
            grouped: dict[str, list[str]] = {}
            unknown: set[str] = set()
            for raw_name, when, fact in notes:
                name = (
                    BOBR
                    if raw_name.casefold() in (BOBR.casefold(), "бобер", "h", "андрей")
                    else None
                )
                if name is None:
                    person = by_name.get(raw_name)
                    if person is None:
                        unknown.add(raw_name)
                        continue
                    name = person.name
                link = f"[[{src.link}]]" if src.kind != "days" else ""
                if src.kind == "days":
                    link = (
                        f"[[{when}]]"
                        if (DIARY / f"{when}.md").exists()
                        else "[[дневник]]"
                    )
                grouped.setdefault(name, []).append(f"- {when} · {fact} · {link}")
            for name, lines in grouped.items():
                n = append_notes(name, lines)
                written[name] = written.get(name, 0) + n
                if (
                    name != BOBR
                    and name in by_name
                    and link_card(VAULT / by_name[name].path, name)
                ):
                    linked.append(name)
            print(
                f"ok   {src.id}: digest={digest_stem or '-'} notes={sum(len(v) for v in grouped.values())}"
                + (f" unknown={sorted(unknown)}" if unknown else "")
            )
    sort_notes()
    print("\nнаблюдений добавлено по людям:")
    for name, n in sorted(written.items(), key=lambda kv: -kv[1]):
        print(f"  {n:4}  {name}")
    if linked:
        print("ссылка добавлена в карточки:", ", ".join(sorted(set(linked))))
    if failures:
        print(f"\nошибок: {len(failures)}")
        sys.exit(1)


if __name__ == "__main__":
    main()