From 5fa81f9b050f4b3ca453a429d2f59bb28db18197 Mon Sep 17 00:00:00 2001 From: max Date: Sun, 25 Oct 2026 08:31:52 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20recall=20=D0=B4=D0=BB=D1=8F=20=D1=80?= =?UTF-8?q?=D0=B5=D0=B3=D0=B8=D1=81=D1=82=D1=80=D0=B0=D1=82=D0=BE=D1=80?= =?UTF-8?q?=D0=B0=20=E2=80=94=20=D0=BF=D1=80=D0=B8=D0=B2=D1=8F=D0=B7=D0=BA?= =?UTF-8?q?=D0=B0=20=D1=81=D0=B5=D1=81=D1=81=D0=B8=D0=B9=20=D0=BA=20=D1=80?= =?UTF-8?q?=D0=B5=D0=BF=D0=BE,=20=D0=B4=D0=BE=D1=81=D0=BA=D0=B0=20=D1=81?= =?UTF-8?q?=D1=82=D0=B0=D1=82=D1=83=D1=81=D0=BE=D0=B2,=20=D1=80=D0=B0?= =?UTF-8?q?=D0=B7=D0=BD=D0=BE=D0=BE=D0=B1=D1=80=D0=B0=D0=B7=D0=BD=D0=B0?= =?UTF-8?q?=D1=8F=20=D0=B2=D1=8B=D0=B4=D0=B0=D1=87=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - repos.py: сессия привязывается к репо по путям во входах вызовов инструментов (Edit, cd, git -C, Codex workdir/exec). Worktree, симлинк-псевдонимы и удалённые Codex-worktree сводятся к одному чекауту; чужому репо нужно >=3 вызова. Факты — в session_repos/session_info/repo_scans, старый индекс догоняется без новых эмбеддингов. - project= во всех поисковых тулзах; scope_cwd включает сессии, работавшие с репо из другой папки; у хитов новое поле repos. - recent_sessions: первый настоящий запрос, последний запрос и ответ, ветка, repos, started_by (user/agent/automation), active. - recall_search: не больше 2 хитов на сессию (per_session) с добором; сниппет вырезается вокруг совпадения с запросом, шапка compaction срезается. - Хаб: project/per_session пробрасываются клиентом и сервером. Co-Authored-By: Claude Opus 5.5 --- README.es-ES.md | 14 + README.md | 12 + README.zh-CN.md | 10 + docs/README.ru.md | 13 + .../2026-10-25-recall-for-registrar.md | 107 ++++ src/session_recall/hub/app.py | 8 +- src/session_recall/hub/remote.py | 18 +- src/session_recall/index.py | 27 +- src/session_recall/models.py | 5 + src/session_recall/repos.py | 338 ++++++++++++ src/session_recall/retrieve.py | 161 ++++-- src/session_recall/server.py | 54 +- src/session_recall/snippets.py | 94 ++++ src/session_recall/store.py | 137 ++++- tests/test_hub_http.py | 12 + tests/test_registrar.py | 489 ++++++++++++++++++ 16 files changed, 1435 insertions(+), 64 deletions(-) create mode 100644 docs/decisions/2026-10-25-recall-for-registrar.md create mode 100644 src/session_recall/repos.py create mode 100644 src/session_recall/snippets.py create mode 100644 tests/test_registrar.py diff --git a/README.es-ES.md b/README.es-ES.md index aba6f48..fb2b35d 100644 --- a/README.es-ES.md +++ b/README.es-ES.md @@ -106,6 +106,20 @@ legible. `grep` escanea bajo demanda **todas** las transcripciones indexadas — turnos internos (salida de herramientas, razonamiento) que nunca se convirtieron en fragmentos de búsqueda. Solo bajo demanda: sin inyección proactiva de contexto en cada prompt. +**A qué proyecto pertenece una sesión.** El `project` de un ancla es la carpeta donde se +lanzó la sesión; sus `repos` son aquello en lo que la sesión trabajó de verdad. Cada ruta que +nombran sus llamadas a herramientas (una edición, un `cd`, un `workdir` de Codex) se resuelve +al repo git que la contiene — los worktrees y los alias por symlink colapsan a un solo +checkout — y un repo distinto de la carpeta de lanzamiento necesita al menos tres llamadas +para contar. Pasa `project="harmonia"` a cualquier herramienta de descubrimiento para obtener +las sesiones que trabajaron en ese repo, se lanzaran donde se lanzaran; `scope_cwd` también +las incluye. `recall_search` devuelve como mucho dos anclas por sesión (`per_session`, `0` = +sin límite) para que una sesión larga no llene la página, y recorta cada fragmento alrededor +del pasaje que coincide con la consulta. `recent_sessions` se lee como un tablero de estado: +la primera petición real, la última petición y respuesta, rama, repos, `started_by` (`user`, +`agent` o `automation`, inferido de los metadatos de la transcripción) y `active` (un turno +en los últimos 10 minutos). +
Ver una llamada completa diff --git a/README.md b/README.md index 577463c..edd69ad 100644 --- a/README.md +++ b/README.md @@ -98,6 +98,18 @@ Ranked anchors carry provenance and a human-readable timestamp. `grep` scans **a transcripts on demand — including under-the-hood turns (tool output, thinking) that never became search chunks. On-demand only: no proactive context injection into every prompt. +**Which project a session belongs to.** A hit's `project` is the folder the session was +launched in; its `repos` are what that session actually worked on. Every path its tool calls +name (an edit, a `cd`, a Codex `workdir`) is resolved to the owning git repo — worktrees and +symlinked aliases collapse to one checkout — and a repo other than the launch folder needs at +least three calls to count. Pass `project="harmonia"` to any discovery tool to get the +sessions that worked on that repo wherever they ran; `scope_cwd` includes them too. +`recall_search` keeps at most two hits per session (`per_session`, `0` = off) so one long +session cannot fill the page, and cuts each snippet around the passage that matches the +query. `recent_sessions` reads like a status board: the first real ask, the latest ask and +answer, branch, repos, `started_by` (`user`, `agent` or `automation`, inferred from +transcript metadata) and `active` (a turn in the last 10 minutes). +
See a complete call diff --git a/README.zh-CN.md b/README.zh-CN.md index f3d7991..b0c07d4 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -95,6 +95,16 @@ flowchart LR `grep` 按需扫描**所有**已索引的会话记录——包括那些从未成为搜索分块的幕后轮次 （工具输出、思考过程）。一切只按需进行：不会向每条提示词主动注入上下文。 +**会话属于哪个项目。** 锚点上的 `project` 是会话启动时所在的文件夹；`repos` 才是该会话真正 +动过的仓库。工具调用里出现的每个路径（一次编辑、一次 `cd`、Codex 的 `workdir`）都会解析到 +拥有它的 git 仓库——worktree 和符号链接别名会折叠成同一个检出——而启动文件夹以外的仓库至少 +要被调用三次才算数。给任何检索工具传 `project="harmonia"`，就能拿到所有动过这个仓库的会话， +无论它们在哪里启动；`scope_cwd` 也会包含它们。`recall_search` 每个会话最多保留两条结果 +（`per_session`，`0` 表示不限），避免一个长会话占满整页，并把摘要截在与查询匹配的段落附近。 +`recent_sessions` 读起来像一块状态看板：第一个真正的请求、最新的请求和回答、分支、仓库、 +`started_by`（`user`、`agent` 或 `automation`，依据会话记录的元数据推断）以及 `active` +（最近 10 分钟内有过轮次）。 +
查看一次完整调用 diff --git a/docs/README.ru.md b/docs/README.ru.md index ca87037..05d3a80 100644 --- a/docs/README.ru.md +++ b/docs/README.ru.md @@ -104,6 +104,19 @@ flowchart LR подкапотные реплики (вывод инструментов, размышления), которые так и не стали поисковыми чанками. Только по требованию: никакой проактивной инъекции контекста в каждый промпт. +**К какому проекту относится сессия.** `project` у хита — папка, где сессию запустили; +`repos` — репозитории, с которыми она на самом деле работала. Каждый путь из вызовов +инструментов (правка файла, `cd`, `workdir` у Codex) сводится к git-репозиторию-владельцу — +worktree и симлинки-псевдонимы схлопываются в один чекаут, — а чужому репо, кроме папки +запуска, нужно минимум три вызова, чтобы засчитаться. Передайте `project="harmonia"` в любой +поисковый инструмент — получите сессии, которые работали с этим репо, где бы их ни запустили; +`scope_cwd` их тоже включает. `recall_search` берёт не больше двух хитов на сессию +(`per_session`, `0` — без лимита), чтобы одна длинная сессия не забивала выдачу, и режет +сниппет вокруг места, совпавшего с запросом. `recent_sessions` читается как доска статусов: +первый настоящий запрос, последний запрос и ответ, ветка, репозитории, `started_by` (`user`, +`agent` или `automation`, по метаданным транскрипта) и `active` (реплика за последние +10 минут). +
Посмотреть полный вызов diff --git a/docs/decisions/2026-10-25-recall-for-registrar.md b/docs/decisions/2026-10-25-recall-for-registrar.md new file mode 100644 index 0000000..133b64d --- /dev/null +++ b/docs/decisions/2026-10-25-recall-for-registrar.md @@ -0,0 +1,107 @@ +# Recall as the registrar's memory: repo attribution, status board, diverse hits + +## Контекст + +Максим проектирует «Командный центр»: он говорит только с регистратором — LLM- +коннектором, у которого из инструментов один поиск. Регистратор должен отвечать +«что происходит в проекте X», «кто из агентов что делает», «к какому проекту это +относится», «что мы решили по Y», и его память — session-recall. Аудит живого +индекса 25.10 (875 сессий, Qwen3-Embedding-8B) показал, что на эти вопросы recall +отвечает плохо: + +- `project` — это папка запуска. Сессия 82ef7c9a числилась `food_sferyx` (запущена + в его worktree), а сделала 254 файловые операции в `~/harmonia` и ни одной в + food_sferyx. `recall_search(scope_cwd=~/harmonia)` вернул 2 нерелевантных хита. + `~/exodia` — симлинк на `~/whytap`, но в индексе это два проекта. +- `recent_sessions.label` — первый промпт, а он часто служебный + (`` 95 раз, бриф side-сессии 40, `` 24). Нет + последнего запроса/ответа, ветки, того, кто открыл сессию. +- «что происходит в проекте Гармония» → 7 из 10 хитов из одной сессии. +- Сниппет — первые 200 символов; у 681 сжатого пересказа это всегда одна и та же + шапка «This session is being continued…». + +## Решение + +1. **Атрибуция к репо по вызовам инструментов** (`repos.py`). Каждый путь во + ВХОДАХ вызовов (Edit `file_path`, `cd`, `git -C`, Codex `workdir`, код Codex + `exec`) сводится к git-репо: подъём по ФС до `.git`, `.git`-файл worktree ведёт к + основному чекауту, затем `realpath` (симлинк-псевдоним → один репо). Вызов без + абсолютного пути засчитывается папке запуска. Чужому репо нужно ≥3 вызова; папка + запуска засчитывается всегда (если не git — под своим именем). Факты лежат в + `session_repos` / `session_info` / `repo_scans`, по `file_path`, в той же + пофайловой транзакции, что и чанки. +2. **`project=` во всех поисковых тулзах** и расширенный `scope_cwd`: сессия + проходит фильтр, если запущена в репо ИЛИ работала с ним. `scope_cwd` из подпапки + сводится к корню репо. Anchor получил поле `repos`. +3. **`recent_sessions` как доска статусов**: `label` = первый настоящий запрос + (из брифа side-сессии — блок `Task:`), `last_user`, `last_assistant`, + `git_branch`, `repos`, `started_by` (`user` / `agent` / `automation`), `active`. +4. **Не больше 2 хитов на сессию** в `recall_search` (`per_session`, 0 = выкл.), с + добором в порядке ранга, если сессий меньше, чем слотов. Реранкер теперь + упорядочивает все кандидаты, а не top-k. +5. **Сниппет по запросу** (`snippets.py`): окно, покрывающее больше всего слов + запроса (префиксное совпадение — «Гармония»/«Гармонии»), шапка compaction- + пересказа срезается; без буквального совпадения — начало текста. + +## Почему + +- Вызовы инструментов — единственное надёжное свидетельство, с каким кодом + работала сессия; текст ответа упоминает чужие репо постоянно, а результаты + инструментов (листинги, `ls ~`) — тем более, поэтому учитываются только входы. +- Скан — только текст, без эмбеддингов: старый индекс догоняется одним потоковым + проходом, без повторной оплаты эмбеддера и без смены `EXTRACTOR_VERSION`. +- `started_by` берётся из метаданных (`entrypoint` у Claude, `originator`/`source` + у Codex), а не из слов: промпт подделать легко, метаданные — нет. Две текстовые + формы (бриф side-сессии, ``) — то, что оставляет сама программа. +- Лимит на сессию с добором: регистратору нужен разрез по сессиям, но узкая + история не должна возвращать меньше k. + +## Что протестировали + +- `tests/test_registrar.py` — сценарии из живого индекса: сессия в worktree + одного репо правит другой; симлинк-псевдоним; «взгляд» на соседний README и + чтение плагин-кэша (git-клон в `~/.claude`) не дают атрибуции; удалённый + worktree и не-git папка; scratch-папка в `~/Library`; Codex `workdir` + код + `exec` + `originator: Claude Code`; догон старого индекса без единого нового + эмбеддинга; рост и удаление транскрипта; доска статусов с брифом, + ``, ``, heartbeat, `sdk-cli`; лимит на + сессию и добор; сниппет у compaction-пересказа; проброс `project` через MCP и + клиент хаба. +- Первая версия `grep` с фильтром `project` молча теряла все хиты: локальная + переменная цикла перезаписывала параметр `project` — поймал существующий тест + Codex-интеграции. +- `scope_cwd` из подпапки (`~/harmonia/src`) сначала не находил работу из других + папок: сравнивался путь, а не корень репо. Исправлено отдельным подъёмом до + `.git` без фильтров атрибуции. +- Первый прогон на клоне живого индекса завис на первом же транскрипте: хвост + URL `https://github.com/…` распознавался как путь `//github.com/…`, а POSIX + `dirname("//") == "//"` — подъём к корню не кончался. Плюс `repo_of("/")` через + `normpath("")` превращался в `.` и находил репо текущего каталога. Оба случая + закрыты и покрыты тестом; после фикса тот же транскрипт сканируется за 0.01 с. +- Второй прогон показал дыру: Codex держит worktree в `~/.codex/worktrees//`, + а скрытые папки отбрасывались до поиска `.git` — 44 сессии whytap остались без + атрибуции. Теперь скрытая папка судится по репо, к которому ведёт (worktree → + основной чекаут засчитывается, git-клон плагина в `~/.claude` — нет), а удалённый + Codex-worktree сводится к одноимённому чекауту в `~`. После этого 0 сессий + whytap/exodia без атрибуции. +- Клон живого индекса (876 сессий, 620 транскриптов, 6.8 GB): догон фактов — + 25 с, без единого нового эмбеддинга. whytap: 224 сессии (было 89 + 136 под + двумя именами); harmonia: 5 сессий вместо 1; 4 сессии «food_sferyx» на деле + работали в другом репо; `started_by`: 451 user / 177 agent. +- Живые запросы: «что происходит в проекте Гармония» — 6 разных сессий в топ-10 + вместо 4 (7 хитов было из одной); `scope_cwd=~/harmonia` находит работу из + worktree food_sferyx вместо двух нерелевантных хитов. Локальная часть поиска + (knn+fts) — 0.8 с против 1.0 с до изменений; реранк 100 кандидатов не дороже 10. + +## Отвергли + +- Определять проект по тексту (упоминания имени репо) — шум: MEMORY.md с именами + всех проектов вставлен в каждую сессию. +- `git remote` вместо пути — нужен `git` на каждый путь и не работает на хабе. +- Переписать `chunks.project` — пришлось бы перевыпускать весь индекс и ломать + смысл поля «где запущено», на который опираются старые клиенты. +- Отдельная тулза `project_state` — следующий шаг поверх атрибуции (пункт 6 + аудита), не в этом PR. +- Чистка шума в `grep` (пункт 5 аудита) — отдельным заходом. + +2026-10-25 · PR: см. историю ветки `claude/recall-for-registrar-1220aa` diff --git a/src/session_recall/hub/app.py b/src/session_recall/hub/app.py index 70cb79e..fe68eb2 100644 --- a/src/session_recall/hub/app.py +++ b/src/session_recall/hub/app.py @@ -271,7 +271,8 @@ def _search(self, owner: str, body: dict) -> None: hits = hub.recall.recall_search( query, k=int(body.get("k") or 10), scope_cwd=body.get("scope_cwd"), source=body.get("source"), - start_ts=start_ts, end_ts=end_ts) + start_ts=start_ts, end_ts=end_ts, project=body.get("project"), + per_session=int(body.get("per_session", 2))) owners = owners_for(hub, (a.session_id for a in hits)) self._send(200, { "anchors": [_anchor(a, owners.get(a.session_id)) for a in hits], @@ -302,7 +303,7 @@ def _grep(self, owner: str, body: dict) -> None: hits = hub.recall.grep( pattern, body.get("session_id"), scope_cwd=body.get("scope_cwd"), source=body.get("source"), limit=int(body.get("limit") or 100), - start_ts=start_ts, end_ts=end_ts) + start_ts=start_ts, end_ts=end_ts, project=body.get("project")) owners = owners_for(hub, (a.session_id for a in hits)) self._send(200, { "anchors": [_anchor(a, owners.get(a.session_id)) for a in hits]}) @@ -311,7 +312,8 @@ def _recent(self, owner: str, body: dict) -> None: start_ts, end_ts = _range(body) sessions = hub.recall.recent_sessions( scope_cwd=body.get("scope_cwd"), limit=int(body.get("limit") or 10), - source=body.get("source"), start_ts=start_ts, end_ts=end_ts) + source=body.get("source"), start_ts=start_ts, end_ts=end_ts, + project=body.get("project")) owners = owners_for(hub, (s["session_id"] for s in sessions)) for session in sessions: session["owner"] = owners.get(session["session_id"]) diff --git a/src/session_recall/hub/remote.py b/src/session_recall/hub/remote.py index 79b5c65..28704fb 100644 --- a/src/session_recall/hub/remote.py +++ b/src/session_recall/hub/remote.py @@ -64,10 +64,12 @@ def _turns(rows: list) -> list[Turn]: def recall_search(self, query: str, k: int = 10, scope_cwd: str | None = None, source: str | None = None, start_ts: int | None = None, - end_ts: int | None = None, **_ignored) -> SearchResult: + end_ts: int | None = None, project: str | None = None, + per_session: int = 2, **_ignored) -> SearchResult: body = self._post("/v1/search", { "query": query, "k": k, "scope_cwd": scope_cwd, "source": source, - "start_ts": start_ts, "end_ts": end_ts}) + "start_ts": start_ts, "end_ts": end_ts, "project": project, + "per_session": per_session}) return SearchResult(self._anchors(body.get("anchors", [])), body.get("degraded")) @@ -86,18 +88,22 @@ def step(self, session_id: str, uuid: str, direction: str, count: int = 1, def grep(self, pattern: str, session_id: str | None = None, scope_cwd: str | None = None, source: str | None = None, limit: int = 100, start_ts: int | None = None, - end_ts: int | None = None) -> list[Anchor]: + end_ts: int | None = None, + project: str | None = None) -> list[Anchor]: return self._anchors(self._post("/v1/grep", { "pattern": pattern, "session_id": session_id, "scope_cwd": scope_cwd, "source": source, "limit": limit, - "start_ts": start_ts, "end_ts": end_ts}).get("anchors", [])) + "start_ts": start_ts, "end_ts": end_ts, + "project": project}).get("anchors", [])) def recent_sessions(self, scope_cwd: str | None = None, limit: int = 10, source: str | None = None, start_ts: int | None = None, - end_ts: int | None = None) -> list[dict]: + end_ts: int | None = None, + project: str | None = None) -> list[dict]: return self._post("/v1/recent", { "scope_cwd": scope_cwd, "limit": limit, "source": source, - "start_ts": start_ts, "end_ts": end_ts}).get("sessions", []) + "start_ts": start_ts, "end_ts": end_ts, + "project": project}).get("sessions", []) def ask(self, question: str, k: int = 12, scope_cwd: str | None = None) -> dict: diff --git a/src/session_recall/index.py b/src/session_recall/index.py index bf9b5af..673552d 100644 --- a/src/session_recall/index.py +++ b/src/session_recall/index.py @@ -6,6 +6,7 @@ from .store import Store from .embed import Embedder from .scope import project_label +from .repos import REPO_SCAN_VERSION, RepoResolver, attributed_repos, scan_file from .transcripts import codex_file_header def _embed_fp() -> str: @@ -61,6 +62,24 @@ def _codex_files(codex_dirs: Iterable[Path]): yield jsonl, "codex", "" +def _scan_sessions(store: Store, path: str, source: str, sig: str, + resolver: RepoResolver) -> None: + """Record which repos each session in `path` worked on, and who opened it. + + Text-only (no embedding), so it also runs for files whose chunks are + already current: an index built before attribution existed is backfilled + by one streaming pass instead of a re-embed. Not committed here.""" + facts = scan_file(path, source, resolver) + store.replace_session_facts( + path, source, + {sid: (attributed_repos(f, resolver), f.started_by) for sid, f in facts.items()}, + _scan_sig(sig)) + + +def _scan_sig(sig: str) -> str: + return f"r{REPO_SCAN_VERSION}:{sig}" + + def _forget_file(store: Store, path: str): """Remove a file that is now recognized as an under-the-hood sidechain.""" store.delete_file(path) @@ -89,7 +108,8 @@ def _roots_safe_to_prune(store: Store, roots: Iterable[Path], source: str) -> bo def index_corpus(store: Store, embedder: Embedder, projects_dir: Path | None, - codex_dirs: Iterable[Path] | None = None) -> int: + codex_dirs: Iterable[Path] | None = None, + resolver: RepoResolver | None = None) -> int: """Incrementally index Claude Code plus optional Codex transcript roots. The fourth argument is optional to preserve the original Claude-only Python @@ -124,6 +144,7 @@ def index_corpus(store: Store, embedder: Embedder, projects_dir: Path | None, failed_sources: set[str] = set() jobs = list(_claude_files(projects_dir) or ()) jobs.extend(_codex_files(codex_roots)) + resolver = resolver or RepoResolver() for jsonl, source, project in jobs: # One transaction per file: delete + re-add + marker commit together. # _file_sig and header reads stay INSIDE this isolation for delete races, @@ -137,6 +158,9 @@ def index_corpus(store: Store, embedder: Embedder, projects_dir: Path | None, project = project_label(meta.cwd) or "codex" sig = _file_sig(jsonl, source) if store.is_indexed(str(jsonl), sig): + if store.repo_scan_sig(str(jsonl)) != _scan_sig(sig): + _scan_sessions(store, str(jsonl), source, sig, resolver) + store.commit() continue # Reuse only vectors produced by the same source extractor version # and embedding space. This makes live append-only updates cheap. @@ -169,6 +193,7 @@ def index_corpus(store: Store, embedder: Embedder, projects_dir: Path | None, reused = cached.get(chunk.content_hash) store.add(chunk, reused if reused is not None else fresh[chunk.content_hash]) store.mark_indexed(str(jsonl), sig, source=source) + _scan_sessions(store, str(jsonl), source, sig, resolver) store.commit() new_count += len(chunks) except Exception as e: diff --git a/src/session_recall/models.py b/src/session_recall/models.py index 45a69dc..712f8aa 100644 --- a/src/session_recall/models.py +++ b/src/session_recall/models.py @@ -38,6 +38,11 @@ class Anchor: # both serialise strictly by these fields, and would drop anything else. # WHY: docs/decisions/2026-08-06-team-hub-central-index.md owner: "str | None" = None + # Repos the hit's session actually worked on (from its tool calls), most- + # touched first; `project` above is only the folder it was launched in. + # None where attribution does not apply (Cursor, meta docs). + # WHY: docs/decisions/2026-10-25-recall-for-registrar.md + repos: "list[str] | None" = None @dataclass class Turn: diff --git a/src/session_recall/repos.py b/src/session_recall/repos.py new file mode 100644 index 0000000..ad06f2d --- /dev/null +++ b/src/session_recall/repos.py @@ -0,0 +1,338 @@ +"""Which repositories a session actually worked on, and who started it. + +`chunks.project` is the folder the session was LAUNCHED in. That is often not +the project: a session opened in one repo's worktree edits another repo by +absolute path, a desktop scratch folder works on a real repo, a symlinked +alias (`~/exodia -> whytap`) splits one repo into two names. A question like +"what happened in harmonia" then misses the sessions that did the work. + +The evidence is already in every transcript: the tool calls. Every path a +call names — an Edit's file_path, a `cd`, a `git -C`, a Codex workdir — is +resolved to the git repository that owns it, and the session is attributed +to each repo it touched often enough. Only tool INPUTS count: a result that +merely lists another repo's files is not work on that repo. + +Resolution walks the real filesystem up to the nearest `.git` (a worktree's +`.git` file points back at its main checkout), then follows symlinks, so a +worktree, its main checkout and a symlinked alias all become one root. +Paths that no longer exist fall back to the nearest existing ancestor; a +session's own cwd that is not a git repo still counts under its plain name, +so a session is never left with less attribution than `project` gave it. + +# WHY: docs/decisions/2026-10-25-recall-for-registrar.md +""" + +import os +import re +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path + +from .scope import repo_root +from .transcripts import iter_transcript_events + +# Bump when attribution logic changes: baked into the per-file scan signature, +# so every transcript is re-scanned (no re-embedding — scans are text-only). +REPO_SCAN_VERSION = 1 + +# A repo other than the session's own cwd must be named by at least this many +# tool calls. One `cat` of a neighbour's README is a glance, not work on it. +MIN_TOUCHES = 3 + +# An absolute (or ~-relative) path inside free text: a JSON string, a shell +# command, JS code. Stops at quotes, shell metacharacters and whitespace. +_PATH = re.compile(r"(?(){}\[\],*?$\\]+)") + +# Never a project: system and tool-state locations. Plugin caches and +# Claude/Codex state live in hidden home dirs and are often git clones, so +# reading one SKILL.md must not attribute the session to a plugin. +_SYSTEM_PREFIXES = ("/tmp", "/private", "/var", "/etc", "/usr", "/opt", "/bin", + "/sbin", "/dev", "/System", "/Library", "/Applications", + "/proc", "/sys") + +# Codex keeps worktrees at ~/.codex/worktrees// and deletes +# them after the thread; a gone one has no `.git` file left to follow. +_CODEX_WORKTREE = re.compile(r"/\.codex/worktrees/[^/]+/(?P[^/]+)(?P/.*)?$") + +# Who opened a session. A transcript says it with metadata, not with words: +# Claude's `entrypoint` (sdk-* = driven by a program), Codex's `originator` +# and `source` (exec = codex exec; "Claude Code" = a Codex thread Claude +# opened). A side session is opened by another session with a written brief; +# a Codex automation wakes up on a heartbeat. +_AGENT_BRIEF = "Background (from the session that started this one)" +_HEARTBEAT = "" + + +@dataclass +class SessionFacts: + """What one transcript says about one session.""" + cwd: str = "" + touches: Counter = field(default_factory=Counter) + started_by: str = "" + + +class RepoResolver: + """Path -> owning repo root, memoized per directory. + + One resolver serves a whole index run: thousands of tool calls name the + same few directories, and each lookup is a handful of stat() calls.""" + + def __init__(self, home: str | None = None): + self.home = (home or os.path.expanduser("~")).rstrip("/") + self._cache: dict[str, str | None] = {} + + @staticmethod + def _system(path: str) -> bool: + return any(path == p or path.startswith(p + "/") for p in _SYSTEM_PREFIXES) + + def _ignored(self, path: str) -> bool: + if path == self.home: + return True + if path.startswith(self.home + "/"): + first = path[len(self.home) + 1:].split("/", 1)[0] + return first.startswith(".") or first == "Library" + return self._system(path) + + @staticmethod + def _main_checkout(dot_git: Path) -> Path | None: + """A worktree's `.git` is a file: `gitdir:
/.git/worktrees/`.""" + try: + text = dot_git.read_text(errors="replace") + except OSError: + return None + if not text.startswith("gitdir:"): + return None + gitdir = text[len("gitdir:"):].strip() + marker = "/.git/worktrees/" + if marker in gitdir: + return Path(gitdir.split(marker, 1)[0]) + return None + + def _revive_codex_worktree(self, path: str) -> str: + """A deleted Codex worktree -> the same-named checkout in home, if any.""" + found = _CODEX_WORKTREE.search(path) + if not found or os.path.exists(path[:found.end("name")]): + return path + main = os.path.join(self.home, found["name"]) + if not os.path.exists(os.path.join(main, ".git")): + return path + return main + (found["rest"] or "") + + def repo_of(self, path: str) -> str | None: + """The git repo that owns `path` (a file or a directory), or None.""" + if not path: + return None + path = os.path.expanduser(path) + # "//host/x" is the tail of a URL ("https://host/x"), not a path — + # and POSIX normpath keeps the double slash, so dirname never ends. + if not path.startswith("/") or path.startswith("//"): + return None + # Worktree folders collapse by name first: a deleted worktree no + # longer has the `.git` file that would lead back to its checkout. + path = os.path.normpath(repo_root(path) or "/") + # Only system locations are skipped up front. Hidden home folders are + # judged by the repo they lead to: a Codex worktree under + # ~/.codex/worktrees/ is real work on its main checkout, a plugin + # cache cloned under ~/.claude/ is not. + if path == "/" or path == self.home: + return None + if not path.startswith(self.home + "/") and self._system(path): + return None + path = self._revive_codex_worktree(path) + visited: list[str] = [] + current = path + found: str | None = None + while current and current != "/": + if current in self._cache: + found = self._cache[current] + break + visited.append(current) + if current == self.home: + break + dot_git = Path(current) / ".git" + try: + if dot_git.is_dir(): + found = current + break + if dot_git.is_file(): + main = self._main_checkout(dot_git) + found = str(main) if main else current + break + except OSError: + pass + parent = os.path.dirname(current) + if parent == current: + break + current = parent + if found: + found = os.path.realpath(found) + if self._ignored(found): + found = None + for directory in visited: + self._cache[directory] = found + return found + + def cwd_root(self, cwd: str) -> str | None: + """A session's own folder: its repo, else the folder itself by name. + + A cwd that is not a git checkout (or no longer exists, as on a hub + that indexes someone else's paths) still names the session's place — + dropping it would attribute less than the old `project` label did.""" + if not cwd: + return None + found = self.repo_of(cwd) + if found: + return found + name = repo_root(os.path.expanduser(cwd)) + if not name.startswith("/") or self._ignored(name): + return None + return name + + +def owning_repo(path: str) -> str: + """The repo a search scope means: the checkout that contains `path` + (an agent often runs from `/src`), else the path itself. + + No attribution filters here — the caller named this place explicitly.""" + current = os.path.normpath(repo_root(os.path.expanduser(path)) or "/") + while current.startswith("/") and current != "/": + dot_git = Path(current) / ".git" + try: + if dot_git.is_dir(): + return os.path.realpath(current) + if dot_git.is_file(): + main = RepoResolver._main_checkout(dot_git) + return os.path.realpath(main or current) + except OSError: + break + parent = os.path.dirname(current) + if parent == current: + break + current = parent + return os.path.realpath(path) + + +def _tool_inputs(event) -> list[str]: + """The text of every tool call this event makes (inputs only).""" + obj = event.obj + if event.source == "claude": + message = obj.get("message") or {} + content = message.get("content") + if obj.get("type") != "assistant" or not isinstance(content, list): + return [] + out = [] + for block in content: + if isinstance(block, dict) and block.get("type") == "tool_use": + out.append(_flatten(block.get("input"))) + return out + if event.source == "codex": + payload = obj.get("payload") + if obj.get("type") != "response_item" or not isinstance(payload, dict): + return [] + if payload.get("type") not in {"function_call", "custom_tool_call", + "local_shell_call"}: + return [] + return [_flatten(payload.get("arguments", payload.get("input", + payload.get("action"))))] + return [] + + +def _flatten(value) -> str: + if isinstance(value, str): + return value + if isinstance(value, dict): + return "\n".join(_flatten(v) for v in value.values()) + if isinstance(value, list): + return "\n".join(_flatten(v) for v in value) + return "" + + +def _first_user_text(event) -> str | None: + obj = event.obj + if event.source == "claude": + message = obj.get("message") or {} + if obj.get("type") == "user" and isinstance(message.get("content"), str): + from .extract import _clean_user_text + return _clean_user_text(message["content"]) + return None + if event.source == "codex" and event.role == "user" and event.content.strip(): + return event.content + return None + + +def classify_starter(source: str, entrypoint: str, originator: str, + codex_source: str, first_text: str) -> str: + """'user' | 'agent' | 'automation' — who opened the session. + + Metadata first (it cannot be faked by a prompt), then the two written + shapes another program leaves as the first message.""" + text = (first_text or "").lstrip() + if text.startswith(_HEARTBEAT): + return "automation" + if text.startswith(_AGENT_BRIEF): + return "agent" + if source == "claude" and entrypoint.startswith("sdk"): + return "agent" + if source == "codex" and (codex_source == "exec" or originator == "Claude Code"): + return "agent" + return "user" + + +def scan_file(path: str, source: str, + resolver: RepoResolver) -> dict[str, SessionFacts]: + """session_id -> facts, streamed in one pass over a transcript.""" + facts: dict[str, SessionFacts] = {} + starters: dict[str, dict] = {} + for event in iter_transcript_events(path, source=source): + sid = event.session_id + if not sid: + continue + session = facts.setdefault(sid, SessionFacts()) + if event.cwd and not session.cwd: + session.cwd = event.cwd + meta = starters.setdefault(sid, {"entrypoint": "", "originator": "", + "codex_source": "", "text": None}) + obj = event.obj + if source == "claude" and not meta["entrypoint"] and obj.get("entrypoint"): + meta["entrypoint"] = str(obj["entrypoint"]) + if source == "codex" and obj.get("type") == "session_meta" \ + and not meta["originator"]: + payload = obj.get("payload") or {} + meta["originator"] = str(payload.get("originator") or "") + raw_source = payload.get("source") + meta["codex_source"] = raw_source if isinstance(raw_source, str) else "" + if meta["text"] is None: + text = _first_user_text(event) + if text: + meta["text"] = text + for text in _tool_inputs(event): + roots = {resolver.repo_of(p) for p in _PATH.findall(text)} + roots.discard(None) + if not roots: + # A call that names no absolute path runs where the session + # is (or its Codex workdir, which _PATH already caught). + here = resolver.cwd_root(event.cwd) + roots = {here} if here else set() + for root in roots: + session.touches[root] += 1 + for sid, session in facts.items(): + meta = starters[sid] + session.started_by = classify_starter( + source, meta["entrypoint"], meta["originator"], meta["codex_source"], + meta["text"] or "") + return facts + + +def attributed_repos(session: SessionFacts, + resolver: RepoResolver) -> list[tuple[str, int]]: + """(root, touches) the session is attributed to, most-touched first. + + The session's own cwd repo always counts (a pure discussion still belongs + where it was held); any other repo needs MIN_TOUCHES tool calls.""" + home = resolver.cwd_root(session.cwd) + out = [(root, n) for root, n in session.touches.items() + if n >= MIN_TOUCHES or root == home] + if home and home not in session.touches: + out.append((home, 0)) + out.sort(key=lambda item: (-item[1], item[0])) + return out diff --git a/src/session_recall/retrieve.py b/src/session_recall/retrieve.py index 8ddbe66..b33fc7e 100644 --- a/src/session_recall/retrieve.py +++ b/src/session_recall/retrieve.py @@ -8,6 +8,7 @@ from .rerank import Reranker from .models import Anchor, Turn from .scope import repo_root, in_scope, project_label +from .snippets import human_ask, query_snippet from .timefmt import humanize_ts from .transcripts import ( TranscriptEvent, @@ -20,6 +21,37 @@ def _snippet(text: str, n: int = 200) -> str: return text[:n] + ("…" if len(text) > n else "") +# A session is a unit of work: ten hits from one session read as one answer +# repeated. Past this many per session, other sessions get the slots first. +PER_SESSION = 2 +# "Active" in recent_sessions: some turn landed within this window. +ACTIVE_WINDOW_S = 600 + + +def diversify(items: list, k: int, per_session: int, key) -> list: + """Top-k in rank order with at most `per_session` items per session. + + Overflow is not thrown away: when there are fewer distinct sessions than + slots, the remaining slots are back-filled in rank order, so a narrow + history still returns k hits. per_session <= 0 disables the cap.""" + if per_session <= 0: + return items[:k] + picked: list[int] = [] + overflow: list[int] = [] + counts: dict = {} + for i, item in enumerate(items): + if len(picked) == k: + break + sid = key(item) + if counts.get(sid, 0) < per_session: + counts[sid] = counts.get(sid, 0) + 1 + picked.append(i) + else: + overflow.append(i) + picked += overflow[:k - len(picked)] + return [items[i] for i in sorted(picked)] + + def _match_snippet(text: str, pattern: str, n: int = 240) -> str: """Return a bounded grep snippet centered near the actual match.""" index = text.find(pattern) @@ -56,16 +88,29 @@ def _validate_source(source: str | None) -> None: f"source must be 'claude', 'codex', 'cursor', or omitted; got {source!r}") @staticmethod - def _anchor(c, score: "float | None") -> Anchor: + def _anchor(c, score: "float | None", query: str = "") -> Anchor: return Anchor(session_id=c.session_id, uuid=c.uuid, role=c.role, - snippet=_snippet(c.text), score=score, project=c.project, - when=c.ts, source=c.source) + snippet=query_snippet(c.text, query), score=score, + project=c.project, when=c.ts, source=c.source) + + def _with_repos(self, anchors: list[Anchor]) -> list[Anchor]: + """Attach each hit's session attribution (labels, most-touched first).""" + keys = {(a.source, a.session_id) for a in anchors + if a.source in ("claude", "codex")} + repos = self.store.session_repos(keys) if keys else {} + for a in anchors: + found = repos.get((a.source, a.session_id)) + if found is not None: + a.repos = [r["repo"] for r in found] + return anchors def recall_search(self, query: str, k: int = 10, candidates: int = 100, scope_cwd: str | None = None, source: str | None = None, start_ts: int | None = None, - end_ts: int | None = None) -> list[Anchor]: + end_ts: int | None = None, + project: str | None = None, + per_session: int = PER_SESSION) -> list[Anchor]: self._validate_source(source) # scope_cwd is the agent's raw cwd; normalize to a repo root so the main # checkout and every worktree under it collapse to one scope. None = global. @@ -93,7 +138,7 @@ def recall_search(self, query: str, k: int = 10, candidates: int = 100, qv = self.embedder.embed_query(query) for cid, d in self.store.knn( qv, candidates, scope_root=root, source=source, - start_ts=start_ts, end_ts=end_ts): + start_ts=start_ts, end_ts=end_ts, project=project): order.append(cid) dist[cid] = d except Exception as exc: @@ -106,7 +151,7 @@ def recall_search(self, query: str, k: int = 10, candidates: int = 100, f"only literal word matches are returned ({type(exc).__name__}: {exc})") for cid in self.store.fts( query, candidates, scope_root=root, source=source, - start_ts=start_ts, end_ts=end_ts): + start_ts=start_ts, end_ts=end_ts, project=project): if cid not in dist: order.append(cid) dist[cid] = None # keyword match, no vector distance @@ -131,27 +176,35 @@ def recall_search(self, query: str, k: int = 10, candidates: int = 100, # Rerank if a reranker is configured AND reachable. The reranker is OPTIONAL (some # embedding providers ship none) and may be down — either way we fall back to the # KNN/FTS candidate order so recall never hard-fails. + # The reranker orders ALL candidates (it scores every document either + # way): the per-session cap below needs the full order to back-fill. ranked: list[tuple[int, float]] | None = None if self.reranker is not None: try: ranked = self.reranker.rerank( - query, [chunk_by_id[cid].text for cid in distinct], top_k=k) + query, [chunk_by_id[cid].text for cid in distinct], + top_k=len(distinct)) except Exception: ranked = None if ranked is not None: - return SearchResult( - [self._anchor(chunk_by_id[distinct[idx]], score) for idx, score in ranked], - degraded) - # no reranker: KNN-similarity order; score monotonic in similarity, metric- - # agnostic. Keyword-only hits carry None (no distance), not a fake 0.0 that - # reads as "irrelevant" at the tool boundary. - out: list[Anchor] = [] - for cid in distinct[:k]: - d = dist.get(cid) - score = round(1.0 / (1.0 + d), 4) if isinstance(d, (int, float)) else None - out.append(self._anchor(chunk_by_id[cid], score)) - return SearchResult(out, degraded) + scored = [(distinct[idx], score) for idx, score in ranked] + else: + # no reranker: KNN-similarity order; score monotonic in similarity, + # metric-agnostic. Keyword-only hits carry None (no distance), not a + # fake 0.0 that reads as "irrelevant" at the tool boundary. + scored = [] + for cid in distinct: + d = dist.get(cid) + scored.append((cid, round(1.0 / (1.0 + d), 4) + if isinstance(d, (int, float)) else None)) + picked = diversify(scored, k, per_session, + key=lambda item: (chunk_by_id[item[0]].source, + chunk_by_id[item[0]].session_id)) + return SearchResult( + self._with_repos([self._anchor(chunk_by_id[cid], score, query) + for cid, score in picked]), + degraded) def _files_for(self, uuid: str, session_id: str | None, source: str | None = None) -> list[str]: @@ -297,20 +350,26 @@ def grep(self, pattern: str, session_id: str | None = None, scope_cwd: str | None = None, source: str | None = None, limit: int = 100, start_ts: int | None = None, - end_ts: int | None = None) -> list[Anchor]: + end_ts: int | None = None, + project: str | None = None) -> list[Anchor]: self._validate_source(source) if limit < 1: return [] root = repo_root(scope_cwd) if scope_cwd else None + # Sessions that worked on the scoped repo / named project from + # elsewhere: they pass the filter whatever their cwd says. + touched_root = self.store.sessions_touching(root=root) if root else set() + touched_project = (self.store.sessions_touching(repo=project) + if project else set()) # Per-path chunk metadata. A file can carry SEVERAL (sid, cwd) pairs — # resumed sessions mix sessionIds in one transcript — so this is only a # fast-path skip (skip a file when NO row could match) and a fallback for # turns lacking their own fields; the authoritative filter is per turn. meta: dict[str, list[tuple[str, str, str, str]]] = {} - for sid, path, project, cwd, chunk_source in self.store.db.execute( + for sid, path, chunk_project, cwd, chunk_source in self.store.db.execute( "SELECT DISTINCT session_id, file_path, project, cwd, source FROM chunks" ).fetchall(): - meta.setdefault(path, []).append((sid, project, cwd, chunk_source)) + meta.setdefault(path, []).append((sid, chunk_project, cwd, chunk_source)) # Scan ALL indexed transcripts, not just chunk-bearing ones: a file whose # every turn was filtered at extract (harness boilerplate, tool-only) is # exactly the under-the-hood content grep exists for. @@ -345,16 +404,22 @@ def grep(self, pattern: str, session_id: str | None = None, if session_id and t_sid != session_id: continue t_cwd = event.cwd or (rows[0][2] if rows else "") - if root: + t_key = (event.source, t_sid) + if root and t_key not in touched_root: if t_cwd: if not in_scope(t_cwd, root): continue else: continue # no cwd evidence -> not provably in scope + label = None + if project and t_key not in touched_project: + label = self._grep_label(t_cwd, rows) + if label.lower() != project.lower(): + continue blob = json.dumps(sanitize_raw(event.obj), ensure_ascii=False) if pattern in blob: - project = (project_label(t_cwd) if t_cwd - else (rows[0][1] if rows else "")) + if label is None: + label = self._grep_label(t_cwd, rows) self._anchor_files[(event.source, t_sid, event.uuid)] = path hits.append(Anchor( session_id=t_sid, @@ -362,21 +427,26 @@ def grep(self, pattern: str, session_id: str | None = None, role=event.role or event.type, snippet=_match_snippet(blob, pattern), score=1.0, - project=project, + project=label, when=event.ts, source=event.source, )) if len(hits) >= limit: - return hits + return self._with_repos(hits) except OSError: continue - return hits + return self._with_repos(hits) + + @staticmethod + def _grep_label(cwd: str, rows) -> str: + return project_label(cwd) if cwd else (rows[0][1] if rows else "") def recent_sessions(self, scope_cwd: str | None = None, limit: int = 10, now: int | None = None, source: str | None = None, start_ts: int | None = None, - end_ts: int | None = None) -> list[dict]: + end_ts: int | None = None, + project: str | None = None) -> list[dict]: self._validate_source(source) # Freshest sessions first — answers "what's the current state / how fresh is # the index" (the top entry's last_activity IS the effective freshness) and @@ -386,15 +456,40 @@ def recent_sessions(self, scope_cwd: str | None = None, limit: int = 10, root = repo_root(scope_cwd) if scope_cwd else None now = int(time.time()) if now is None else now out: list[dict] = [] - for row_source, sid, project, last_ts, turns in self.store.recent_sessions( - root, limit, source=source, start_ts=start_ts, end_ts=end_ts): + rows = self.store.recent_sessions( + root, limit, source=source, start_ts=start_ts, end_ts=end_ts, + project=project) + repos = self.store.session_repos( + (s, sid) for s, sid, *_ in rows if s in ("claude", "codex")) + for row_source, sid, label, last_ts, turns in rows: + # Who is doing what, at a glance: the first real ask (not the + # side-session envelope or a desktop event), the latest ask and + # the latest answer, where it runs and what it actually touches. + first_ask = self._ask(sid, row_source, newest=False) out.append({ "session_id": sid, "source": row_source, - "project": project, + "project": label, + "repos": repos.get((row_source, sid), []), + "git_branch": self.store.latest_branch(sid, row_source), + "started_by": self.store.started_by(row_source, sid), "turns": turns, "last_activity": last_ts, "last_activity_human": humanize_ts(last_ts, now), - "label": _snippet(self.store.first_user_text(sid, row_source), 120), + "active": bool(last_ts) and now - last_ts <= ACTIVE_WINDOW_S, + "label": _snippet(first_ask or self.store.first_user_text( + sid, row_source), 120), + "last_user": _snippet(self._ask(sid, row_source, newest=True) or "", 200), + "last_assistant": _snippet(next(iter(self.store.session_texts( + sid, row_source, "assistant", newest=True, limit=1)), ""), 200), }) return out + + def _ask(self, session_id: str, source: str, newest: bool) -> str | None: + """The first (or latest) turn in which someone actually asked something.""" + for text in self.store.session_texts(session_id, source, "user", + newest=newest, limit=25): + ask = human_ask(text) + if ask: + return ask + return None diff --git a/src/session_recall/server.py b/src/session_recall/server.py index a28e96a..2a2de67 100644 --- a/src/session_recall/server.py +++ b/src/session_recall/server.py @@ -23,6 +23,8 @@ def _adict(a) -> dict: # ten times learns nothing, and the hub keeps the field meaningful. if d.get("owner") is None: d.pop("owner", None) + if d.get("repos") is None: + d.pop("repos", None) return d @@ -58,7 +60,8 @@ def _r() -> Recall: def recall_search(query: str, k: int = 10, scope_cwd: str | None = None, source: str | None = None, start_date: str | None = None, end_date: str | None = None, timezone: str | None = None, - on_date: str | None = None) -> dict: + on_date: str | None = None, project: str | None = None, + per_session: int = 2) -> dict: """Semantically search past Claude Code, Codex and Cursor sessions. Returns {"anchors": [...ranked anchors...], "degraded": null | str}. @@ -69,9 +72,19 @@ def recall_search(query: str, k: int = 10, scope_cwd: str | None = None, Retry with the exact identifiers you expect (error strings, symbols, file names), tell the user the search is degraded, and treat a miss as inconclusive. + Each anchor: project = the folder the session was launched in; repos = the + repos that session actually worked on (from its tool calls), most-touched + first — use repos to say which project a hit belongs to. + scope_cwd: pass your current working directory to restrict results to the - current project/repo (worktrees collapse to the repo root). Omit it for a + current project/repo (worktrees collapse to the repo root; sessions that + worked on the repo from another folder are included). Omit it for a global, cross-project search. + project: a repo/project name (e.g. "harmonia"). Matches sessions that + actually worked on that repo — by their tool calls, wherever they were + launched — plus sessions launched in a folder of that name. + per_session: at most this many hits per session before other sessions get + the slots (back-filled when history is narrow); 0 = no cap. source: optionally restrict to "claude", "codex", or "cursor"; omit for the shared index. start_date/end_date: inclusive local calendar dates (YYYY-MM-DD). Either may @@ -83,7 +96,8 @@ def recall_search(query: str, k: int = 10, scope_cwd: str | None = None, start_date, end_date, timezone, on_date=on_date) hits = _r().recall_search( query, k=k, scope_cwd=scope_cwd, source=source, - start_ts=start_ts, end_ts=end_ts) + start_ts=start_ts, end_ts=end_ts, project=project, + per_session=per_session) return {"anchors": [_adict(a) for a in hits], "degraded": getattr(hits, "degraded", None)} @@ -108,11 +122,15 @@ def step(session_id: str, uuid: str, direction: str, count: int = 1, def grep(pattern: str, session_id: str | None = None, scope_cwd: str | None = None, source: str | None = None, limit: int = 100, start_date: str | None = None, end_date: str | None = None, - timezone: str | None = None, on_date: str | None = None) -> list[dict]: + timezone: str | None = None, on_date: str | None = None, + project: str | None = None) -> list[dict]: """On-demand substring scan over raw session transcripts. scope_cwd: pass your current working directory to restrict the scan to the current project/repo; omit for a global scan. + project: a repo/project name (e.g. "harmonia"). Matches sessions that + actually worked on that repo — by their tool calls, wherever they were + launched — plus sessions launched in a folder of that name. source: optionally restrict to "claude", "codex", or "cursor". limit: maximum number of matches returned (default 100). start_date/end_date: inclusive local calendar dates (YYYY-MM-DD). @@ -123,7 +141,7 @@ def grep(pattern: str, session_id: str | None = None, scope_cwd: str | None = No start_date, end_date, timezone, on_date=on_date) return [_adict(a) for a in _r().grep( pattern, session_id, scope_cwd=scope_cwd, source=source, limit=limit, - start_ts=start_ts, end_ts=end_ts)] + start_ts=start_ts, end_ts=end_ts, project=project)] @mcp.tool() @@ -131,26 +149,36 @@ def recent_sessions(scope_cwd: str | None = None, limit: int = 10, source: str | None = None, start_date: str | None = None, end_date: str | None = None, timezone: str | None = None, - on_date: str | None = None) -> list[dict]: - """List the most recently active past sessions, freshest first — use to see the - current state of work and how fresh the index is (the top entry's - last_activity_human is the effective freshness). Also surfaces the sessions of a - thread split across resume-created session_ids so you can reassemble the arc. + on_date: str | None = None, + project: str | None = None) -> list[dict]: + """List the most recently active past sessions, freshest first — use to see + who is working on what right now, the current state of work, and how fresh + the index is (the top entry's last_activity_human is the effective + freshness). Also surfaces the sessions of a thread split across + resume-created session_ids so you can reassemble the arc. scope_cwd: pass your current working directory to restrict to the current project/repo (worktrees collapse to the repo root); omit for all projects. + project: a repo/project name (e.g. "harmonia"). Matches sessions that + actually worked on that repo — by their tool calls, wherever they were + launched — plus sessions launched in a folder of that name. source: optionally restrict to "claude", "codex", or "cursor". start_date/end_date: inclusive local calendar dates (YYYY-MM-DD). on_date: shorthand for one local calendar day; cannot be combined with a range. timezone: IANA timezone override; omit it to use the user's computer timezone. - Each entry: source, session_id, project, turns, last_activity (epoch), - last_activity_human, label (the session's first user prompt). + Each entry: source (the agent: claude/codex/cursor), session_id, project + (launch folder), repos ([{repo, path, touches}] actually worked on, most + first), git_branch, started_by ("user" | "agent" — another session or a + program opened it | "automation" — a scheduled heartbeat; inferred from + transcript metadata), turns, last_activity (epoch), last_activity_human, + active (a turn in the last 10 minutes), label (the first real ask), + last_user (the latest ask), last_assistant (the latest answer). """ start_ts, end_ts = date_range_to_epoch( start_date, end_date, timezone, on_date=on_date) return _r().recent_sessions( scope_cwd=scope_cwd, limit=limit, source=source, - start_ts=start_ts, end_ts=end_ts) + start_ts=start_ts, end_ts=end_ts, project=project) def main(): diff --git a/src/session_recall/snippets.py b/src/session_recall/snippets.py new file mode 100644 index 0000000..0e02cb1 --- /dev/null +++ b/src/session_recall/snippets.py @@ -0,0 +1,94 @@ +"""Short human-readable text for tool output: hit snippets and session labels. + +A snippet is the only part of a hit an agent reads before deciding to expand +it. The first 200 characters of a chunk are often the wrong 200: a compacted +session's summary opens with the same boilerplate sentence every time, and a +long answer mentions the asked-about thing only halfway down. So a snippet is +cut around the passage that matches the query, and boilerplate that every +such chunk shares is dropped first. + +The same problem hits session labels: the first "user" message of a session +is often not a person asking anything — a side-session brief, a desktop +event, a pasted block, a Codex heartbeat. `human_ask` finds the actual ask. + +# WHY: docs/decisions/2026-10-25-recall-for-registrar.md +""" + +import re + +_COMPACTION = re.compile( + r"^This session is being continued from a previous conversation[^\n]*\n" + r"(?:[^\n]*\n){0,3}?\s*Summary:\s*", re.IGNORECASE) +_WORD = re.compile(r"\w+", re.UNICODE) + +# Machine blocks that reach the index inside user turns. Matched pairs only, +# like extract._HARNESS_BLOCK; content of a pasted block is the user's own. +_EVENT_BLOCK = re.compile( + r"<(side-sessions-event|ci-monitor-event|task-notification|system-reminder)" + r"\b[^>]*>.*?", re.DOTALL) +_PASTED_TAG = re.compile(r"]*>") +_BRIEF = "Background (from the session that started this one)" +_BRIEF_TASK = re.compile(r"(?m)^Task:\s*") + + +def strip_boilerplate(text: str) -> str: + """Drop the fixed preamble of a compacted-session summary.""" + return _COMPACTION.sub("", text, count=1) + + +def _stem(term: str) -> str: + # Crude, language-agnostic prefix match: "Гармония"/"Гармонии", + # "indexing"/"indexed" meet on a shared prefix. + return term if len(term) <= 4 else term[:max(4, len(term) - 2)] + + +def query_snippet(text: str, query: str, n: int = 200) -> str: + """A ≤n-char window of `text` that covers the most distinct query terms. + + Falls back to the opening of the text when no term occurs literally — + a purely semantic hit has no better anchor than its start.""" + body = strip_boilerplate(text).strip() or text + stems = {_stem(w.lower()) for w in _WORD.findall(query or "") if len(w) >= 3} + lower = body.lower() + best_start, best_score = 0, 0 + if stems and len(body) > n: + for stem in stems: + start = 0 + for _ in range(50): + pos = lower.find(stem, start) + if pos < 0: + break + start = pos + 1 + window_start = max(0, pos - n // 4) + window = lower[window_start:window_start + n] + score = sum(1 for s in stems if s in window) + if score > best_score or (score == best_score and window_start < best_start): + best_start, best_score = window_start, score + if best_start: + # Start on a word boundary rather than mid-word. + space = body.find(" ", best_start, best_start + 20) + if space >= 0: + best_start = space + 1 + piece = body[best_start:best_start + n] + return (("…" if best_start else "") + piece + + ("…" if best_start + n < len(body) else "")) + + +def human_ask(text: str) -> str | None: + """What a person (or the delegating agent) actually asked in this turn. + + None when the turn is pure machinery: a compaction summary, a desktop + event, a heartbeat. A side-session brief yields its Task part.""" + if not text: + return None + if _COMPACTION.match(text): + return None + cleaned = _EVENT_BLOCK.sub("", text) + cleaned = _PASTED_TAG.sub("", cleaned).strip() + if cleaned.startswith(""): + return None + if cleaned.startswith(_BRIEF): + task = _BRIEF_TASK.search(cleaned) + if task: + cleaned = cleaned[task.end():].strip() + return cleaned or None diff --git a/src/session_recall/store.py b/src/session_recall/store.py index fc911ec..8503f87 100644 --- a/src/session_recall/store.py +++ b/src/session_recall/store.py @@ -2,6 +2,7 @@ # sqlite3 is compiled without SQLITE_ENABLE_LOAD_EXTENSION, so enable_load_extension() # and load_extension() are absent. pysqlite3 (wheel) provides them, which sqlite-vec # requires to load vec0. The rest of the API (execute, fetchall, etc.) is identical. +import os import re import sqlite3 if not hasattr(sqlite3.Connection, "enable_load_extension"): @@ -10,6 +11,7 @@ import sqlite_vec from .config import EMBED_DIM from .models import Chunk +from .repos import owning_repo from .scope import scope_clause _COLS = ["session_id", "uuid", "role", "text", "project", "cwd", @@ -97,6 +99,29 @@ def _schema(self): "CREATE INDEX IF NOT EXISTS idx_chunks_ts ON chunks(ts)") self.db.execute( "CREATE INDEX IF NOT EXISTS idx_chunks_source_ts ON chunks(source, ts)") + # Per-session facts read from tool calls and metadata (repos.py): which + # repos a session worked on and who opened it. Keyed by file_path like + # chunks, so the per-file delete/re-index cycle keeps them in step; + # repo_scans records which file signature they were computed from. + # WHY: docs/decisions/2026-10-25-recall-for-registrar.md + self.db.execute( + "CREATE TABLE IF NOT EXISTS session_repos(file_path TEXT, source TEXT, " + "session_id TEXT, root TEXT, repo TEXT, touches INTEGER)") + self.db.execute( + "CREATE INDEX IF NOT EXISTS idx_session_repos_session " + "ON session_repos(source, session_id)") + self.db.execute( + "CREATE INDEX IF NOT EXISTS idx_session_repos_file ON session_repos(file_path)") + self.db.execute( + "CREATE TABLE IF NOT EXISTS session_info(file_path TEXT, source TEXT, " + "session_id TEXT, started_by TEXT)") + self.db.execute( + "CREATE INDEX IF NOT EXISTS idx_session_info_session " + "ON session_info(source, session_id)") + self.db.execute( + "CREATE INDEX IF NOT EXISTS idx_session_info_file ON session_info(file_path)") + self.db.execute( + "CREATE TABLE IF NOT EXISTS repo_scans(path TEXT PRIMARY KEY, sig TEXT)") self.db.commit() def add(self, chunk: Chunk, embedding: "list[float] | bytes") -> int: @@ -157,6 +182,74 @@ def delete_file(self, path: str): f"DELETE FROM {table} WHERE chunk_id IN " "(SELECT id FROM chunks WHERE file_path = ?)", (path,)) self.db.execute("DELETE FROM chunks WHERE file_path = ?", (path,)) + self._delete_session_facts(path) + + def _delete_session_facts(self, path: str) -> None: + self.db.execute("DELETE FROM session_repos WHERE file_path = ?", (path,)) + self.db.execute("DELETE FROM session_info WHERE file_path = ?", (path,)) + self.db.execute("DELETE FROM repo_scans WHERE path = ?", (path,)) + + def replace_session_facts(self, path: str, source: str, + sessions: dict[str, tuple[list[tuple[str, int]], str]], + sig: str) -> None: + """Swap one file's session facts for a fresh scan. + + `sessions` is session_id -> ([(root, touches), ...], started_by). Not + committed here — rides the caller's per-file transaction.""" + self._delete_session_facts(path) + for session_id, (repos, started_by) in sessions.items(): + for root, touches in repos: + self.db.execute( + "INSERT INTO session_repos(file_path, source, session_id, root, " + "repo, touches) VALUES (?, ?, ?, ?, ?, ?)", + (path, source, session_id, root, + os.path.basename(root.rstrip("/")), touches)) + self.db.execute( + "INSERT INTO session_info(file_path, source, session_id, started_by) " + "VALUES (?, ?, ?, ?)", (path, source, session_id, started_by)) + self.db.execute( + "INSERT INTO repo_scans(path, sig) VALUES (?, ?) " + "ON CONFLICT(path) DO UPDATE SET sig = excluded.sig", (path, sig)) + + def repo_scan_sig(self, path: str) -> "str | None": + row = self.db.execute( + "SELECT sig FROM repo_scans WHERE path = ?", (path,)).fetchone() + return row[0] if row else None + + def session_repos(self, keys) -> dict[tuple[str, str], list[dict]]: + """(source, session_id) -> [{"repo", "path", "touches"}], most-touched + first. A session spread over several files sums its touches.""" + out: dict[tuple[str, str], list[dict]] = {} + for source, session_id in set(keys): + rows = self.db.execute( + "SELECT root, repo, SUM(touches) t FROM session_repos " + "WHERE source = ? AND session_id = ? GROUP BY root " + "ORDER BY t DESC, root", (source, session_id)).fetchall() + out[(source, session_id)] = [ + {"repo": repo, "path": root, "touches": int(t or 0)} + for root, repo, t in rows] + return out + + def started_by(self, source: str, session_id: str) -> "str | None": + row = self.db.execute( + "SELECT started_by FROM session_info WHERE source = ? AND session_id = ? " + "ORDER BY started_by = 'user' LIMIT 1", (source, session_id)).fetchone() + return row[0] if row else None + + def sessions_touching(self, root: str | None = None, + repo: str | None = None) -> set[tuple[str, str]]: + """(source, session_id) of sessions attributed to a repo root or label.""" + if root: + rows = self.db.execute( + "SELECT DISTINCT source, session_id FROM session_repos WHERE root = ?", + (owning_repo(root),)).fetchall() + elif repo: + rows = self.db.execute( + "SELECT DISTINCT source, session_id FROM session_repos " + "WHERE repo = ? COLLATE NOCASE", (repo,)).fetchall() + else: + return set() + return {(s, sid) for s, sid in rows} def prune_deleted(self, source: str | None = None) -> int: """Drop index rows for transcripts that no longer exist on disk. @@ -184,13 +277,22 @@ def prune_deleted(self, source: str | None = None) -> int: @staticmethod def _filters(scope_root: str | None, source: str | None, start_ts: int | None = None, end_ts: int | None = None, - *, alias: str = "c") -> tuple[str, list[object]]: + *, alias: str = "c", + project: str | None = None) -> tuple[str, list[object]]: clauses: list[str] = [] params: list[object] = [] + touched = (f"EXISTS (SELECT 1 FROM session_repos r WHERE r.source = {alias}.source " + f"AND r.session_id = {alias}.session_id AND ") scope, scope_params = scope_clause(f"{alias}.cwd", scope_root) if scope: - clauses.append(scope) + # Launched in the repo, OR worked on it from somewhere else. + clauses.append(f"({scope} OR {touched}r.root = ?))") params.extend(scope_params) + params.append(owning_repo(scope_root)) + if project: + clauses.append(f"({alias}.project = ? COLLATE NOCASE OR " + f"{touched}r.repo = ? COLLATE NOCASE))") + params.extend([project, project]) if source: clauses.append(f"{alias}.source = ?") params.append(source) @@ -208,8 +310,10 @@ def _filters(scope_root: str | None, source: str | None, def knn(self, query_vec: list[float], n: int, scope_root: str | None = None, source: str | None = None, start_ts: int | None = None, - end_ts: int | None = None) -> list[tuple[int, float]]: - clause, params = self._filters(scope_root, source, start_ts, end_ts) + end_ts: int | None = None, + project: str | None = None) -> list[tuple[int, float]]: + clause, params = self._filters(scope_root, source, start_ts, end_ts, + project=project) if not clause: # The IN-subquery is not a filter, it is the orphan guard: a vec row # whose chunk is gone is not a candidate. Scoped queries got this for @@ -236,12 +340,13 @@ def knn(self, query_vec: list[float], n: int, scope_root: str | None = None, def fts(self, query: str, n: int, scope_root: str | None = None, source: str | None = None, start_ts: int | None = None, - end_ts: int | None = None) -> list[int]: + end_ts: int | None = None, project: str | None = None) -> list[int]: terms = [t for t in query.split() if t] if not terms: return [] match = " OR ".join('"' + t.replace('"', '""') + '"' for t in terms) - clause, params = self._filters(scope_root, source, start_ts, end_ts) + clause, params = self._filters(scope_root, source, start_ts, end_ts, + project=project) # ORDER BY rank (bm25, best first) — without it FTS5 returns rowid order # and LIMIT keeps an arbitrary oldest slice instead of the best matches. if not clause: @@ -276,12 +381,13 @@ def get_chunk(self, chunk_id: int) -> "Chunk | None": def recent_sessions(self, scope_root: str | None, limit: int, source: str | None = None, start_ts: int | None = None, - end_ts: int | None = None) -> list[tuple]: + end_ts: int | None = None, + project: str | None = None) -> list[tuple]: """Sessions by most-recent activity (max ts), optionally scoped to a repo. Returns (source, session_id, project, last_ts, turns). Tiebreaks make equal-timestamp ordering deterministic.""" clause, params = self._filters( - scope_root, source, start_ts, end_ts, alias="chunks") + scope_root, source, start_ts, end_ts, alias="chunks", project=project) where = f" WHERE {clause}" if clause else "" return self.db.execute( f"SELECT source, session_id, project, max(ts) AS last_ts, count(*) AS turns " @@ -298,6 +404,21 @@ def first_user_text(self, session_id: str, source: str | None = None) -> str: f"{source_sql} ORDER BY turn_index LIMIT 1", params).fetchone() return row[0] if row else "" + def session_texts(self, session_id: str, source: str, role: str, + newest: bool, limit: int) -> list[str]: + """A session's surface texts of one role, oldest or newest first.""" + order = "ts DESC, turn_index DESC" if newest else "ts, turn_index" + return [r[0] for r in self.db.execute( + "SELECT text FROM chunks WHERE session_id = ? AND source = ? AND role = ? " + f"ORDER BY {order} LIMIT ?", (session_id, source, role, limit)).fetchall()] + + def latest_branch(self, session_id: str, source: str) -> str: + row = self.db.execute( + "SELECT git_branch FROM chunks WHERE session_id = ? AND source = ? " + "AND git_branch IS NOT NULL AND git_branch != '' " + "ORDER BY ts DESC LIMIT 1", (session_id, source)).fetchone() + return row[0] if row else "" + def get_meta(self, key: str) -> str | None: row = self.db.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone() return row[0] if row else None diff --git a/tests/test_hub_http.py b/tests/test_hub_http.py index 260519d..697ed6d 100644 --- a/tests/test_hub_http.py +++ b/tests/test_hub_http.py @@ -162,6 +162,18 @@ def test_search_returns_ranked_anchors(server, hub): assert body["degraded"] is None +def test_search_forwards_project_and_per_session(server, hub, monkeypatch): + # The handler thread holds its own Recall, so record at the class. + seen = [] + original = FakeRecall.recall_search + monkeypatch.setattr(FakeRecall, "recall_search", + lambda self, q, **kw: seen.append(kw) or original(self, q, **kw)) + key = hub.keys.issue("egor") + post(server, "/v1/search", + {"query": "q", "project": "harmonia", "per_session": 0}, key=key) + assert seen[-1]["project"] == "harmonia" and seen[-1]["per_session"] == 0 + + def test_search_without_a_query_is_a_bad_request(server, hub): with pytest.raises(urllib.error.HTTPError) as raised: post(server, "/v1/search", {"k": 3}, key=hub.keys.issue("egor")) diff --git a/tests/test_registrar.py b/tests/test_registrar.py new file mode 100644 index 0000000..ed2714d --- /dev/null +++ b/tests/test_registrar.py @@ -0,0 +1,489 @@ +"""Recall as the memory of a "registrar": an agent that only answers questions +like "what is happening in project X", "who is working on what", "which +project does this belong to". Scenarios are built from real shapes seen in +the history: a session launched in one repo's worktree that edits another +repo, a symlinked alias, desktop side sessions, Codex threads opened by +Claude, a compacted summary as the best hit.""" + +import json +import os +from pathlib import Path + +import pytest + +from session_recall.embed import FakeEmbedder +from session_recall.index import index_corpus +from session_recall.repos import MIN_TOUCHES, RepoResolver, classify_starter +from session_recall.rerank import FakeReranker +from session_recall.retrieve import Recall, diversify +from session_recall.snippets import human_ask, query_snippet +from session_recall.store import Store +import session_recall.server as server + +T0 = 1792900000 # 2026-10-25 + + +def _iso(ts: int) -> str: + import datetime as dt + return dt.datetime.fromtimestamp(ts, dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z") + + +def user(sid, uuid, cwd, ts, text, entrypoint="claude-desktop", branch="main"): + return {"type": "user", "sessionId": sid, "uuid": uuid, "cwd": cwd, + "gitBranch": branch, "timestamp": _iso(ts), "entrypoint": entrypoint, + "message": {"role": "user", "content": text}} + + +def assistant(sid, uuid, cwd, ts, text="", tools=(), branch="main"): + content = [{"type": "text", "text": text}] if text else [] + content += [{"type": "tool_use", "id": f"t{i}", "name": name, "input": inp} + for i, (name, inp) in enumerate(tools)] + return {"type": "assistant", "sessionId": sid, "uuid": uuid, "cwd": cwd, + "gitBranch": branch, "timestamp": _iso(ts), + "message": {"role": "assistant", "content": content}} + + +def write_session(projects: Path, name: str, lines: list[dict]) -> Path: + folder = projects / f"-proj-{name}" + folder.mkdir(parents=True, exist_ok=True) + path = folder / f"{name}.jsonl" + path.write_text("".join(json.dumps(line) + "\n" for line in lines)) + return path + + +@pytest.fixture +def home(tmp_path): + """A home with real git checkouts: two repos, a worktree of one, a + symlinked alias, a plugin cache that is itself a git clone, and a folder + that is not a repo at all.""" + h = tmp_path / "home" + for repo in ("food", "harmonia", "whytap", "neighbour"): + (h / repo / ".git").mkdir(parents=True) + wt = h / "food" / ".claude" / "worktrees" / "wt" + wt.mkdir(parents=True) + (wt / ".git").write_text(f"gitdir: {h / 'food' / '.git' / 'worktrees' / 'wt'}\n") + os.symlink(h / "whytap", h / "exodia") + (h / ".claude" / "plugins" / "cache" / "tool" / ".git").mkdir(parents=True) + (h / "notes").mkdir() + cx = h / ".codex" / "worktrees" / "1a2b" / "whytap" + cx.mkdir(parents=True) + (cx / ".git").write_text(f"gitdir: {h / 'whytap' / '.git' / 'worktrees' / '1a2b'}\n") + return h + + +def _index(tmp_path, home, embedder=None): + store = Store(tmp_path / "r.db") + index_corpus(store, embedder or FakeEmbedder(), tmp_path / "projects", + resolver=RepoResolver(home=str(home))) + return store + + +def _recall(store): + return Recall(store, FakeEmbedder(), FakeReranker()) + + +def _cross_repo_session(home, sid="cross", ts=T0): + """Launched in food's worktree, all real work done in ~/harmonia.""" + wt = str(home / "food" / ".claude" / "worktrees" / "wt") + hm = home / "harmonia" + edits = [("Edit", {"file_path": str(hm / "src" / f"f{i}.ts")}) for i in range(4)] + return [ + user(sid, f"{sid}-u1", wt, ts, "сделай память двух уровней для оркестратора"), + assistant(sid, f"{sid}-a1", wt, ts + 10, "Берусь за память оркестратора.", + tools=edits + [("Bash", {"command": f"cd {hm} && npm test"})]), + assistant(sid, f"{sid}-a2", wt, ts + 20, + "Готово: память двух уровней в оркестраторе harmonia работает."), + ] + + +# --- 1. which repo a session belongs to ------------------------------------ + +def test_session_is_attributed_to_the_repo_it_edited_not_its_launch_folder(tmp_path, home): + write_session(tmp_path / "projects", "cross", _cross_repo_session(home)) + store = _index(tmp_path, home) + repos = store.session_repos([("claude", "cross")])[("claude", "cross")] + assert [r["repo"] for r in repos] == ["harmonia", "food"] + assert repos[0]["path"] == os.path.realpath(home / "harmonia") + assert repos[0]["touches"] == 5 # four Edits + one cd, each call counted once + + +def test_project_filter_finds_cross_repo_work_in_every_tool(tmp_path, home): + write_session(tmp_path / "projects", "cross", _cross_repo_session(home)) + recall = _recall(_index(tmp_path, home)) + + hits = recall.recall_search("память оркестратора", project="harmonia") + assert hits and {h.session_id for h in hits} == {"cross"} + assert hits[0].project == "food" # where it was launched + assert hits[0].repos == ["harmonia", "food"] # what it worked on + + assert recall.recall_search("память оркестратора", project="neighbour") == [] + assert recall.grep("двух уровней", project="harmonia") + assert recall.grep("двух уровней", project="neighbour") == [] + recent = recall.recent_sessions(project="HARMONIA", now=T0 + 30) # case-insensitive + assert [s["session_id"] for s in recent] == ["cross"] + + +def test_scope_cwd_of_the_edited_repo_includes_work_done_from_elsewhere(tmp_path, home): + """The live failure: scope_cwd=~/harmonia returned only unrelated hits, + because the work happened in a session launched elsewhere.""" + write_session(tmp_path / "projects", "cross", _cross_repo_session(home)) + recall = _recall(_index(tmp_path, home)) + scoped = recall.recall_search("память оркестратора", scope_cwd=str(home / "harmonia")) + assert {h.session_id for h in scoped} == {"cross"} + assert recall.grep("двух уровней", scope_cwd=str(home / "harmonia" / "src")) + # The old cwd scope still works too. + assert recall.recall_search("память оркестратора", scope_cwd=str(home / "food")) + + +def test_symlinked_alias_is_one_repo(tmp_path, home): + alias = str(home / "exodia") + write_session(tmp_path / "projects", "alias", [ + user("alias", "al-u", alias, T0, "добавь Meeting Assistant"), + assistant("alias", "al-a", alias, T0 + 5, "Meeting Assistant подключён.", + tools=[("Bash", {"command": "swift build"})]), + ]) + recall = _recall(_index(tmp_path, home)) + assert recall.recall_search("Meeting Assistant", project="whytap")[0].repos == ["whytap"] + # The launch-folder name keeps working for anyone who knows it by that name. + assert recall.recall_search("Meeting Assistant", project="exodia") + assert recall.recall_search("Meeting Assistant", scope_cwd=str(home / "whytap")) + + +def test_a_glance_or_a_plugin_cache_is_not_work_on_a_repo(tmp_path, home): + plugin = home / ".claude" / "plugins" / "cache" / "tool" / "SKILL.md" + glance = [("Read", {"file_path": str(home / "neighbour" / "README.md")})] + reads = [("Read", {"file_path": str(plugin)})] * 5 + tmp = [("Bash", {"command": "ls /tmp/x /private/var/folders/y"})] * 5 + write_session(tmp_path / "projects", "glance", [ + user("glance", "g-u", str(home / "food"), T0, "посмотри"), + assistant("glance", "g-a", str(home / "food"), T0 + 1, "смотрю", + tools=glance * (MIN_TOUCHES - 1) + reads + tmp), + ]) + store = _index(tmp_path, home) + repos = store.session_repos([("claude", "glance")])[("claude", "glance")] + assert [r["repo"] for r in repos] == ["food"] + + +def test_urls_and_odd_paths_never_hang_or_attribute(home): + """Live hang: "https://github.com/x" yields "//github.com/x", and POSIX + dirname("//") == "//" — the walk to the root never ended.""" + resolver = RepoResolver(home=str(home)) + for odd in ("//github.com/AbsoluteMode/harmonia", "//", "/", "~", "relative/x", + "", "/nonexistent/deep/path/file.py"): + assert resolver.repo_of(odd) is None + from session_recall.repos import owning_repo + assert owning_repo("//github.com/x") == os.path.realpath("//github.com/x") + assert owning_repo(str(home / "harmonia" / "src")) == os.path.realpath(home / "harmonia") + + +def test_codex_worktree_in_a_hidden_folder_belongs_to_its_main_checkout(tmp_path, home): + """Live gap: Codex keeps worktrees under ~/.codex/worktrees//; + 44 whytap sessions had no attribution while hidden folders were skipped.""" + cwd = str(home / ".codex" / "worktrees" / "1a2b" / "whytap") + write_session(tmp_path / "projects", "cxwt", [ + user("cxwt", "cw-u", cwd, T0, "почини таймер"), + assistant("cxwt", "cw-a", cwd, T0 + 1, "починил", + tools=[("Edit", {"file_path": cwd + "/Sources/Timer.swift"})]), + ]) + store = _index(tmp_path, home) + assert [r["repo"] for r in store.session_repos( + [("claude", "cxwt")])[("claude", "cxwt")]] == ["whytap"] + + +def test_deleted_codex_worktree_falls_back_to_the_same_named_checkout(tmp_path, home): + gone = str(home / ".codex" / "worktrees" / "9z9z" / "exodia") # alias, deleted + stray = str(home / ".codex" / "worktrees" / "8y8y" / "no-such-repo") + write_session(tmp_path / "projects", "gonecx", [ + user("gonecx", "gc-u", gone, T0, "бэкап конфигов"), + assistant("gonecx", "gc-a", gone, T0 + 1, "ок", + tools=[("Edit", {"file_path": gone + "/a.swift"}), + ("Read", {"file_path": stray + "/b.py"})] * MIN_TOUCHES), + ]) + store = _index(tmp_path, home) + assert [r["repo"] for r in store.session_repos( + [("claude", "gonecx")])[("claude", "gonecx")]] == ["whytap"] + + +def test_deleted_worktree_and_non_repo_folder_still_attribute(tmp_path, home): + gone = str(home / "food" / ".claude" / "worktrees" / "deleted-long-ago") + notes = str(home / "notes") + write_session(tmp_path / "projects", "gone", [ + user("gone", "go-u", gone, T0, "старая ветка"), + assistant("gone", "go-a", gone, T0 + 1, "ок", + tools=[("Edit", {"file_path": gone + "/src/a.py"})]), + ]) + write_session(tmp_path / "projects", "chat", [ + user("chat", "ch-u", notes, T0, "просто поговорить"), + assistant("chat", "ch-a", notes, T0 + 1, "давай"), + ]) + store = _index(tmp_path, home) + facts = store.session_repos([("claude", "gone"), ("claude", "chat")]) + assert [r["repo"] for r in facts[("claude", "gone")]] == ["food"] + # A pure discussion in a non-repo folder belongs to that folder. + assert facts[("claude", "chat")] == [ + {"repo": "notes", "path": notes, "touches": 0}] + + +def test_a_scratch_session_in_hidden_or_library_folder_has_only_real_repos(tmp_path, home): + scratch = str(home / "Library" / "Application Support" / "Claude" / "scratch-1") + hm = str(home / "harmonia") + write_session(tmp_path / "projects", "scratch", [ + user("scratch", "sc-u", scratch, T0, "вспомни harmonia"), + assistant("scratch", "sc-a", scratch, T0 + 1, "вспомнил", + tools=[("Bash", {"command": f"git -C {hm} log"})] * 3), + ]) + store = _index(tmp_path, home) + assert [r["repo"] for r in store.session_repos( + [("claude", "scratch")])[("claude", "scratch")]] == ["harmonia"] + + +def test_codex_workdir_and_exec_code_paths_are_attributed(tmp_path, home): + codex_root = tmp_path / "codex" + path = codex_root / "2026" / "10" / "25" / "rollout-x-01a1.jsonl" + path.parent.mkdir(parents=True) + hm, food = str(home / "harmonia"), str(home / "food") + rows = [ + {"timestamp": _iso(T0), "type": "session_meta", "payload": { + "id": "01a1", "cwd": food, "originator": "Claude Code", "source": "vscode"}}, + {"timestamp": _iso(T0 + 1), "type": "event_msg", "payload": { + "type": "user_message", "message": "почини тесты harmonia"}}, + ] + [ + {"timestamp": _iso(T0 + 2 + i), "type": "response_item", "payload": { + "type": "function_call", "name": "exec_command", "call_id": f"c{i}", + "arguments": json.dumps({"cmd": "npm test", "workdir": hm})}} + for i in range(2) + ] + [ + {"timestamp": _iso(T0 + 5), "type": "response_item", "payload": { + "type": "custom_tool_call", "name": "exec", "call_id": "c9", + "input": f'await tools.exec_command({{cmd:"cat {hm}/package.json"}})'}}, + {"timestamp": _iso(T0 + 6), "type": "event_msg", "payload": { + "type": "agent_message", "message": "Тесты harmonia зелёные."}}, + ] + path.write_text("".join(json.dumps(r) + "\n" for r in rows)) + store = Store(tmp_path / "r.db") + index_corpus(store, FakeEmbedder(), None, (codex_root,), + resolver=RepoResolver(home=str(home))) + recent = _recall(store).recent_sessions(now=T0 + 60) + assert recent[0]["repos"][0] == {"repo": "harmonia", "path": hm, "touches": 3} + assert recent[0]["started_by"] == "agent" # a Codex thread Claude opened + assert recent[0]["source"] == "codex" + + +# --- index lifecycle -------------------------------------------------------- + +class CountingEmbedder(FakeEmbedder): + def __init__(self): + super().__init__() + self.documents = 0 + + def embed_documents(self, texts): + self.documents += len(texts) + return super().embed_documents(texts) + + +def test_an_existing_index_is_backfilled_without_re_embedding(tmp_path, home): + write_session(tmp_path / "projects", "cross", _cross_repo_session(home)) + embedder = CountingEmbedder() + store = _index(tmp_path, home, embedder) + embedded = embedder.documents + # An index built before attribution existed: chunks, no facts. + for table in ("session_repos", "session_info", "repo_scans"): + store.db.execute(f"DELETE FROM {table}") + store.commit() + index_corpus(store, embedder, tmp_path / "projects", + resolver=RepoResolver(home=str(home))) + assert embedder.documents == embedded + assert store.session_repos([("claude", "cross")])[("claude", "cross")][0]["repo"] == "harmonia" + # A further run with nothing new does no work at all. + index_corpus(store, embedder, tmp_path / "projects", + resolver=RepoResolver(home=str(home))) + assert store.db.execute("SELECT COUNT(*) FROM session_info").fetchone()[0] == 1 + + +def test_growing_and_deleted_transcripts_keep_facts_in_step(tmp_path, home): + lines = _cross_repo_session(home) + path = write_session(tmp_path / "projects", "cross", lines) + store = _index(tmp_path, home) + nb = str(home / "neighbour") + lines.append(assistant("cross", "cross-a3", lines[0]["cwd"], T0 + 40, "и сосед", + tools=[("Edit", {"file_path": nb + "/x.py"})] * MIN_TOUCHES)) + path.write_text("".join(json.dumps(line) + "\n" for line in lines)) + os.utime(path, (T0 + 100, T0 + 100)) + index_corpus(store, FakeEmbedder(), tmp_path / "projects", + resolver=RepoResolver(home=str(home))) + repos = store.session_repos([("claude", "cross")])[("claude", "cross")] + assert [r["repo"] for r in repos] == ["harmonia", "neighbour", "food"] + assert store.db.execute("SELECT COUNT(*) FROM session_info").fetchone()[0] == 1 + + path.unlink() + index_corpus(store, FakeEmbedder(), tmp_path / "projects", + resolver=RepoResolver(home=str(home))) + for table in ("session_repos", "session_info", "repo_scans"): + assert store.db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0 + + +# --- 2. who is doing what --------------------------------------------------- + +def test_recent_sessions_reads_like_a_status_board(tmp_path, home): + food = str(home / "food") + write_session(tmp_path / "projects", "side", [ + user("side", "sd-u1", food, T0, + "worktree" + "Background (from the session that started this one):\nМаксим строит центр.\n\n" + "Task:\nДоработай recall для регистратора", branch="claude/side"), + assistant("side", "sd-a1", food, T0 + 5, "Начинаю с аудита.", branch="claude/side"), + user("side", "sd-u2", food, T0 + 50, + "None of the sessions is working", + branch="claude/side"), + user("side", "sd-u3", food, T0 + 60, + 'лог ошибки\nпочини это', + branch="claude/side"), + assistant("side", "sd-a2", food, T0 + 70, "Починил, PR открыт.", branch="claude/side"), + ]) + write_session(tmp_path / "projects", "sdk", [ + user("sdk", "sk-u", food, T0 - 5000, "summarize this", entrypoint="sdk-cli"), + assistant("sdk", "sk-a", food, T0 - 4990, "summary"), + ]) + recall = _recall(_index(tmp_path, home)) + board = {s["session_id"]: s for s in recall.recent_sessions(now=T0 + 100)} + + side = board["side"] + assert side["label"] == "Доработай recall для регистратора" + assert side["last_user"].startswith("лог ошибки") + assert "почини это" in side["last_user"] + assert side["last_assistant"] == "Починил, PR открыт." + assert side["started_by"] == "agent" + assert side["git_branch"] == "claude/side" + assert side["active"] is True + assert side["repos"][0]["repo"] == "food" + + assert board["sdk"]["started_by"] == "agent" + assert board["sdk"]["active"] is False + + +def test_recent_sessions_label_falls_back_when_every_turn_is_machinery(tmp_path, home): + food = str(home / "food") + write_session(tmp_path / "projects", "beat", [ + user("beat", "b-u", food, T0, "w"), + assistant("beat", "b-a", food, T0 + 1, "NOTIFY"), + ]) + recall = _recall(_index(tmp_path, home)) + entry = recall.recent_sessions(now=T0 + 10)[0] + assert entry["started_by"] == "automation" + assert entry["label"].startswith("") # nothing better exists + assert entry["last_user"] == "" + + +def test_classify_starter_edges(): + assert classify_starter("claude", "", "", "", "") == "user" + assert classify_starter("claude", "cli", "", "", " x") == "automation" + assert classify_starter("codex", "", "codex_exec", "exec", "do x") == "agent" + assert classify_starter("codex", "", "Codex Desktop", "vscode", "do x") == "user" + + +# --- 3. diverse results ----------------------------------------------------- + +def test_one_long_session_does_not_take_every_slot(tmp_path, home): + food = str(home / "food") + long_lines = [user("long", "l-u", food, T0, "оркестратор")] + long_lines += [assistant("long", f"l-a{i}", food, T0 + i, f"оркестратор шаг {i}") + for i in range(1, 7)] + write_session(tmp_path / "projects", "long", long_lines) + write_session(tmp_path / "projects", "other", [ + user("other", "o-u", food, T0, "оркестратор где мы"), + assistant("other", "o-a", food, T0 + 1, "оркестратор: итог решения"), + ]) + recall = _recall(_index(tmp_path, home)) + hits = recall.recall_search("оркестратор", k=4) + by_session = [h.session_id for h in hits] + assert by_session.count("long") == 2 and by_session.count("other") == 2 + uncapped = recall.recall_search("оркестратор", k=4, per_session=0) + assert len(uncapped) == 4 + # Narrow history: fewer sessions than slots -> back-filled, never short. + assert len(recall.recall_search("шаг", k=5)) == 5 + + +def test_diversify_keeps_rank_order_and_backfills(): + items = [("a", 1), ("a", 2), ("a", 3), ("b", 4), ("a", 5)] + assert diversify(items, 3, 2, key=lambda x: x[0]) == [("a", 1), ("a", 2), ("b", 4)] + assert diversify(items, 4, 1, key=lambda x: x[0]) == [ + ("a", 1), ("a", 2), ("a", 3), ("b", 4)] + assert diversify([], 3, 2, key=lambda x: x) == [] + assert diversify(items, 2, 0, key=lambda x: x[0]) == items[:2] + + +# --- 4. snippets around the query ------------------------------------------- + +def test_snippet_centres_on_the_query_and_drops_compaction_preamble(): + summary = ("This session is being continued from a previous conversation that ran " + "out of context. The summary below covers the earlier portion of the " + "conversation.\n\nSummary:\n1. Primary Request: " + "контекст " * 60 + + "решили оставить хаб выключенным до проверки ключа" + " хвост" * 40) + snip = query_snippet(summary, "почему хаб выключен") + assert "хаб выключенным" in snip + assert "This session is being continued" not in snip + assert snip.startswith("…") and snip.endswith("…") + assert len(snip) <= 202 + + +def test_snippet_without_literal_match_starts_at_the_beginning(): + text = "Primary answer first. " + "x" * 400 + assert query_snippet(text, "семантический запрос").startswith("Primary answer first.") + assert query_snippet("short", "") == "short" + assert query_snippet("", "q") == "" + + +def test_search_snippet_shows_the_matching_passage(tmp_path, home): + food = str(home / "food") + text = "Длинный отчёт. " + "детали " * 80 + "Ключ Voyage на сервере невалиден." + " ещё" * 50 + write_session(tmp_path / "projects", "rep", [ + user("rep", "r-u", food, T0, "что с хабом"), + assistant("rep", "r-a", food, T0 + 1, text), + ]) + recall = _recall(_index(tmp_path, home)) + hit = next(h for h in recall.recall_search("ключ Voyage невалиден", k=5) + if h.uuid == "r-a") + assert "Voyage" in hit.snippet + + +def test_human_ask_edges(): + assert human_ask("") is None + assert human_ask("x") is None + assert human_ask("tick") is None + assert human_ask("This session is being continued from a previous conversation.\n" + "\nSummary:\nstuff") is None + assert human_ask('текст') == "текст" + brief = "Background (from the session that started this one):\nфон\n\nTask:\nсделай X" + assert human_ask(brief) == "сделай X" + assert human_ask("Background (from the session that started this one): без задачи") \ + .startswith("Background") + # An unclosed tag is the user quoting it, not machinery. + assert human_ask("что значит ?") == "что значит ?" + + +# --- tool boundary ---------------------------------------------------------- + +def test_mcp_tools_forward_project_and_drop_empty_repos(tmp_path, home, monkeypatch): + write_session(tmp_path / "projects", "cross", _cross_repo_session(home)) + monkeypatch.setattr(server, "_recall", _recall(_index(tmp_path, home))) + anchors = server.recall_search("память оркестратора", project="harmonia")["anchors"] + assert anchors and anchors[0]["repos"] == ["harmonia", "food"] + assert server.grep("двух уровней", project="harmonia") + assert server.recent_sessions(project="harmonia")[0]["session_id"] == "cross" + from session_recall.models import Anchor + bare = server._adict(Anchor(session_id="s", uuid="u", role="doc", snippet="", + score=None, project="p", when=0, source="metadocs")) + assert "repos" not in bare + + +def test_hub_client_sends_project_to_the_server(monkeypatch): + from session_recall.hub.client import HubConfig + from session_recall.hub.remote import RemoteRecall + sent = [] + remote = RemoteRecall(HubConfig(url="http://hub", key="k")) + monkeypatch.setattr(remote, "_post", lambda path, body: sent.append((path, body)) or {}) + remote.recall_search("q", project="harmonia", per_session=0) + remote.grep("p", project="harmonia") + remote.recent_sessions(project="harmonia") + assert [b["project"] for _, b in sent] == ["harmonia"] * 3 + assert sent[0][1]["per_session"] == 0