diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index e9814a0..ef5e6f7 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "session-recall", - "version": "0.5.0", + "version": "0.6.0", "description": "Shared semantic recall over local Claude Code, Codex, and Cursor history. Bundles MCP search tools, deep recall, guided setup, and a background freshness hook.", "author": { "name": "Max Butorin" } } diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index b95d58f..44d27b8 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "session-recall", - "version": "0.5.0", + "version": "0.6.0", "description": "Shared semantic recall over local Claude Code, Codex, and Cursor history, with durable deep navigation and guided setup.", "author": { "name": "Max Butorin" diff --git a/.cursor-plugin/plugin.json b/.cursor-plugin/plugin.json index c53dc77..1aadae5 100644 --- a/.cursor-plugin/plugin.json +++ b/.cursor-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "session-recall", "displayName": "Session Recall", - "version": "0.5.0", + "version": "0.6.0", "minClientVersions": { "cursor": "2.5.0" }, diff --git a/docs/decisions/2026-08-06-team-hub-central-index.md b/docs/decisions/2026-08-06-team-hub-central-index.md new file mode 100644 index 0000000..e03f763 --- /dev/null +++ b/docs/decisions/2026-08-06-team-hub-central-index.md @@ -0,0 +1,162 @@ +# Командный хаб: центральный индекс вместо local-first + +## Контекст + +У Максима выросла команда мидлов, которые работают через агентов. Задача, +сформулированная им прямо: «мне стало трудно за ними следить, мне легче в их +историю глянуть». Плюс вторая, обратная по направлению: любой участник должен +мочь спросить «а как ты делал то или это» и получить ответ по истории команды, +а не дёргать человека в мессенджере. + +Это ломает заявленный инвариант проекта. README говорит, что индекс и история +живут локально, а весь P2P-share (`2026-07-30-p2p-sharing-v1-security-gate.md`) +построен именно вокруг того, чтобы **не** централизовать: там ответ собирает +воркер на машине владельца, а наружу уходит только явно одобренный текст. + +Развилка обсуждалась 25.07.2026 (сессия `07bc6547`), но тогда свернула в +сторону egress-туннеля: выяснилось, что Voyage отбивает домашний IP, и вопрос +«перенести ли сервис на сервер» подменился вопросом «как достучаться до +Voyage». Сейчас вернулись к исходному. + +## Решение + +Отдельный режим — **хаб**, включается наличием клиентского конфига. Без него +всё работает ровно как раньше (репозиторий публичный, solo-путь ломать +нельзя). + +Три выбора, сделанные Максимом явно: + +- **Общий пул.** Все ищут по истории всех. Не «своё + расшаренное». +- **Уезжает всё**, а не выбранные проекты. +- **Только сервер.** Локальный индекс в хаб-режиме не ведётся. + +Ключевое архитектурное решение: **хаб — это обычный session-recall, у которого +транскрипты приходят по сети, а не от агента.** Сервер складывает присланное в +ту же раскладку каталогов, которая есть на ноутбуке +(`/claude//.jsonl`), и запускает по ней штатный +`index_corpus`. Никакого серверного экстрактора, второго формата хранения и +параллельного пути поиска. + +Транспорт — stdlib `http.server` и `urllib`, как в relay, а не MCP-over-HTTP: +ключ живёт в клиентском файле `0600`, а не в конфиге агента; плагин у +сотрудника ставится без изменений; новых зависимостей ноль. + +Маскировка секретов — **два слоя, ловящие разное**: + +1. на клиенте, до отправки — регэкспы форматов (`share/scanner.redact`); +2. на сервере, до записи на диск — точное совпадение со значениями из Doppler, + замена на имя переменной: `${servers/NETCUP_PASSWORD}`. + +Сервер хранит **хеши** значений под солью, а не значения. + +Ответы на вопросы (`/v1/ask`) пишет Codex по подписке (`codex exec`), модель +`gpt-5.6-terra`. + +## Почему + +- **Регэкспы и Doppler ловят разные вещи.** `sk-ant-…` узнаётся по форме; + пароль формы не имеет и ловится только точным совпадением. Ни один слой не + покрывает второй, поэтому их два. +- **Хеши вместо значений.** Иначе `claude-recall` становится копией всего + Doppler в памяти процесса, и его компрометация отдаёт разом все секреты + команды — ровно то, чего централизация и так добавляет риска. +- **Фильтр по длине и энтропии обязателен.** В Doppler рядом с ключами лежит + `DOPPLER_CONFIG=dev`. Слепая замена всех значений вырезала бы слово «dev» из + истории всей команды. На живом прогоне из 650 секретов маскируются 348. +- **Учёт принятых байт отдельно от размера файла.** Маскировка меняет длину, и + если считать по размеру файла на диске, докачка хвостов разъедется на первом + же замаскированном ключе. +- **`owner` берётся из ключа, а не из тела запроса.** Иначе любой участник + может писать в историю другого. +- **`--ephemeral` у Codex — не гигиена, а безопасность.** Хаб индексирует то, + что хранит; сохранённая сессия воркера положила бы вопрос коллеги в общий + индекс, откуда он позже читается как «наша прошлая работа» — stored-инъекция, + описанная в гейте P2P. +- **Промпт — второй слой, не первый.** Правило «не разглашай личное» задано + промптом по просьбе Максима, но нагрузку несёт маскировка: вырезанного текста + модель не разгласит в принципе, а инструкцию можно обойти. +- **Согласие сотрудника — не формальность.** Скрытый мониторинг это и правовая + проблема, и практическая: он вскроется и обойдётся дороже открытого. Экран + согласия в `hub join` стоит одну команду. + +## Что протестировали + +- **Voyage с Netcup напрямую** — `http=200`, 265 мс. Значит серверу не нужен + SOCKS-туннель, а сотрудникам не нужен ключ Voyage вообще. +- **Раскладка «как на ноутбуке» реально индексируется штатным кодом** — залили + настоящий транскрипт по HTTP, `hub index` его переварил, поиск нашёл. +- **Маскировка на живом Doppler** — 650 секретов прочитано, 348 маскируется; + реальные `VOYAGE_API_KEY` и `NETCUP_PASSWORD` заменились на имена переменных, + в карте их нет. +- **Общий пул** — второй сотрудник нашёл историю первого. +- **Инкрементальность** — повторный `push` отправляет ноль байт. +- **Подписка, а не API** — с заведомо битым `OPENAI_API_KEY` и несуществующим + `OPENAI_BASE_URL` композер всё равно ответил. +- **`--ephemeral`** — сессий Codex на диске было 1344 и осталось 1344. +- **`ask` целиком** — вопрос → поиск по чужой истории → written-ответ с точной + переменной окружения и честным «фрагменты не отвечают на…». + +Два бага поймали тесты до деплоя: валидатор путей отбивал ведущий дефис и +зарубил бы **каждый** claude-транскрипт (`-Users-egor-proj`), а `sk-ant-…` +помечался как `[REDACTED:openai-key]`, потому что паттерн OpenAI шире и стоял +раньше. + +## Чего тесты поймать не могли (правки по итогам первого боевого корпуса) + +Первая настоящая заливка 3.5 ГБ вскрыла три вещи, невидимые на фикстурах. + +**Скорость.** 256 КБ на чанк давали ~0.1 МБ/с: узкое место не байты, а +одно TLS-рукопожатие и один fsync на чанк, потому что `urllib` не +переиспользует соединение. 4 МБ → ~60 МБ/мин, история укладывается в час +вместо девяти. + +**Маскировка резала конфигурацию.** Отбор по длине и энтропии не отличает +`claude-opus-4` от токена — обе строки короткие и разноклассовые. По корпусу +замаскировались 851 упоминание имени модели, 2429 hostname и 1536 IP: искать +«какую модель брали» стало невозможно, читать историю — тяжело. Решает **имя** +переменной, а не форма значения: `PASSWORD|TOKEN|SECRET|KEY|CREDENTIAL` — креды, +`*_MODEL|*_HOSTNAME|*_IP|*_REGION|*_BUCKET` — конфигурация. Из 650 секретов +Doppler маскируется 200 вместо 348. + +**Авто-редакт съедал текст.** `password-assignment` +(`\w*(token|secret|key)\s*[=:]\s*\S{8,}`) сработал 27324 раза против ~1900 у +всех настоящих форматов ключей вместе взятых: он ловит любое поле JSON и любую +фразу вида «token: …». Для `scan` это нормально — находку читает человек и +ложное срабатывание бесплатно. Для автоматического вырезания оно удаляет +работу, а не секреты, поэтому в `redact` попадает только явный allowlist +форматов. + +Общий урок: **у флагующего и у режущего детектора разная цена ошибки**, и один +набор паттернов не может обслуживать оба. Корпус после этих правок перезалит +заново. + +## Отвергли + +- **Своё + расшаренные проекты** — выбор Максима в пользу общего пула; следить + за командой иначе не выходит. +- **Гибрид «локальный + серверный индекс» со слиянием выдачи** — два разных + эмбеддера дают несравнимые скоры, склейка получилась бы эвристикой. +- **MCP-over-HTTP напрямую из агента** — ключ уехал бы в конфиг агента, плагин + пришлось бы переделывать, а упавший сервер давал бы агенту транспортную + ошибку вместо внятной фразы. +- **Postgres + pgvector** — на текущем объёме (36k чанков на человека) SQLite с + sqlite-vec справляется; переезжать, когда упрёмся. +- **Маскировка хранением значений** — быстрее и ловит секреты с разделителями, + но превращает сервис в хранилище всех секретов команды. +- **Anthropic API для ответов** — Максим выбрал подписку Codex; ключ не нужен. + +## Открытые вопросы + +- `codex exec --sandbox read-only` не запрещает чтение файлов: инъекция во + фрагменте теоретически может попросить модель прочитать что-то с сервера. + Ограничение остаётся на уровне unix-пользователя сервиса. +- `project` в выдаче берётся из `cwd` транскрипта, поэтому для сессий во + временных каталогах получается мусорная метка. Поведение унаследовано от + solo-режима. +- Удаление истории уволившегося: отзыв ключа закрывает доступ, но данные + остаются. Команды `hub purge ` пока нет. + +--- + +06.08.2026 · реализация в ветке `claude/session-recall-server-6171d6`, +модуль `src/session_recall/hub/`. Инструкция — `docs/team-hub.ru.md`. diff --git a/docs/team-hub.ru.md b/docs/team-hub.ru.md new file mode 100644 index 0000000..e56ab82 --- /dev/null +++ b/docs/team-hub.ru.md @@ -0,0 +1,181 @@ +# claude-recall — подключение к общему индексу команды + +Общая память команды: транскрипты сессий Claude Code и Codex уезжают на наш +сервер, там векторизуются и складываются в один индекс. Дальше любой участник +может спросить — «как это чинили», «почему выбрали такой вариант» — и получить +ответ по всей истории команды, а не только по своей. + +--- + +## Часть 1. Для сотрудника + +### Что нужно знать до подключения + +- **Твои сессии станут видны всей команде.** Промпты, ответы, вызовы + инструментов и их вывод — всё это попадает в общий поиск. +- **Ключи и токены вырезаются автоматически**, в два слоя: на твоей машине + режутся значения известных форматов (`sk-ant-…`, `ghp_…`, PEM-блоки), а на + сервере дополнительно маскируется всё, что команда хранит в Doppler — + например пароль от сервера заменится на `${servers/NETCUP_PASSWORD}`. +- **Это растяжка, а не гарантия.** Пароль, придуманный на ходу и никуда не + записанный, не поймает ни один из слоёв. Личное в рабочие сессии писать не + стоит — как и раньше. +- **Отключиться можно одной командой**, но уже загруженное останется на + сервере: удалить его может только оператор. + +### Шаг 1. Установка + +```bash +pipx install git+https://github.com/AbsoluteMode/session-recall +``` + +Плагин для агента (даёт ему инструменты поиска и автообновление): + +```bash +/plugin marketplace add AbsoluteMode/session-recall +/plugin install session-recall +``` + +В Cursor — добавь репозиторий как marketplace и выполни +`/add-plugin session-recall`. + +### Шаг 2. Подключение + +Ключ тебе выдаёт оператор — он выглядит как `sr_имя_32симвода`. Ключ личный, +по нему видно, чьи сессии приходят. + +```bash +session-recall hub join https://recall.terra-sandbox.ru sr_твоёимя_… +``` + +Команда покажет, что именно будет отправляться, и спросит подтверждение. +После согласия ключ ложится в `~/.local/share/session-recall/hub.json` +с правами `600`. + +### Шаг 3. Первая заливка + +```bash +session-recall hub push +``` + +Первый раз уезжает вся история. Скорость — около **60 МБ в минуту**, то есть +несколько гигабайт займут примерно час; можно запустить и заниматься своими +делами. Дальше всё инкрементально: отправляются только новые байты, обычная +дозаливка — это килобайты и доли секунды. + +Заливка возобновляемая: если прервать на середине (или закрыть ноутбук), +следующий запуск продолжит с того же места, а не начнёт заново. + +### Шаг 4. Проверка + +```bash +session-recall hub status +``` + +Дальше ничего делать не нужно: при старте каждой сессии агента новые +транскрипты уходят на сервер сами. + +### Как этим пользоваться + +Просто спрашивай агента про прошлую работу — свою или чужую: + +> помнишь, как мы чинили падение CI на сборке тестов? + +> как в команде поднимали relay — кто-то это уже делал? + +Агент сам сходит в общий индекс. Плюс есть прямой вопрос без агента: + +```bash +session-recall hub ask "как настраивали nginx под wildcard-сертификат?" +``` + +### Если что-то не так + +| Симптом | Что делать | +|---|---| +| `hub rejected the key` | Ключ отозван или скопирован с опечаткой — попроси новый | +| `hub unreachable` | Сервер недоступен; проверь VPN и что URL с `https://` | +| `skip (unsupported name)` при заливке | Каталог проекта с необычным именем (например с пробелами) — пропускается намеренно, остальное льётся | +| Нужно временно перестать отправлять | `session-recall hub leave` | + +### Как отключиться + +```bash +session-recall hub leave +``` + +Машина перестаёт что-либо отправлять. Уже загруженное остаётся на сервере — +если нужно удалить, скажи оператору. + +--- + +## Часть 2. Для оператора + +### Выдать ключ новому сотруднику + +На сервере: + +```bash +session-recall hub key issue egor --note "ноутбук Егора" +``` + +Ключ печатается **один раз** — восстановить его нельзя, только выпустить +новый. Передавай тем же каналом, каким передаёшь остальные доступы. + +### Посмотреть, кто подключён + +```bash +session-recall hub key list +``` + +Показывает имя, короткий id ключа, пометку и статус. Самого ключа там нет — +на сервере хранится только его хеш. + +### Отозвать доступ + +```bash +session-recall hub key revoke egor # все устройства сотрудника разом +session-recall hub key revoke a3f91c2b # одно устройство по id +``` + +Действует немедленно: следующий же запрос получит `401`. Загруженная история +при этом остаётся в индексе — отзыв закрывает доступ, а не стирает данные. + +### Маскировка секретов + +Карта строится из Doppler и обновляется по таймеру: + +```bash +session-recall hub secrets refresh # пересобрать из Doppler +session-recall hub secrets status # сколько секретов маскируется и когда обновлялось +``` + +Маскируются не все значения подряд: короткие и низкоэнтропийные пропускаются +намеренно. Иначе `DOPPLER_CONFIG=dev` вырезал бы слово «dev» из истории всей +команды. + +На сервере хранятся **хеши** значений, а не сами значения: утечка масккарты +не отдаёт ничьи секреты. + +### Индексация + +```bash +session-recall hub index +``` + +Запускается по таймеру. Идёт тем же кодом, что и локальная индексация, — по +одному проходу на владельца. Один сбойный транскрипт не останавливает +остальных. + +--- + +## Что где лежит + +| | Путь | +|---|---| +| Конфиг сотрудника (URL + ключ, `600`) | `~/.local/share/session-recall/hub.json` | +| Данные сервера | `/var/lib/claude-recall/` | +| Сырые транскрипты на сервере | `/var/lib/claude-recall/transcripts/<кто>/` | +| Индекс | `/var/lib/claude-recall/index.db` | +| Ключи (только хеши) | `/var/lib/claude-recall/keys.json` | +| Масккарта (только хеши) | `/var/lib/claude-recall/secrets.json` | diff --git a/hooks/hooks-cursor.json b/hooks/hooks-cursor.json index 3483bda..710df55 100644 --- a/hooks/hooks-cursor.json +++ b/hooks/hooks-cursor.json @@ -3,7 +3,7 @@ "hooks": { "sessionStart": [ { - "command": "export PATH=\"$HOME/.local/bin:$HOME/bin:$PATH\"; sr=session-recall; pgrep -f \"$sr index\" >/dev/null 2>&1 || { command -v \"$sr\" >/dev/null 2>&1 && (\"$sr\" index >/tmp/session-recall-index.log 2>&1 &) || echo 'session-recall: CLI not on PATH - index not refreshed (see README: Keeping the index fresh)' >>/tmp/session-recall-index.log; }" + "command": "export PATH=\"$HOME/.local/bin:$HOME/bin:$PATH\"; sr=session-recall; pgrep -f \"$sr sync\" >/dev/null 2>&1 || { command -v \"$sr\" >/dev/null 2>&1 && (\"$sr\" sync >/tmp/session-recall-index.log 2>&1 &) || echo 'session-recall: CLI not on PATH - history not synced (see README: Keeping the index fresh)' >>/tmp/session-recall-index.log; }" } ] } diff --git a/hooks/hooks.json b/hooks/hooks.json index 3243073..0385243 100644 --- a/hooks/hooks.json +++ b/hooks/hooks.json @@ -7,7 +7,7 @@ { "type": "command", "timeout": 10, - "command": "export PATH=\"$HOME/.local/bin:$HOME/bin:$PATH\"; sr=session-recall; pgrep -f \"$sr index\" >/dev/null 2>&1 || { command -v \"$sr\" >/dev/null 2>&1 && (\"$sr\" index >/tmp/session-recall-index.log 2>&1 &) || echo 'session-recall: CLI not on PATH - index not refreshed (see README: Keeping the index fresh)' >>/tmp/session-recall-index.log; }" + "command": "export PATH=\"$HOME/.local/bin:$HOME/bin:$PATH\"; sr=session-recall; pgrep -f \"$sr sync\" >/dev/null 2>&1 || { command -v \"$sr\" >/dev/null 2>&1 && (\"$sr\" sync >/tmp/session-recall-index.log 2>&1 &) || echo 'session-recall: CLI not on PATH - history not synced (see README: Keeping the index fresh)' >>/tmp/session-recall-index.log; }" } ] } diff --git a/pyproject.toml b/pyproject.toml index dd6a41e..9552eb3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "session-recall" -version = "0.5.0" +version = "0.6.0" description = "Semantic recall over your local Claude Code, Codex, and Cursor history — one shared index, searchable by meaning" readme = "README.md" requires-python = ">=3.11" diff --git a/src/session_recall/cli.py b/src/session_recall/cli.py index 14ac83f..c73281f 100644 --- a/src/session_recall/cli.py +++ b/src/session_recall/cli.py @@ -1,4 +1,5 @@ import argparse +import sys from pathlib import Path from . import config from .store import Store, corpus_summary @@ -112,16 +113,39 @@ def main(argv=None): pp = sub.add_parser("prune") # drop rows for transcripts deleted from disk pp.add_argument("--source", choices=("claude", "codex", "cursor")) sub.add_parser("health") # is recall actually working right now? + # What the SessionStart hook calls. One command whose meaning follows the + # install: keep the local index fresh, or push to the team hub. The hook + # must not have to know which mode this machine is in. + sub.add_parser("sync") from .share import cli as share_cli share_cli.add_parser(sub) from .metadocs import cli as metadocs_cli metadocs_cli.add_parser(sub) + from .hub import cli as hub_cli + hub_cli.add_parser(sub) from . import onboarding onboarding.add_parser(sub) args = parser.parse_args(argv) if args.cmd == "share": # no Store: share state lives outside the index DB return share_cli.run(args) + if args.cmd == "hub": # own data directory, never the local index + return hub_cli.run(args) + if args.cmd == "sync": + from .hub.client import HubConfig, HubError, push + cfg = HubConfig.load() + if cfg is None: + # Solo install: sync IS the old index run, unchanged. Re-entering + # main() keeps that one behaviour defined in exactly one place. + return main(["index"]) + try: + stats = push(cfg) + except HubError as failure: + print(f"session-recall: hub push failed: {failure}", file=sys.stderr) + return 1 + print(f"pushed {stats['files']} transcript(s), " + f"{stats['uploaded_bytes']} B, {stats['redacted']} secret(s) redacted") + return 1 if stats["failed"] else 0 if args.cmd == "metadocs": # reads the index read-only via plain sqlite3 return metadocs_cli.run(args) if args.cmd == "setup": # indexes via a child process (fresh env resolve) diff --git a/src/session_recall/hub/__init__.py b/src/session_recall/hub/__init__.py new file mode 100644 index 0000000..65641f3 --- /dev/null +++ b/src/session_recall/hub/__init__.py @@ -0,0 +1,14 @@ +"""Team hub — one shared index, fed by every member's raw transcripts. + +The solo product is local-first by design: the index never leaves the machine. +A hub is the deliberate opposite, for a team that has agreed to pool its +history on a server it owns. Nothing here activates by accident — a client +without hub configuration behaves exactly as before. + +The shape of the server is chosen to keep the two modes from diverging: a hub +stores the raw transcripts its clients send, in the same directory layout they +have locally, and then runs the ORDINARY indexer over them. There is no +server-side extractor, no second storage format, and no parallel search path +to keep in sync — `extract`, `index`, `retrieve` and `grep` are the same code +in both modes. +""" diff --git a/src/session_recall/hub/app.py b/src/session_recall/hub/app.py new file mode 100644 index 0000000..3b31a6d --- /dev/null +++ b/src/session_recall/hub/app.py @@ -0,0 +1,316 @@ +"""The hub's HTTP surface. + +stdlib http.server, like the share relay: the route table is small and a +framework would buy nothing here. TLS, timeouts and the public name are +nginx's job — this listens on localhost. + +Routes: + GET /healthz liveness, unauthenticated, says nothing about content + POST /v1/ingest/manifest what we already hold for the caller + POST /v1/ingest append (or replace) one transcript's bytes + POST /v1/search semantic recall over the pooled index + +Everything except /healthz requires `Authorization: Bearer `, and the +owner comes from the key — never from the request body. A client cannot write +into, or claim to be, another member's history. + +Concurrency shapes two decisions here. Requests are threaded, so each thread +gets its OWN sqlite connection (a connection may not cross threads), and the +indexer writes to the same database while searches read it, so the hub opens +in WAL mode with a busy timeout instead of serializing everything behind one +writer. +""" + +import base64 +import gzip +import json +import threading +import time +from dataclasses import asdict +from http.server import BaseHTTPRequestHandler +from pathlib import Path + +from ..store import Store +from ..embed import make_embedder +from ..rerank import make_reranker +from ..retrieve import Recall +from ..timefmt import date_range_to_epoch, humanize_ts +from . import ask as hub_ask +from . import storage +from .auth import KeyStore +from .masking import SecretMap + +MAX_BODY = 16 * 1024 * 1024 # one ingest chunk plus base64 overhead +_BUSY_TIMEOUT_MS = 10_000 +_UNSET = object() # "not built yet", distinct from "no composer" + + +class Hub: + """Server-side state: where the data lives and how to search it.""" + + def __init__(self, root: Path, keys: KeyStore | None = None, + recall_factory=None, composer=_UNSET): + self._composer = composer + self.root = Path(root) + self.transcripts = self.root / "transcripts" + self.db_path = self.root / "index.db" + self.keys = keys or KeyStore(self.root / "keys.json") + self.ledger = storage.Ledger(self.root / "state") + self.secrets_path = self.root / "secrets.json" + self._recall_factory = recall_factory or self._build_recall + self._local = threading.local() + self._secret_map: SecretMap | None = None + self._secret_mtime = -1.0 + self._secret_lock = threading.Lock() + + @property + def composer(self): + """The model that writes `ask` answers, or None for the digest path. + + Built once and cached: `make_composer` only reads configuration, but + the hub's privacy rule has to be attached at construction, and doing + that per request would be a silent invitation to forget it. + """ + if self._composer is _UNSET: + from ..share.compose import make_composer + self._composer = make_composer(system_extra=hub_ask.PRIVACY_RULE) + return self._composer + + @property + def secret_map(self) -> SecretMap: + """The masking map, reloaded whenever the file changes. + + `hub secrets refresh` runs on a timer in a separate process, and a map + that only loaded at boot would quietly stop covering credentials added + since — the failure mode being "the new API key is in the index". + """ + try: + mtime = self.secrets_path.stat().st_mtime + except OSError: + mtime = 0.0 + with self._secret_lock: + if self._secret_map is None or mtime != self._secret_mtime: + self._secret_map = SecretMap.load(self.secrets_path) + self._secret_mtime = mtime + return self._secret_map + + def _build_recall(self) -> Recall: + # The service is a READER, never a writer: the indexer owns writes. + # Anything else and sqlite-vec's shadow tables lose the indexer's + # writes — measured, not theoretical (see Store.__init__). + if not self.db_path.exists(): + # A hub with nothing indexed yet has no file to open read-only. + # Create the schema once, then drop the writable handle. + Store(self.db_path).close() + store = Store(self.db_path, read_only=True) + store.db.execute(f"PRAGMA busy_timeout={_BUSY_TIMEOUT_MS}") + return Recall(store, make_embedder(), make_reranker()) + + @property + def recall(self) -> Recall: + """One Recall per thread — sqlite connections are not shareable.""" + found = getattr(self._local, "recall", None) + if found is None: + found = self._local.recall = self._recall_factory() + return found + + +def _anchor(a) -> dict: + d = asdict(a) + d["when_human"] = humanize_ts(a.when, int(time.time())) + return d + + +def _range(body: dict) -> tuple[int | None, int | None]: + """Time bounds from either form. + + A human caller sends calendar dates and a timezone; the MCP proxy has + already resolved those on the client (where the user's timezone actually + is) and sends epochs. Accepting both keeps the resolution in one place per + caller instead of guessing a server-side timezone. + """ + if body.get("start_ts") is not None or body.get("end_ts") is not None: + return body.get("start_ts"), body.get("end_ts") + return date_range_to_epoch( + body.get("start_date"), body.get("end_date"), + body.get("timezone"), on_date=body.get("on_date")) + + +def make_handler(hub: Hub): + class Handler(BaseHTTPRequestHandler): + server_version = "claude-recall" + sys_version = "" + protocol_version = "HTTP/1.1" + + # --- plumbing ------------------------------------------------- + + def log_message(self, fmt, *args): + # Method, status and owner only. Request paths carry project and + # session names — team history metadata that does not belong in a + # log that outlives the request. + owner = getattr(self, "_owner", "-") + print(f"{self.command} {self.path.split('?')[0][:40]} " + f"owner={owner} {args[1] if len(args) > 1 else ''}".strip(), + flush=True) + + def _send(self, code: int, payload: dict) -> None: + body = json.dumps(payload).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _body(self) -> dict | None: + try: + length = int(self.headers.get("Content-Length") or 0) + except ValueError: + return None + if length <= 0 or length > MAX_BODY: + return None + raw = self.rfile.read(length) + if (self.headers.get("Content-Encoding") or "").lower() == "gzip": + try: + raw = gzip.decompress(raw) + except (OSError, EOFError, gzip.BadGzipFile): + return None + try: + parsed = json.loads(raw) + except ValueError: + return None + return parsed if isinstance(parsed, dict) else None + + def _auth(self) -> str | None: + owner = hub.keys.verify(self.headers.get("Authorization")) + self._owner = owner or "-" + return owner + + # --- routes --------------------------------------------------- + + def do_GET(self): + if self.path == "/healthz": + return self._send(200, {"ok": True}) + self._send(404, {"error": "not found"}) + + def do_POST(self): + route = { + "/v1/ingest/manifest": self._manifest, + "/v1/ingest": self._ingest, + "/v1/search": self._search, + "/v1/expand": self._expand, + "/v1/step": self._step, + "/v1/grep": self._grep, + "/v1/recent": self._recent, + "/v1/ask": self._ask, + }.get(self.path) + if route is None: + return self._send(404, {"error": "not found"}) + owner = self._auth() + if owner is None: + # No hint about which half was wrong: unknown, malformed and + # revoked keys are one answer. + return self._send(401, {"error": "unauthorized"}) + body = self._body() + if body is None: + return self._send(400, {"error": "bad request body"}) + try: + route(owner, body) + except storage.OffsetMismatch as mismatch: + # 409, with our size, so the client can resume without a full + # re-upload of a months-long transcript. + self._send(409, {"error": "offset mismatch", "size": mismatch.actual}) + except storage.UnsafePath as bad: + self._send(400, {"error": str(bad)}) + except Exception as unexpected: # noqa: BLE001 + self._send(500, {"error": type(unexpected).__name__}) + + def _manifest(self, owner: str, body: dict) -> None: + self._send(200, {"files": storage.manifest( + hub.transcripts, owner, hub.ledger)}) + + def _ingest(self, owner: str, body: dict) -> None: + path = body.get("path") + offset = body.get("offset", 0) + if not isinstance(path, str) or not isinstance(offset, int): + return self._send(400, {"error": "path and offset required"}) + try: + data = base64.b64decode(body.get("data") or "", validate=True) + except (ValueError, TypeError): + return self._send(400, {"error": "data must be base64"}) + # How much of the CLIENT's file this covers. It differs from + # len(data) whenever the client redacted before sending, and the + # client's own count is the only correct one — the hub never sees + # the original bytes. Trusting it is safe: the only thing a wrong + # number corrupts is that member's own resume point. + raw_len = body.get("raw_len") + if raw_len is not None and (not isinstance(raw_len, int) or raw_len < 0): + return self._send(400, {"error": "raw_len must be a non-negative int"}) + # Mask BEFORE the bytes touch the disk: a credential that lands in + # the transcript tree is already in the backup and already indexed + # by the time anyone notices. surrogateescape keeps a transcript + # with odd bytes byte-exact instead of failing the upload. + text = data.decode("utf-8", errors="surrogateescape") + masked, hits = hub.secret_map.mask(text) + stored = masked.encode("utf-8", errors="surrogateescape") + size = storage.append(hub.transcripts, owner, path, offset, stored, + hub.ledger, + received_len=raw_len if raw_len is not None + else len(data)) + self._send(200, {"size": size, "masked": hits}) + + def _search(self, owner: str, body: dict) -> None: + query = body.get("query") + if not isinstance(query, str) or not query.strip(): + return self._send(400, {"error": "query required"}) + start_ts, end_ts = _range(body) + 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) + self._send(200, {"anchors": [_anchor(a) for a in hits], + "degraded": getattr(hits, "degraded", None)}) + + def _expand(self, owner: str, body: dict) -> None: + turns = hub.recall.expand_around( + body.get("session_id", ""), body.get("uuid", ""), + int(body.get("before", 2)), int(body.get("after", 2)), + source=body.get("source")) + self._send(200, {"turns": [asdict(t) for t in turns]}) + + def _step(self, owner: str, body: dict) -> None: + try: + turns = hub.recall.step( + body.get("session_id", ""), body.get("uuid", ""), + body.get("direction", "next"), int(body.get("count", 1)), + source=body.get("source")) + except ValueError as bad: + return self._send(400, {"error": str(bad)}) + self._send(200, {"turns": [asdict(t) for t in turns]}) + + def _grep(self, owner: str, body: dict) -> None: + pattern = body.get("pattern") + if not isinstance(pattern, str) or not pattern: + return self._send(400, {"error": "pattern required"}) + start_ts, end_ts = _range(body) + 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) + self._send(200, {"anchors": [_anchor(a) for a in hits]}) + + def _recent(self, owner: str, body: dict) -> None: + start_ts, end_ts = _range(body) + self._send(200, {"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)}) + + def _ask(self, owner: str, body: dict) -> None: + question = body.get("question") + if not isinstance(question, str) or not question.strip(): + return self._send(400, {"error": "question required"}) + self._send(200, hub_ask.answer( + hub, question, k=int(body.get("k") or hub_ask.MAX_FRAGMENTS), + scope_cwd=body.get("scope_cwd"), source=body.get("source"), + composer=hub.composer)) + + return Handler diff --git a/src/session_recall/hub/ask.py b/src/session_recall/hub/ask.py new file mode 100644 index 0000000..d94ab76 --- /dev/null +++ b/src/session_recall/hub/ask.py @@ -0,0 +1,114 @@ +"""Answering a plain question from the pooled index. + +`/v1/search` returns anchors an agent can walk. `ask` is for the other case — +a person (or an agent that just wants the conclusion) asking "how did we do +X", getting prose back. Retrieval feeds the composer; the composer writes. + +With no composer configured the endpoint still answers, with a deterministic +digest of the fragments it found. That is deliberately not an error path: an +operator who has not wired a model yet should get the same evidence, just +unwritten, rather than a broken feature. + +Attribution is part of the answer's value here — "Егор делал это в проекте X" +is often the actual thing the asker needed — so sources carry the owner. What +they do NOT carry is the transcript path: it names a directory on the server +and helps nobody. +""" + +import time + +from ..timefmt import humanize_ts +from .indexer import owner_of + +# Appended to the shared composer system prompt. The hub pools everyone's +# history, including whatever people typed on a bad day, so the model is told +# what not to repeat. This is a second layer and is treated as one: the +# guarantees live in the masking map and the redactor, because a prompt is a +# request and not a boundary. +PRIVACY_RULE = """ + +This index pools the history of a whole team, and fragments may contain \ +material that is not about the work: health, family, money, private messages, \ +anything from someone's life outside the job. Never repeat, summarise or \ +allude to such material, even when a fragment contains it and even when the \ +question asks for it directly — answer the work part and drop the rest \ +silently. Saying whose work a finding came from is fine and useful; \ +speculating about a person is not.""" + +MAX_FRAGMENTS = 12 + + +def _digest(chunks: list) -> str: + """The no-composer answer: the fragments themselves, plainly labelled.""" + if not chunks: + return "Ничего похожего в истории команды не нашлось." + lines = ["Готового ответа нет (модель не подключена), вот что нашлось:"] + for c in chunks: + who = f"{c['owner']}, " if c.get("owner") else "" + lines.append(f"\n— {who}{c['project']} ({c['when_human']}):\n {c['snippet']}") + return "\n".join(lines) + + +def answer(hub, question: str, k: int = MAX_FRAGMENTS, + scope_cwd: str | None = None, composer=None, + source: str | None = None) -> dict: + """Retrieve, then write. Returns answer + the sources it used.""" + hits = hub.recall.recall_search(question, k=k, scope_cwd=scope_cwd, + source=source) + owners = _owners_for(hub, hits) + # Humanised here, not taken off the anchor: `when_human` is added when an + # anchor is serialised for a tool response, so an answer built straight + # from Recall would print "( )" where the date belongs. + now = int(time.time()) + chunks = [] + for anchor in hits: + chunks.append({ + "session_id": anchor.session_id, + "uuid": anchor.uuid, + "role": anchor.role, + "project": anchor.project, + "snippet": anchor.snippet, + "when": anchor.when, + "when_human": humanize_ts(anchor.when, now), + "source": anchor.source, + "owner": owners.get(anchor.session_id), + }) + + written = None + if composer is not None and chunks: + # The three-part request the composer expects. A hub question arrives + # as one sentence, so task/problem are filled from it rather than + # invented — retrieval already ran, and these only steer the writing. + written = composer({"question": question, "task": "(вопрос к общей истории команды)", + "problem": "(не указано)"}, chunks) + return { + "answer": written or _digest(chunks), + "composed": bool(written), + "degraded": getattr(hits, "degraded", None), + "sources": [{k: c[k] for k in + ("owner", "project", "session_id", "uuid", "when_human", + "source", "snippet")} for c in chunks], + } + + +def _owners_for(hub, anchors) -> dict[str, str]: + """session_id -> member, for a whole result set in one query. + + Derived from the stored transcript path rather than a column: the tree + layout already encodes ownership, so this needs no schema change and is + correct for rows indexed before this endpoint existed. + """ + ids = {a.session_id for a in anchors} + if not ids: + return {} + placeholders = ",".join("?" * len(ids)) + rows = hub.recall.store.db.execute( + f"SELECT session_id, file_path FROM chunks " + f"WHERE session_id IN ({placeholders}) GROUP BY session_id", + tuple(ids)).fetchall() + found = {} + for session_id, file_path in rows: + owner = owner_of(hub, file_path) if file_path else None + if owner: + found[session_id] = owner + return found diff --git a/src/session_recall/hub/auth.py b/src/session_recall/hub/auth.py new file mode 100644 index 0000000..e68682c --- /dev/null +++ b/src/session_recall/hub/auth.py @@ -0,0 +1,130 @@ +"""Who is talking to the hub. + +A key looks like `sr__<32 hex>`. The owner travels in the clear on +purpose — an operator reading a config file, a log line or a `key list` should +know whose key it is without a lookup — and the random half is the secret. + +Only a SHA-256 of the whole key is stored. A hub database that leaks therefore +hands out no access, which matters more here than in the solo product: this +file sits next to a pooled index of the whole team's history. + +The owner in the key is a claim, not the trust: it counts only because the +stored hash proves the bearer holds a key the operator issued for that name. +Ingest writes into the tree of the owner resolved this way, never a name the +client sends in the request body — otherwise anyone could write into anyone's +history. + +Deliberately absent in v1: expiry and per-key rate limits. The hub is reachable +only over TLS by people the operator issued keys to, and revocation is +immediate; adding clocks and counters before there is a second failure mode to +defend against would be ceremony. +""" + +import hashlib +import json +import os +import re +import secrets +import time +from pathlib import Path + +_KEY_RE = re.compile(r"^sr_([a-z0-9][a-z0-9-]{0,31})_([0-9a-f]{32})$") +_BEARER_RE = re.compile(r"^Bearer\s+(\S+)$", re.IGNORECASE) + + +def key_hash(key: str) -> str: + return hashlib.sha256(key.encode()).hexdigest() + + +def parse_owner(key: str) -> str | None: + """The owner a key claims. Says nothing about whether the key is valid.""" + found = _KEY_RE.match(key or "") + return found.group(1) if found else None + + +def bearer(header: str | None) -> str | None: + """Pull the credential out of an Authorization header.""" + found = _BEARER_RE.match((header or "").strip()) + return found.group(1) if found else None + + +class KeyStore: + """Issued keys, on disk as JSON keyed by hash. + + Re-read on every verify: the file is tiny, the OS caches it, and an + operator who revokes a key expects the next request to fail — not the next + restart. + """ + + def __init__(self, path: Path, clock=time.time): + self.path = Path(path) + self.clock = clock + + def _load(self) -> dict: + try: + return json.loads(self.path.read_text()) + except (OSError, ValueError): + return {} + + def _save(self, data: dict) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + tmp = self.path.with_suffix(".tmp") + tmp.write_text(json.dumps(data, indent=2, sort_keys=True)) + os.chmod(tmp, 0o600) + tmp.replace(self.path) # atomic: a crash mid-write never truncates the store + + def issue(self, owner: str, note: str = "") -> str: + """Mint a key for `owner` and return it — the only time it exists in + readable form. Nothing recoverable is kept, so a lost key is reissued, + never recovered.""" + if not re.match(r"^[a-z0-9][a-z0-9-]{0,31}$", owner or ""): + raise ValueError( + f"bad owner name {owner!r}: lowercase letters, digits and dashes") + key = f"sr_{owner}_{secrets.token_hex(16)}" + data = self._load() + data[key_hash(key)] = { + "owner": owner, "note": note, + "issued": int(self.clock()), "revoked": None, + } + self._save(data) + return key + + def verify(self, header: str | None) -> str | None: + """Owner for a valid Authorization header, else None. + + Lookup is by hash of the presented key, so no comparison walks the + secret byte by byte and a wrong key cannot be distinguished from an + unknown one by timing. + """ + key = bearer(header) + if not key or not parse_owner(key): + return None + record = self._load().get(key_hash(key)) + if not record or record.get("revoked"): + return None + return record["owner"] + + def revoke(self, selector: str) -> int: + """Revoke by key id (the short hash shown in `list`) or by owner name. + + Owner-wide revocation is the common case and the reason it is one call: + somebody leaves, and every device they ever enrolled must stop working + in one command, not one command per laptop. + """ + data = self._load() + hit = 0 + for digest, record in data.items(): + if record.get("revoked"): + continue + if digest.startswith(selector) or record.get("owner") == selector: + record["revoked"] = int(self.clock()) + hit += 1 + if hit: + self._save(data) + return hit + + def listing(self) -> list[dict]: + """Issued keys, newest first. Never contains anything usable as a key.""" + rows = [{"id": digest[:8], **record} + for digest, record in self._load().items()] + return sorted(rows, key=lambda r: r.get("issued", 0), reverse=True) diff --git a/src/session_recall/hub/cli.py b/src/session_recall/hub/cli.py new file mode 100644 index 0000000..bbabf42 --- /dev/null +++ b/src/session_recall/hub/cli.py @@ -0,0 +1,254 @@ +"""`session-recall hub …` — the team hub, from both sides. + +Operator commands run the service, issue and revoke member keys, refresh the +masking map and index what has arrived. Member commands are just `join`, +`push`, `status` and `leave`. They share a namespace but not an +implementation: the member half lives in `hub/client.py` and needs nothing +from the server module. + +Key issuance is intentionally CLI-only and never exposed over HTTP: minting +credentials is an operator action at a shell, not a request an agent or a +compromised member key can make. +""" + +import argparse +import json +import os +from http.server import ThreadingHTTPServer +from pathlib import Path + +DEFAULT_DIR = "/var/lib/claude-recall" +DEFAULT_PORT = 8788 + + +def hub_dir(args) -> Path: + return Path(getattr(args, "dir", None) + or os.environ.get("SESSION_RECALL_HUB_DIR") + or DEFAULT_DIR).expanduser() + + +def add_parser(sub) -> None: + hp = sub.add_parser("hub", help="team hub: serve, keys, indexing") + hp.add_argument("--dir", default=None, + help=f"hub data directory (default {DEFAULT_DIR}, " + f"or SESSION_RECALL_HUB_DIR)") + hsub = hp.add_subparsers(dest="hub_cmd", required=True) + + sp = hsub.add_parser("serve", help="run the hub HTTP service") + sp.add_argument("--host", default="127.0.0.1", + help="bind address; keep it on localhost behind a TLS proxy") + sp.add_argument("--port", type=int, default=DEFAULT_PORT) + + kp = hsub.add_parser("key", help="member keys") + ksub = kp.add_subparsers(dest="key_cmd", required=True) + ki = ksub.add_parser("issue", help="mint a key for a member") + ki.add_argument("owner", help="short name: lowercase letters, digits, dashes") + ki.add_argument("--note", default="", help="free text, e.g. which laptop") + ksub.add_parser("list", help="issued keys (never shows a usable key)") + kr = ksub.add_parser("revoke", help="revoke by key id or by member name") + kr.add_argument("selector") + + hsub.add_parser("index", help="index everything members have uploaded") + hsub.add_parser("remask", + help="re-apply the current masking map to stored transcripts") + + # --- member side --------------------------------------------------- + jp = hsub.add_parser("join", help="connect this machine to a team hub") + jp.add_argument("url", help="https://recall.example.com") + jp.add_argument("key", help="the key your operator handed you") + jp.add_argument("--yes", action="store_true", + help="accept the disclosure without the interactive prompt") + pp = hsub.add_parser("push", help="upload new transcript bytes to the hub") + pp.add_argument("--quiet", action="store_true", help="only print the summary") + ap = hsub.add_parser("ask", help="ask the team's history a question") + ap.add_argument("question") + ap.add_argument("-k", type=int, default=12, help="fragments to retrieve") + ap.add_argument("--scope", default=None, help="limit to one repo (cwd)") + hsub.add_parser("status", help="what this machine sends, and where") + hsub.add_parser("leave", help="stop sending (uploaded history stays on the hub)") + + secp = hsub.add_parser("secrets", help="the Doppler masking map") + secsub = secp.add_subparsers(dest="secrets_cmd", required=True) + secsub.add_parser("refresh", help="rebuild the map from Doppler") + secsub.add_parser("status", help="how many secrets are masked, and when") + + +def _print_answer(result: dict) -> None: + print(result["answer"]) + if result.get("degraded"): + print(f"\n⚠ поиск деградировал: {result['degraded']}") + if not result.get("composed"): + print("\n(модель не подключена — показаны найденные фрагменты)") + seen = [] + for source in result.get("sources", []): + label = f"{source.get('owner') or '?'} · {source.get('project') or '?'}" + if label not in seen: + seen.append(label) + if seen: + print("\nисточники: " + ", ".join(seen)) + + +def _member(args) -> int | None: + """Member-side commands. Return None when `args` is not one of them.""" + from .client import CONFIG_PATH, CONSENT, HubConfig, HubError, join, push + + if args.hub_cmd == "join": + print(CONSENT) + if not args.yes: + try: + answer = input("Подключить эту машину к общему индексу? [y/N] ") + except EOFError: + answer = "" + if answer.strip().lower() not in ("y", "yes", "д", "да"): + print("отменено — ничего не отправлено") + return 1 + try: + cfg = join(args.url, args.key) + except HubError as failure: + print(f"не подключились: {failure}") + return 1 + print(f"подключено к {cfg.url}\n" + f"дальше: session-recall hub push (первая заливка — вся история)") + return 0 + + if args.hub_cmd == "push": + cfg = HubConfig.load() + if cfg is None: + print("эта машина не подключена — сначала `session-recall hub join`") + return 1 + try: + stats = push(cfg, progress=None if args.quiet else print) + except HubError as failure: + print(f"заливка не удалась: {failure}") + return 1 + print(f"файлов отправлено: {stats['files']}, " + f"байт: {stats['uploaded_bytes']}, " + f"вырезано секретов: {stats['redacted']}, " + f"пропущено: {stats['skipped']}, ошибок: {stats['failed']}") + return 1 if stats["failed"] else 0 + + if args.hub_cmd == "ask": + cfg = HubConfig.load() + if cfg is None: + return None # no membership: fall through to the operator path + from .remote import RemoteRecall + try: + result = RemoteRecall(cfg).ask(args.question, k=args.k, + scope_cwd=args.scope) + except HubError as failure: + print(f"не удалось спросить: {failure}") + return 1 + _print_answer(result) + return 0 + + if args.hub_cmd == "status": + cfg = HubConfig.load() + if cfg is None: + print("не подключено к хабу (solo-режим: индекс остаётся локальным)") + return 0 + print(json.dumps({"url": cfg.url, "config": str(CONFIG_PATH), + "consented": cfg.consented}, indent=2)) + return 0 + + if args.hub_cmd == "leave": + if CONFIG_PATH.exists(): + CONFIG_PATH.unlink() + print("отключено — эта машина больше ничего не отправляет.\n" + "Уже загруженное остаётся на сервере: попроси оператора удалить.") + else: + print("эта машина и так не подключена") + return 0 + return None + + +def run(args) -> int: + member = _member(args) + if member is not None: + return member + + from .app import Hub, make_handler + from .auth import KeyStore + + root = hub_dir(args) + hub = Hub(root) + + if args.hub_cmd == "serve": + httpd = ThreadingHTTPServer((args.host, args.port), make_handler(hub)) + masked = len(hub.secret_map.labels) + print(f"claude-recall on {args.host}:{args.port}, data in {root}") + print(f"masking {masked} known secret(s)" + if masked else + "WARNING: no masking map — run `session-recall hub secrets refresh` " + "before enrolling members, or credentials land in the index") + try: + httpd.serve_forever() + except KeyboardInterrupt: + pass + finally: + httpd.server_close() + return 0 + + if args.hub_cmd == "key": + keys = KeyStore(root / "keys.json") + if args.key_cmd == "issue": + key = keys.issue(args.owner, args.note) + print(key) + print("\nHand this to the member ONCE, over a channel you trust. " + "It is not stored and cannot be shown again.") + return 0 + if args.key_cmd == "list": + for row in keys.listing(): + state = f"revoked" if row.get("revoked") else "active" + note = f" {row['note']}" if row.get("note") else "" + print(f"{row['id']} {row['owner']:<16} {state}{note}") + return 0 + if args.key_cmd == "revoke": + count = keys.revoke(args.selector) + print(f"revoked {count} key(s)") + return 0 if count else 1 + + if args.hub_cmd == "index": + from .indexer import index_all + print(json.dumps(index_all(hub), indent=2)) + return 0 + + if args.hub_cmd == "remask": + from .indexer import remask_all + print(json.dumps(remask_all(hub), indent=2)) + return 0 + + if args.hub_cmd == "ask": + # Reached only when this machine has not joined a hub — i.e. the + # operator asking their own server directly. + from . import ask as hub_ask + _print_answer(hub_ask.answer(hub, args.question, k=args.k, + scope_cwd=args.scope, + composer=hub.composer)) + return 0 + + if args.hub_cmd == "secrets": + from .masking import SecretMap, collect_from_doppler + if args.secrets_cmd == "refresh": + try: + entries = collect_from_doppler() + except Exception as failure: # noqa: BLE001 + print(f"doppler unavailable: {failure}\n" + f"keeping the previous map " + f"({len(hub.secret_map.labels)} secrets)") + return 1 + previous = SecretMap.load(hub.secrets_path) + # Reuse the salt so the map stays comparable across refreshes and + # an operator can diff counts without every hash changing. + fresh = SecretMap.build(entries, salt=previous.salt or None) + fresh.save(hub.secrets_path) + print(f"read {len(entries)} secret(s) from Doppler, " + f"masking {len(fresh.labels)} of them " + f"(short or low-entropy values are skipped on purpose)") + return 0 + if args.secrets_cmd == "status": + smap = hub.secret_map + print(json.dumps({"masked_secrets": len(smap.labels), + "updated": smap.updated}, indent=2)) + return 0 + + raise argparse.ArgumentTypeError(f"unknown hub command: {args.hub_cmd}") diff --git a/src/session_recall/hub/client.py b/src/session_recall/hub/client.py new file mode 100644 index 0000000..cea8a09 --- /dev/null +++ b/src/session_recall/hub/client.py @@ -0,0 +1,266 @@ +"""The member side: enroll once, then keep the hub current. + +Two commands are the whole surface. `join` stores the URL and the key the +operator handed over; `push` walks the local transcript roots and uploads what +the hub does not have yet. Everything else — embedding, indexing, answering — +happens on the server, which is the point: a member installs, joins, and never +thinks about it again. + +Uploads are incremental by construction. Transcripts only ever grow, so the +hub's manifest of "bytes consumed per file" is enough to compute the tail to +send; a re-push after a day of work moves kilobytes, not gigabytes. A file +that has SHRUNK locally was rewritten rather than appended to, and is re-sent +whole — the only way for the two sides to converge again. + +Redaction happens here, before anything leaves the machine, and it is +deliberately the weaker of the two secret defences: it matches formats, so it +catches an API key and cannot catch a password. The hub's Doppler map catches +what the team actually stores. Doing the format pass client-side means those +findings never travel at all. +""" + +import base64 +import gzip +import json +import os +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from pathlib import Path + +from .. import config +from ..share.scanner import redact +from . import storage + +CONFIG_PATH = config.DATA_DIR / "hub.json" +# Sized from the first real deploy, not from taste. At 256 KB a 3.5 GB +# history moved at ~0.1 MB/s: every chunk is its own request, and urllib +# opens a fresh connection each time, so the cost was one TLS handshake plus +# one fsync per 256 KB rather than the bytes themselves. 4 MB cuts that by +# 16x and still leaves room under the hub's 16 MB body cap once base64 +# expands it by a third (the body is gzipped on top, so the wire figure is +# lower again). +CHUNK_BYTES = 4 * 1024 * 1024 +_TIMEOUT_S = 300 # a 4 MB chunk on a slow uplink outlasts 60s + +CONSENT = """\ +Подключение к общему индексу команды + +Что произойдёт: транскрипты твоих сессий Claude Code и Codex — включая +промпты, ответы, вызовы инструментов и их вывод — будут загружаться на +сервер команды и станут доступны для поиска ВСЕМ участникам, а не только +тебе. + +Что защищено: перед отправкой вырезаются ключи и токены известных форматов, +а на сервере дополнительно маскируются все значения, которые команда хранит +в Doppler. Это тревожная растяжка, а не гарантия — личное в рабочие сессии +лучше не писать. + +Отключиться можно в любой момент: `session-recall hub leave`. Уже загруженное +при этом остаётся на сервере — попроси оператора удалить. +""" + + +class HubError(RuntimeError): + pass + + +@dataclass +class HubConfig: + url: str + key: str + consented: int = 0 + + @classmethod + def load(cls, path: Path | None = None) -> "HubConfig | None": + try: + data = json.loads(Path(path or CONFIG_PATH).read_text()) + except (OSError, ValueError): + return None + if not data.get("url") or not data.get("key"): + return None + return cls(data["url"].rstrip("/"), data["key"], + int(data.get("consented", 0))) + + def save(self, path: Path | None = None) -> None: + path = Path(path or CONFIG_PATH) + path.parent.mkdir(parents=True, exist_ok=True) + # 0600 before the key is written, not after: a world-readable moment + # is all it takes on a shared machine. + fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as fh: + json.dump({"url": self.url, "key": self.key, + "consented": self.consented}, fh, indent=2) + + +class HubClient: + def __init__(self, cfg: HubConfig, timeout: int = _TIMEOUT_S): + self.cfg = cfg + self.timeout = timeout + + def _post(self, path: str, body: dict) -> dict: + raw = gzip.compress(json.dumps(body).encode()) + request = urllib.request.Request( + self.cfg.url + path, data=raw, + headers={"Content-Type": "application/json", + "Content-Encoding": "gzip", + "Authorization": f"Bearer {self.cfg.key}"}) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + return json.loads(response.read()) + except urllib.error.HTTPError as failure: + detail = "" + try: + detail = json.loads(failure.read()).get("error", "") + except Exception: # noqa: BLE001 + pass + if failure.code == 401: + raise HubError( + "hub rejected the key — it may have been revoked; " + "ask the operator for a new one") from failure + if failure.code == 409: + raise _Conflict() from failure + raise HubError(f"hub error {failure.code}: {detail}") from failure + except urllib.error.URLError as failure: + raise HubError(f"hub unreachable: {failure.reason}") from failure + + def manifest(self) -> dict[str, int]: + return self._post("/v1/ingest/manifest", {}).get("files", {}) + + def send(self, rel: str, offset: int, data: bytes, raw_len: int) -> int: + return self._post("/v1/ingest", { + "path": rel, "offset": offset, "raw_len": raw_len, + "data": base64.b64encode(data).decode()})["size"] + + +class _Conflict(Exception): + """The hub holds a different length than we assumed — resend from zero.""" + + +def local_files(claude_root: Path | None = None, + codex_sessions: Path | None = None, + codex_archive: Path | None = None) -> list[tuple[str, Path]]: + """(relative path on the hub, local file) for everything uploadable. + + Cursor is absent by design: its history lives in a SQLite store, not in + files, so there is nothing to ship byte-for-byte. It stays local until the + hub speaks that format too. + """ + claude_root = config.CLAUDE_PROJECTS if claude_root is None else claude_root + codex_sessions = config.CODEX_SESSIONS if codex_sessions is None else codex_sessions + codex_archive = (config.CODEX_ARCHIVED_SESSIONS if codex_archive is None + else codex_archive) + + found: list[tuple[str, Path]] = [] + if Path(claude_root).is_dir(): + for project in sorted(Path(claude_root).iterdir()): + if not project.is_dir(): + continue + # Non-recursive, matching the indexer: nested files are subagent + # sidechains, which the extractor does not treat as sessions. + for jsonl in sorted(project.glob("*.jsonl")): + found.append((f"claude/{project.name}/{jsonl.name}", jsonl)) + for root, prefix in ((codex_sessions, "codex"), + (codex_archive, "codex/archived")): + if not Path(root).is_dir(): + continue + for jsonl in sorted(Path(root).rglob("*.jsonl")): + rel = jsonl.relative_to(root).as_posix() + found.append((f"{prefix}/{rel}", jsonl)) + return found + + +def push(cfg: HubConfig, progress=None, roots: dict | None = None) -> dict: + """Bring the hub up to date with this machine. Returns a summary. + + One file failing does not abort the run: a member's history is thousands + of independent transcripts, and stopping at the first unreadable one would + mean the whole push never completes. + """ + progress = progress or (lambda _msg: None) + client = HubClient(cfg) + have = client.manifest() + files = local_files(**(roots or {})) + + stats = {"files": 0, "uploaded_bytes": 0, "redacted": 0, + "skipped": 0, "failed": 0} + for rel, path in files: + if not storage.is_safe_rel(rel): + # A name the hub would reject. Warn rather than fail: it is one + # project directory, not the push. + progress(f"skip (unsupported name): {rel}") + stats["skipped"] += 1 + continue + try: + stats["uploaded_bytes"] += _push_one( + client, rel, path, have.get(rel, 0), stats, progress) + except HubError: + raise # auth/network: the whole run is doomed + except Exception as failure: # noqa: BLE001 + progress(f"failed: {rel}: {type(failure).__name__}: {failure}") + stats["failed"] += 1 + return stats + + +def _push_one(client: HubClient, rel: str, path: Path, offset: int, + stats: dict, progress) -> int: + size = path.stat().st_size + if size < offset: + # Locally shorter than what the hub consumed: the transcript was + # rewritten, not appended to. Resend from the top. + offset = 0 + if size == offset: + return 0 + sent_total = 0 + with open(path, "rb") as fh: + fh.seek(offset) + cursor = offset + while True: + piece = fh.read(CHUNK_BYTES) + if not piece: + break + text = piece.decode("utf-8", errors="surrogateescape") + clean, hits = redact(text) + stats["redacted"] += hits + payload = clean.encode("utf-8", errors="surrogateescape") + try: + client.send(rel, cursor, payload, raw_len=len(piece)) + except _Conflict: + # Someone else's copy of this member's history got there + # first, or a previous run died mid-chunk. Restart the file. + progress(f"resync: {rel}") + return _resend_whole(client, rel, path, stats, progress) + cursor += len(piece) + sent_total += len(piece) + stats["files"] += 1 + progress(f"sent {rel} (+{sent_total} B)") + return sent_total + + +def _resend_whole(client: HubClient, rel: str, path: Path, stats: dict, + progress) -> int: + sent_total = 0 + with open(path, "rb") as fh: + cursor = 0 + while True: + piece = fh.read(CHUNK_BYTES) + if not piece: + break + clean, hits = redact(piece.decode("utf-8", errors="surrogateescape")) + stats["redacted"] += hits + client.send(rel, cursor, clean.encode("utf-8", errors="surrogateescape"), + raw_len=len(piece)) + cursor += len(piece) + sent_total += len(piece) + stats["files"] += 1 + progress(f"resent {rel} ({sent_total} B)") + return sent_total + + +def join(url: str, key: str, path: Path | None = None) -> HubConfig: + """Store the enrollment and prove it works before claiming success.""" + cfg = HubConfig(url.rstrip("/"), key, consented=int(time.time())) + HubClient(cfg).manifest() # raises HubError on a bad key or URL + cfg.save(path) + return cfg diff --git a/src/session_recall/hub/indexer.py b/src/session_recall/hub/indexer.py new file mode 100644 index 0000000..dee8e6b --- /dev/null +++ b/src/session_recall/hub/indexer.py @@ -0,0 +1,114 @@ +"""Turn everything members have uploaded into one searchable index. + +This is where the hub's "no second implementation" rule pays off: it walks +each member's tree and hands it to the ordinary `index_corpus`, the same +function a laptop runs against its own `~/.claude/projects`. Incremental +signatures, vector reuse for unchanged chunks, pruning of deleted transcripts +and the embedding-space fingerprint all come along for free. + +Runs from a timer rather than from the ingest request: an upload should +return as soon as the bytes are durable, and embedding a freshly arrived +month of history behind a POST would time out. One run at a time, enforced by +a lock — two indexers would fight over the same per-file transactions. +""" + +import sys +from pathlib import Path + +from ..embed import make_embedder +from ..index import index_corpus +from ..metadocs.lock import acquire_lock +from ..store import Store +from . import storage + +LOCK_NAME = "hub-index.lock" + + +def index_all(hub, embedder=None) -> dict: + """Index every member's transcripts. Returns per-owner new-chunk counts. + + A member whose tree fails to index does not stop the others: their history + is separate data, and one corrupt transcript should not freeze the team's + index at yesterday. + """ + fd = acquire_lock(hub.root, LOCK_NAME) + if fd is None: + return {"skipped": "another indexer run holds the lock"} + counts: dict[str, object] = {} + store = None + try: + store = Store(hub.db_path) + store.db.execute("PRAGMA journal_mode=WAL") + store.db.execute("PRAGMA busy_timeout=30000") + embedder = embedder or make_embedder() + for owner in storage.owners(hub.transcripts): + roots = storage.source_roots(hub.transcripts, owner) + try: + counts[owner] = index_corpus( + store, embedder, roots["claude"], [roots["codex"]]) + except Exception as failure: # noqa: BLE001 + counts[owner] = f"failed: {type(failure).__name__}: {failure}" + print(f"claude-recall: indexing {owner} failed: {failure}", + file=sys.stderr) + finally: + if store is not None: + store.close() + import os + os.close(fd) + return counts + + +def remask_all(hub) -> dict: + """Re-apply the CURRENT masking map to everything already stored. + + Masking happens at ingest, so anything uploaded before a map was fixed or + extended keeps whatever was masked at the time. Re-uploading gigabytes to + fix that is absurd when the bytes are already here; this rewrites them in + place instead. + + The ledger is deliberately NOT touched: it counts bytes of the CLIENT's + file, which this does not change. Rewriting a transcript does change its + mtime and size, so the next index run re-reads it — and unchanged chunks + keep their vectors through the content-hash cache, so re-embedding costs + only what actually changed. + """ + smap = hub.secret_map + if not smap: + return {"skipped": "no masking map configured"} + changed, masked_total = 0, 0 + for owner in storage.owners(hub.transcripts): + base = storage.owner_root(hub.transcripts, owner) + for path in base.rglob("*.jsonl"): + if path.is_symlink() or not path.is_file(): + continue + try: + text = path.read_text(encoding="utf-8", errors="surrogateescape") + masked, hits = smap.mask(text) + if not hits: + continue + # Write through a temp file in the same directory: a crash + # mid-rewrite must not leave a half-masked transcript. + tmp = path.with_suffix(".jsonl.remask") + tmp.write_text(masked, encoding="utf-8", errors="surrogateescape") + tmp.replace(path) + changed += 1 + masked_total += hits + except OSError as failure: + print(f"claude-recall: remask failed for {path}: {failure}", + file=sys.stderr) + return {"files_rewritten": changed, "secrets_masked": masked_total} + + +def owner_of(hub, file_path: str) -> str | None: + """Which member a stored transcript belongs to. + + Attribution comes from the path rather than a column: the tree layout + already encodes it, and deriving it keeps the shared schema identical to + the solo one. + """ + try: + rel = Path(file_path).resolve().relative_to( + Path(hub.transcripts).resolve()) + except (ValueError, OSError): + return None + return rel.parts[0] if rel.parts else None diff --git a/src/session_recall/hub/masking.py b/src/session_recall/hub/masking.py new file mode 100644 index 0000000..7d848d1 --- /dev/null +++ b/src/session_recall/hub/masking.py @@ -0,0 +1,234 @@ +"""Mask the secrets we already know about, by their Doppler name. + +The regex scanner in `share/scanner.py` guesses at FORMATS — it catches +`sk-ant-…` because that shape is recognisable. It cannot catch +`NETCUP_PASSWORD`, because a password has no shape. This module works the +other way round: everything the team keeps in Doppler is known by value, so a +literal match is exact, has no false negatives for anything stored there, and +replaces the hit with something more useful than `[REDACTED]` — the variable +name it came from: + + ssh root@1.2.3.4 hunter2xyzzy… -> ssh root@1.2.3.4 ${servers/NETCUP_PASSWORD} + +A reader still learns which credential the session used, which is usually the +part that matters when reading someone's history, and the value is gone. + +**The hub never stores the secret values.** It keeps HMACs of them under a +local salt, and matches by hashing candidate tokens out of the incoming text. +A masking map that leaks is therefore worthless to whoever takes it — which is +the entire point of putting the team's Doppler inventory on a shared server. +The salt makes precomputation useless too, so short low-entropy values are the +only weak case, and those are excluded anyway (below). + +Two deliberate limits, both of which the regex layer still covers: + +- **Values that do not survive tokenisation** — connection strings, URLs with + inline credentials, multi-line PEM blocks — never appear as one token, so + they are not matched here. `scanner.py` recognises those by shape. +- **Short or low-entropy values are skipped.** Doppler holds `DOPPLER_CONFIG=dev` + next to real credentials; masking every value blindly would delete the word + "dev" from every transcript the team owns. The bar is length plus some + variety, and names known to be non-secret are excluded outright. +""" + +import hashlib +import json +import os +import re +import secrets as pysecrets +import subprocess +import time +from pathlib import Path + +MIN_LENGTH = 12 +_DOPPLER_TIMEOUT_S = 30 + +# Doppler injects these into every config; they are labels, not credentials. +_NEVER_MASK = { + "DOPPLER_PROJECT", "DOPPLER_CONFIG", "DOPPLER_ENVIRONMENT", +} + +# Only variables whose NAME says "credential" are masked. Learned the hard way +# on the first real corpus: shape alone cannot tell `claude-opus-4` from a +# token — both are short, mixed-class strings — so an entropy filter masked +# 851 mentions of a model name, 2429 of a hostname and 1536 of a server IP, +# which broke reading and searching the team's history far worse than the +# leak it prevented. A Doppler config holds credentials AND plain +# configuration; only the first kind belongs here. +_SECRET_NAME_RE = re.compile( + r"PASSWORD|PASSWD|SECRET|TOKEN|CREDENTIAL|PRIVATE|SIGNING|" + r"API[_-]?KEY|ACCESS[_-]?KEY|[_-]KEY$|^KEY$|DSN|SALT|SESSION[_-]?ID", + re.IGNORECASE) + +# Three alphabets, widest first. One tokenisation is not enough: with only the +# wide class, `--key=sk-ant-AAAA…` is a single token and never equals the +# stored value, while the medium class splits it at `=` and yields the key +# itself. Cheap enough to run all three — a handful of regex passes over text +# that is about to be embedded anyway. +_ALPHABETS = ( + re.compile(rf"[A-Za-z0-9_\-+/=~.]{{{MIN_LENGTH},}}"), + re.compile(rf"[A-Za-z0-9_\-]{{{MIN_LENGTH},}}"), + re.compile(rf"[A-Za-z0-9]{{{MIN_LENGTH},}}"), +) +_TRIM = "-._=+/~" + +# Transcripts are JSON, so a newline inside a string is the two characters +# `\` and `n`. The backslash is not in any alphabet above and ends a token, +# but the `n` IS, which glues it to whatever follows: a secret written right +# after a line break tokenises as `n` and matches nothing. Found in +# production — one real API key survived masking this way, in a message that +# read "вот замени в doppler…\n\n". Same for \t, \r, \b, \f. +_JSON_ESCAPE_LETTERS = "ntrbf" + + +def _candidates(token: str, text: str, start: int) -> list[str]: + """Forms of `token` worth testing against the secret map. + + Extra candidates are free: matching is by exact hash, so a form that is + not a secret simply misses. Missing a form, by contrast, leaks. + """ + forms = [token, token.strip(_TRIM)] + if start > 0 and text[start - 1] == "\\" and token[0] in _JSON_ESCAPE_LETTERS: + forms.append(token[1:]) + forms.append(token[1:].strip(_TRIM)) + return [f for f in forms if len(f) >= MIN_LENGTH] + + +def maskable(name: str, value: str) -> bool: + """Is this Doppler entry a credential worth masking on sight? + + Two gates, and the NAME gate is the important one: a Doppler config mixes + credentials with ordinary configuration (model names, hostnames, regions), + and only the former should ever be rewritten. The shape gate then rejects + the short, low-entropy values that would eat ordinary words. + """ + if name in _NEVER_MASK or not value or len(value) < MIN_LENGTH: + return False + if not _SECRET_NAME_RE.search(name): + return False + if any(ch.isspace() for ch in value): + return False # multi-line/PEM: never a single token anyway + classes = sum(( + any(c.islower() for c in value), + any(c.isupper() for c in value), + any(c.isdigit() for c in value), + any(not c.isalnum() for c in value), + )) + # Two character classes is the ordinary credential. A single-class value + # has to be long instead: 20+ characters with no whitespace, matching a + # Doppler value exactly, is a generated passphrase and not prose. + return classes >= 2 or len(value) >= 20 + + +class SecretMap: + """Hashed secret values -> the variable name to show instead. + + Persisted as JSON so a hub restart (or a Doppler outage) keeps masking + with the last known inventory rather than silently starting to store + credentials in the clear. + """ + + def __init__(self, salt: str, labels: dict[str, str], updated: int = 0): + self.salt = salt + self.labels = labels # hmac hex -> "project/VAR" + self.updated = updated + + @staticmethod + def _digest(salt: str, token: str) -> str: + return hashlib.blake2b(token.encode(), key=salt.encode()[:64], + digest_size=16).hexdigest() + + @classmethod + def build(cls, entries: dict[str, str], salt: str | None = None) -> "SecretMap": + """`entries` maps "project/VAR" -> value. Values are hashed and dropped.""" + salt = salt or pysecrets.token_hex(16) + labels = {cls._digest(salt, value): name + for name, value in entries.items() + if maskable(name.split("/")[-1], value)} + return cls(salt, labels, int(time.time())) + + @classmethod + def load(cls, path: Path) -> "SecretMap": + try: + data = json.loads(Path(path).read_text()) + except (OSError, ValueError): + return cls(salt="", labels={}) + return cls(data.get("salt", ""), data.get("labels", {}), + int(data.get("updated", 0))) + + def save(self, path: Path) -> None: + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps( + {"salt": self.salt, "labels": self.labels, "updated": self.updated}, + indent=2, sort_keys=True)) + os.chmod(tmp, 0o600) + tmp.replace(path) + + def __bool__(self) -> bool: + return bool(self.salt and self.labels) + + def mask(self, text: str) -> tuple[str, int]: + """Replace every known secret with `${project/VAR}`. + + Returns the masked text and how many replacements were made — the + count is what an operator watches to know the map is actually live. + """ + if not self: + return text, 0 + hits = 0 + for pattern in _ALPHABETS: + def swap(match: re.Match) -> str: + nonlocal hits + token = match.group(0) + for candidate in _candidates(token, match.string, match.start()): + label = self.labels.get(self._digest(self.salt, candidate)) + if label: + hits += 1 + return token.replace(candidate, "${" + label + "}") + return token + text = pattern.sub(swap, text) + return text, hits + + +def collect_from_doppler(runner=None) -> dict[str, str]: + """Every secret the caller's Doppler token can read, as "project/VAR". + + Shells out to the `doppler` CLI rather than the API: the CLI already holds + the machine's credentials, so the hub never needs a Doppler token of its + own in its config. + """ + def run(args: list[str]) -> str: + done = subprocess.run(args, capture_output=True, text=True, + timeout=_DOPPLER_TIMEOUT_S) + if done.returncode != 0: + raise RuntimeError((done.stderr or "doppler failed").strip()[:200]) + return done.stdout + + runner = runner or run + entries: dict[str, str] = {} + projects = json.loads(runner(["doppler", "projects", "--json"])) + for project in projects: + name = project.get("id") or project.get("name") + if not name: + continue + try: + configs = json.loads( + runner(["doppler", "configs", "--json", "-p", name])) + except (RuntimeError, ValueError): + continue # a project this token cannot read is not an error + for config in configs: + config_name = config.get("name") + if not config_name: + continue + try: + values = json.loads(runner([ + "doppler", "secrets", "download", "--no-file", + "--format", "json", "-p", name, "-c", config_name])) + except (RuntimeError, ValueError): + continue + for var, value in values.items(): + if isinstance(value, str): + entries[f"{name}/{var}"] = value + return entries diff --git a/src/session_recall/hub/remote.py b/src/session_recall/hub/remote.py new file mode 100644 index 0000000..79b5c65 --- /dev/null +++ b/src/session_recall/hub/remote.py @@ -0,0 +1,105 @@ +"""A Recall that lives on the hub. + +Implements the same five operations as the local `Recall`, over HTTP, so +`server.py` can hand either one to the MCP tools without knowing which. The +agent-facing contract does not change: same tool names, same arguments, same +shapes coming back. + +Why proxy at all, instead of pointing the MCP host straight at a remote +server: the key stays in a 0600 file managed by `hub join` rather than in the +host's config, the plugin installs unchanged, and a hub that is down produces +one clear sentence instead of a transport error the agent has to interpret. + +Date arguments are resolved to epochs HERE, on the machine that knows the +user's timezone, and sent as numbers. A server guessing at "yesterday" for +someone three timezones away would be quietly wrong. +""" + +import gzip +import json +import urllib.error +import urllib.request + +from ..models import Anchor, Turn +from ..retrieve import SearchResult +from .client import HubConfig, HubError + +_TIMEOUT_S = 60 + + +class RemoteRecall: + def __init__(self, cfg: HubConfig, timeout: int = _TIMEOUT_S): + self.cfg = cfg + self.timeout = timeout + + def _post(self, path: str, body: dict) -> dict: + payload = gzip.compress(json.dumps(body).encode()) + request = urllib.request.Request( + self.cfg.url + path, data=payload, + headers={"Content-Type": "application/json", + "Content-Encoding": "gzip", + "Authorization": f"Bearer {self.cfg.key}"}) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + return json.loads(response.read()) + except urllib.error.HTTPError as failure: + if failure.code == 401: + raise HubError("hub rejected this machine's key — ask the " + "operator to issue a new one") from failure + raise HubError(f"hub error {failure.code}") from failure + except urllib.error.URLError as failure: + raise HubError(f"hub unreachable: {failure.reason}") from failure + + @staticmethod + def _anchors(rows: list) -> list[Anchor]: + fields = Anchor.__dataclass_fields__ + return [Anchor(**{k: v for k, v in row.items() if k in fields}) + for row in rows] + + @staticmethod + def _turns(rows: list) -> list[Turn]: + fields = Turn.__dataclass_fields__ + return [Turn(**{k: v for k, v in row.items() if k in fields}) + for row in rows] + + 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: + body = self._post("/v1/search", { + "query": query, "k": k, "scope_cwd": scope_cwd, "source": source, + "start_ts": start_ts, "end_ts": end_ts}) + return SearchResult(self._anchors(body.get("anchors", [])), + body.get("degraded")) + + def expand_around(self, session_id: str, uuid: str, before: int = 2, + after: int = 2, source: str | None = None) -> list[Turn]: + return self._turns(self._post("/v1/expand", { + "session_id": session_id, "uuid": uuid, "before": before, + "after": after, "source": source}).get("turns", [])) + + def step(self, session_id: str, uuid: str, direction: str, count: int = 1, + source: str | None = None) -> list[Turn]: + return self._turns(self._post("/v1/step", { + "session_id": session_id, "uuid": uuid, "direction": direction, + "count": count, "source": source}).get("turns", [])) + + 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]: + 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", [])) + + 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]: + return self._post("/v1/recent", { + "scope_cwd": scope_cwd, "limit": limit, "source": source, + "start_ts": start_ts, "end_ts": end_ts}).get("sessions", []) + + def ask(self, question: str, k: int = 12, + scope_cwd: str | None = None) -> dict: + return self._post("/v1/ask", { + "question": question, "k": k, "scope_cwd": scope_cwd}) diff --git a/src/session_recall/hub/storage.py b/src/session_recall/hub/storage.py new file mode 100644 index 0000000..3a04ccc --- /dev/null +++ b/src/session_recall/hub/storage.py @@ -0,0 +1,216 @@ +"""Where a hub keeps what clients send, and the boundary that keeps it safe. + +One tree per member, mirroring the roots each client has locally: + + /transcripts//claude//.jsonl + /transcripts//codex///
/.jsonl + +That mirroring is load-bearing, not cosmetic: `index.index_corpus` walks +exactly these shapes, so a hub indexes a colleague's history through the same +code that indexes a laptop's own. + +Every segment of a stored path arrives over the network, which makes `resolve` +the security boundary of the ingest side. Three things it must survive, all of +them cheap to get wrong: + +- `..` and absolute paths escaping into another member's tree (or out of the + data directory entirely); +- names that are not path traversal but still hostile to the filesystem — + empty, dot-only, control characters, a leading dash; +- a symlink planted by an earlier write redirecting a later one. Pattern + matching alone cannot see that, so the resolved path is re-checked against + the resolved owner root. + +Transcripts are append-only in normal operation, so the wire protocol is +offset-based: a client says "at byte N I have these bytes". A mismatch is +reported rather than papered over — silently appending at the wrong offset +would interleave two writers into a file that no longer parses. + +**Received bytes are counted separately from file size.** Secret masking +rewrites the incoming text, so what lands on disk is a different length from +what the client sent. The ledger therefore records how much of the CLIENT's +file we have consumed; sizing the manifest off the stored file instead would +make every masked transcript look truncated and re-upload forever. +""" + +import json +import os +import re +from pathlib import Path + +# One segment of a relative path. Deliberately narrow: transcript names are +# session ids, project directories and date parts, none of which need more. +# The first character cannot be a dot, which rules out `..` and hidden files. +# A LEADING DASH IS ALLOWED and must stay allowed: Claude Code names its +# project directories after the encoded cwd (`-Users-egor-proj`), so rejecting +# them would reject every Claude transcript there is. Nothing here reaches a +# shell or an argv, so a flag-shaped name is inert. +_SEGMENT_RE = re.compile(r"^[A-Za-z0-9_-][A-Za-z0-9._-]{0,127}$") +_OWNER_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,31}$") + +# The only roots a client may write into. `source` names match the indexer's. +SOURCES = ("claude", "codex") + +MAX_DEPTH = 8 + + +class UnsafePath(ValueError): + """A relative path from a client that will not be written anywhere.""" + + +class OffsetMismatch(ValueError): + """The client's idea of the file length disagrees with ours.""" + + def __init__(self, actual: int): + super().__init__(f"file is {actual} bytes") + self.actual = actual + + +def valid_owner(owner: str) -> bool: + return bool(_OWNER_RE.match(owner or "")) + + +def owner_root(root: Path, owner: str) -> Path: + if not valid_owner(owner): + raise UnsafePath(f"bad owner name: {owner!r}") + return Path(root) / owner + + +def resolve(root: Path, owner: str, rel: str) -> Path: + """Absolute path for a client-supplied relative path, or raise UnsafePath. + + `rel` must start with a known source directory and consist of plain + segments; the result is verified to sit under the owner's root even after + symlinks are followed. + """ + if not rel or rel.startswith("/") or "\\" in rel or "\0" in rel: + raise UnsafePath(f"bad path: {rel!r}") + parts = rel.split("/") + if len(parts) < 2 or len(parts) > MAX_DEPTH: + raise UnsafePath(f"bad path depth: {rel!r}") + if parts[0] not in SOURCES: + raise UnsafePath(f"unknown source directory: {parts[0]!r}") + for part in parts: + if not _SEGMENT_RE.match(part): + raise UnsafePath(f"bad path segment: {part!r}") + if not parts[-1].endswith(".jsonl"): + raise UnsafePath(f"not a transcript: {parts[-1]!r}") + + base = owner_root(root, owner) + target = base.joinpath(*parts) + # Second, independent check: the pattern above cannot see a symlink that an + # earlier write left behind, so compare the resolved paths. strict=False — + # the file legitimately does not exist yet on a first upload. + resolved_base = base.resolve(strict=False) + if not target.resolve(strict=False).is_relative_to(resolved_base): + raise UnsafePath(f"path escapes its owner root: {rel!r}") + return target + + +def is_safe_rel(rel: str) -> bool: + """Would `resolve` accept this path? Used by the CLIENT before uploading. + + Shared with the server on purpose: a client that applies the same rule + skips the handful of local directories the hub would reject (odd + characters in a project name) with a warning, instead of discovering it as + a 400 in the middle of a long push. + """ + try: + resolve(Path("/nonexistent-probe"), "probe", rel) + except UnsafePath: + return False + return True + + +class Ledger: + """How many bytes of each client-side file we have consumed. + + Kept outside the transcript trees (the indexer walks those) and written + atomically, so an interrupted upload resumes from a truthful number rather + than a half-written one. + """ + + def __init__(self, state_dir: Path): + self.dir = Path(state_dir) + + def _path(self, owner: str) -> Path: + if not valid_owner(owner): + raise UnsafePath(f"bad owner name: {owner!r}") + return self.dir / f"{owner}.json" + + def read(self, owner: str) -> dict[str, int]: + try: + data = json.loads(self._path(owner).read_text()) + except (OSError, ValueError): + return {} + return {k: int(v) for k, v in data.items() if isinstance(v, int)} + + def write(self, owner: str, rel: str, received: int) -> None: + data = self.read(owner) + data[rel] = received + path = self._path(owner) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(data, sort_keys=True)) + tmp.replace(path) + + +def manifest(root: Path, owner: str, ledger: Ledger) -> dict[str, int]: + """Relative path -> bytes of the client's file already consumed. + + The client diffs this against its own files and uploads only the tails, + which is what keeps a re-push of a months-long history nearly free. Entries + whose stored transcript has since disappeared are dropped, so a deleted + file on the hub is re-sent whole instead of being silently skipped. + """ + base = owner_root(root, owner) + return {rel: received for rel, received in ledger.read(owner).items() + if (base / rel).is_file()} + + +def append(root: Path, owner: str, rel: str, offset: int, data: bytes, + ledger: Ledger, received_len: int | None = None) -> int: + """Store `data` for the client's byte range starting at `offset`. + + `received_len` is how many bytes of the CLIENT's file this covers, which + differs from len(data) once masking has rewritten the text. Returns the new + received total. + + offset == current received total appends (the normal case for a growing + transcript). offset == 0 replaces the file: a client sends that when its + local copy no longer matches what we hold — a rewritten or rotated + transcript — and replacing is the only way to converge. Anything else is a + real disagreement and raises. + """ + if offset < 0: + raise UnsafePath(f"negative offset: {offset}") + target = resolve(root, owner, rel) + current = ledger.read(owner).get(rel, 0) if target.exists() else 0 + if offset not in (0, current): + raise OffsetMismatch(current) + target.parent.mkdir(parents=True, exist_ok=True) + # "wb" for a replacement, "ab" to extend. Not "r+b": appending through the + # append flag is atomic against a concurrent push from the same member's + # second machine, where a seek+write pair is not. + with open(target, "wb" if offset == 0 else "ab") as fh: + fh.write(data) + fh.flush() + os.fsync(fh.fileno()) + total = offset + (len(data) if received_len is None else received_len) + ledger.write(owner, rel, total) + return total + + +def owners(root: Path) -> list[str]: + """Members that have sent anything, for the indexer to walk.""" + base = Path(root) + if not base.is_dir(): + return [] + return sorted(p.name for p in base.iterdir() + if p.is_dir() and not p.is_symlink() and valid_owner(p.name)) + + +def source_roots(root: Path, owner: str) -> dict[str, Path]: + """The per-source directories to hand to the indexer for one member.""" + base = owner_root(root, owner) + return {source: base / source for source in SOURCES} diff --git a/src/session_recall/metadocs/lock.py b/src/session_recall/metadocs/lock.py index b6b2452..8d22c11 100644 --- a/src/session_recall/metadocs/lock.py +++ b/src/session_recall/metadocs/lock.py @@ -8,10 +8,13 @@ from pathlib import Path -def acquire_lock(data_dir: Path) -> int | None: - """Returns the fd holding the lock, or None when another run owns it.""" +def acquire_lock(data_dir: Path, name: str = "metadocs.lock") -> int | None: + """Returns the fd holding the lock, or None when another run owns it. + + `name` lets a second long-running job (the hub indexer) take its own lock + with the same semantics instead of copying this file.""" data_dir.mkdir(parents=True, exist_ok=True) - fd = os.open(data_dir / "metadocs.lock", os.O_CREAT | os.O_WRONLY, 0o600) + fd = os.open(data_dir / name, os.O_CREAT | os.O_WRONLY, 0o600) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: diff --git a/src/session_recall/server.py b/src/session_recall/server.py index dc0db1e..30d2260 100644 --- a/src/session_recall/server.py +++ b/src/session_recall/server.py @@ -21,7 +21,21 @@ def _adict(a) -> dict: return d -def build_recall() -> Recall: +def build_recall(): + """The local index, or the team hub when this machine has joined one. + + The five tools below do not change shape either way — that is the point. + A member's agent asks the same questions and gets the same answers; only + the corpus is bigger and the embedding happens on the server. + + Deciding here (rather than per tool) keeps the solo install untouched: no + hub config, no import, no behaviour change. + """ + from .hub.client import HubConfig + cfg = HubConfig.load() + if cfg is not None: + from .hub.remote import RemoteRecall + return RemoteRecall(cfg) return Recall(Store(DB_PATH), make_embedder(), make_reranker()) diff --git a/src/session_recall/share/compose.py b/src/session_recall/share/compose.py index 9c0d9ec..cdb6c33 100644 --- a/src/session_recall/share/compose.py +++ b/src/session_recall/share/compose.py @@ -21,6 +21,7 @@ from typing import Callable MODEL = "claude-opus-5" +CODEX_MODEL = "gpt-5.6-terra" MAX_TOKENS = 4000 CLI_TIMEOUT_S = 180 @@ -88,7 +89,7 @@ def _prompt(req: dict, chunks: list, turns: list | None = None) -> str: "Write the answer.") -def _cli_composer(runner=None) -> Composer: +def _cli_composer(runner=None, system_extra: str = "") -> Composer: """Compose through the locally installed `claude` CLI. Costs nothing beyond the existing subscription and needs no API key, but the @@ -120,7 +121,7 @@ def compose(req: dict, chunks: list, turns: list | None = None) -> str | None: "claude", "-p", "--tools", "", "--strict-mcp-config", - "--system-prompt", _SYSTEM, + "--system-prompt", _SYSTEM + system_extra, _prompt(req, chunks, turns), ], empty) except (OSError, subprocess.SubprocessError): @@ -132,7 +133,95 @@ def compose(req: dict, chunks: list, turns: list | None = None) -> str | None: return compose -def _api_composer(client=None) -> Composer | None: +# Present in the environment, these route codex away from the signed-in +# subscription and onto metered API billing. +_API_ROUTING_VARS = ("OPENAI_API_KEY", "OPENAI_BASE_URL") + + +def _subscription_env() -> dict: + return {k: v for k, v in os.environ.items() if k not in _API_ROUTING_VARS} + + +def _codex_composer(runner=None, model: str | None = None, + system_extra: str = "") -> Composer: + """Compose through the locally installed `codex` CLI. + + Same bargain as the Claude CLI path — an existing subscription instead of + an API key — but Codex needs more stripping, because `codex exec` is a full + agent with a shell: + + - `--ephemeral` keeps the run out of `$CODEX_HOME/sessions`. This is not + tidiness: a hub indexes what it stores, so a persisted worker session + would put a colleague's question into the team index, where it later + reads as the team's own past work and fires as a stored injection. The + Claude CLI path documents this hazard and leaves it to the operator; + here it is closed by construction. + - `--ignore-user-config` and `--ignore-rules` drop the operator's hooks, + notifications and execpolicy. On a laptop those merely pollute stdout; + on a server they are someone else's automation running inside an + answer to an untrusted question. + - `--sandbox read-only` plus an empty working directory removes writes. + Reads are NOT removed — the CLI exposes no flag for that — so the + remaining containment is the unix user the hub runs this as. That is a + deployment property, not a code one, and the hub's service unit is + where it gets enforced. + + The model is passed explicitly because `--ignore-user-config` also drops + the configured default; a service should not inherit whatever an operator + last set for themselves anyway. + + **Subscription, never an API key.** Codex authenticates from + `$CODEX_HOME/auth.json` (which `--ignore-user-config` deliberately still + reads), but an `OPENAI_API_KEY` in the environment would silently divert + the run to metered API billing instead. The variable is therefore stripped + from the child environment: on a server that answers a whole team, the + difference between "subscription" and "API" is a bill nobody agreed to. + """ + model = model or os.environ.get("SESSION_RECALL_COMPOSE_MODEL") or CODEX_MODEL + + def run(args, cwd, env): + return subprocess.run(args, cwd=cwd, env=env, capture_output=True, + text=True, stdin=subprocess.DEVNULL, + timeout=CLI_TIMEOUT_S) + + runner = runner or run + + def compose(req: dict, chunks: list, turns: list | None = None) -> str | None: + if not chunks: + return None + with tempfile.TemporaryDirectory() as empty: + answer = os.path.join(empty, "answer.txt") + try: + done = runner([ + "codex", "exec", + "--ephemeral", + "--ignore-user-config", + "--ignore-rules", + "--skip-git-repo-check", + "--sandbox", "read-only", + "-C", empty, + "-m", model, + "-o", answer, + # No --system-prompt in codex exec: the instructions ride + # at the head of the prompt, above the untrusted material. + f"{_SYSTEM}{system_extra}\n\n{_prompt(req, chunks, turns)}", + ], empty, _subscription_env()) + except (OSError, subprocess.SubprocessError): + return None + if getattr(done, "returncode", 1) != 0: + return None + # Read the last message from the file rather than stdout: stdout + # carries event logs and a token tally around the answer. + try: + with open(answer, encoding="utf-8") as fh: + return fh.read().strip() or None + except OSError: + return None + + return compose + + +def _api_composer(client=None, system_extra: str = "") -> Composer | None: if client is None: try: import anthropic @@ -149,7 +238,7 @@ def compose(req: dict, chunks: list, turns: list | None = None) -> str | None: max_tokens=MAX_TOKENS, betas=["server-side-fallback-2026-07-01"], fallbacks="default", - system=_SYSTEM, + system=_SYSTEM + system_extra, messages=[{"role": "user", "content": _prompt(req, chunks, turns)}], ) @@ -163,14 +252,20 @@ def compose(req: dict, chunks: list, turns: list | None = None) -> str | None: return compose -def make_composer(env: dict | None = None, client=None, runner=None) -> Composer | None: +def make_composer(env: dict | None = None, client=None, runner=None, + system_extra: str = "") -> Composer | None: """None means "no composer configured" — the caller keeps the deterministic digest. `client`/`runner` are injectable so tests never touch the network or - spawn a process.""" + spawn a process. `system_extra` appends caller-specific rules to the shared + system prompt — the team hub uses it to add its privacy rule without + forking the whole prompt.""" env = os.environ if env is None else env engine = (env.get("SESSION_RECALL_COMPOSE") or "none").strip().lower() if engine in ("claude-cli", "cli"): - return _cli_composer(runner=runner) + return _cli_composer(runner=runner, system_extra=system_extra) if engine in ("claude", "api", "claude-api"): - return _api_composer(client=client) + return _api_composer(client=client, system_extra=system_extra) + if engine in ("codex", "codex-cli"): + return _codex_composer(runner=runner, system_extra=system_extra, + model=env.get("SESSION_RECALL_COMPOSE_MODEL")) return None diff --git a/src/session_recall/share/scanner.py b/src/session_recall/share/scanner.py index b57de4e..be8b391 100644 --- a/src/session_recall/share/scanner.py +++ b/src/session_recall/share/scanner.py @@ -11,10 +11,14 @@ import re from dataclasses import dataclass +# ORDER MATTERS for `redact`, which replaces on first match: a specific +# pattern must precede any broader one that also matches it, or the +# replacement carries the wrong label. `sk-ant-…` is also a valid `sk-…`, so +# anthropic-key comes first. (`scan` reports every match and is order-blind.) _PATTERNS: list[tuple[str, re.Pattern]] = [ ("aws-access-key", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b")), - ("openai-key", re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b")), ("anthropic-key", re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b")), + ("openai-key", re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b")), ("github-token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{36,}\b")), ("gitlab-token", re.compile(r"\bglpat-[A-Za-z0-9_-]{20,}\b")), ("slack-token", re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b")), @@ -43,3 +47,40 @@ def scan(text: str) -> list[Finding]: for m in pattern.finditer(text): findings.append(Finding(kind=kind, excerpt=_mask(m.group(0)))) return findings + + +# Patterns precise enough to cut with nobody looking. `password-assignment` is +# deliberately NOT here: `\w*(token|secret|key)\s*[=:]\s*\S{8,}` matches any +# JSON field, config line or sentence shaped like "token: …", and on the first +# real corpus it fired 27324 times against ~1900 for every real key format +# combined — it deletes the surrounding work, not the secrets. It stays in +# `scan`, where a human reads the finding and decides. +_AUTO_REDACT = { + "aws-access-key", "anthropic-key", "openai-key", "github-token", + "gitlab-token", "slack-token", "jwt", "private-key-pem", + "telegram-bot-token", "connection-string", +} + + +def redact(text: str) -> tuple[str, int]: + """Replace confidently-identified secrets with `[REDACTED:]`. + + `scan` flags for a human who then decides; this one decides for them, and + exists for the path where no human is in the loop: a hub client uploading + a member's transcripts. There the same finding cannot become a preview to + approve — the alternative to cutting it is shipping it — so the tripwire + is wired to act, and therefore has to be far more conservative about what + it fires on than the flagging path. + + Still a tripwire, not a guarantee: it knows FORMATS, so it catches an + `sk-ant-…` and cannot catch a password. Values the team actually keeps in + Doppler are handled by exact match on the hub (`hub/masking.py`); the two + layers miss different things on purpose. + """ + count = 0 + for kind, pattern in _PATTERNS: + if kind not in _AUTO_REDACT: + continue + text, hits = pattern.subn(f"[REDACTED:{kind}]", text) + count += hits + return text, count diff --git a/src/session_recall/store.py b/src/session_recall/store.py index 12ebe26..fc911ec 100644 --- a/src/session_recall/store.py +++ b/src/session_recall/store.py @@ -19,13 +19,26 @@ class Store: - def __init__(self, db_path: Path): + def __init__(self, db_path: Path, read_only: bool = False): + """`read_only` opens the index for search only. + + A hub runs two processes against one database: the HTTP service reads, + a timer indexes. sqlite-vec keeps its vectors in shadow tables, and a + long-lived read-write connection from the service made the indexer's + writes fail with "could not write to vector blob" — 63 transcripts + were silently dropped from one real run before this was found. A + read-only connection cannot take part in that: one writer, many + readers, which is what SQLite is actually good at.""" Path(db_path).parent.mkdir(parents=True, exist_ok=True) - self.db = sqlite3.connect(str(db_path)) + if read_only: + self.db = sqlite3.connect(f"file:{Path(db_path)}?mode=ro", uri=True) + else: + self.db = sqlite3.connect(str(db_path)) self.db.enable_load_extension(True) sqlite_vec.load(self.db) self.db.enable_load_extension(False) - self._schema() + if not read_only: + self._schema() self._assert_dim_matches(db_path) def _assert_dim_matches(self, db_path: Path) -> None: diff --git a/tests/test_hub_ask.py b/tests/test_hub_ask.py new file mode 100644 index 0000000..69d7176 --- /dev/null +++ b/tests/test_hub_ask.py @@ -0,0 +1,248 @@ +"""`ask` and the MCP proxy: the two ways a person or an agent reads the +pooled index.""" + +import json +import threading +import urllib.error +import urllib.request +from http.server import ThreadingHTTPServer + +import pytest + +from session_recall.hub import ask as hub_ask +from session_recall.hub.app import Hub, make_handler +from session_recall.hub.client import HubConfig +from session_recall.hub.remote import RemoteRecall +from session_recall.models import Anchor, Turn +from session_recall.retrieve import SearchResult + + +class FakeStore: + def __init__(self, rows): + self.db = self + self._rows = rows + + def execute(self, sql, params=()): + return self + + def fetchall(self): + return self._rows + + +class FakeRecall: + def __init__(self, hub_root=""): + self.store = FakeStore([ + ("s1", f"{hub_root}/transcripts/egor/claude/-Users-egor-proj/s1.jsonl")]) + self.searched = [] + + def recall_search(self, query, **kw): + self.searched.append((query, kw)) + return SearchResult([Anchor( + session_id="s1", uuid="u1", role="assistant", + snippet="запинили mcp<2 и перевыпустили конфиг", score=0.8, + project="pr-review", when=1785000000, source="claude")]) + + def expand_around(self, session_id, uuid, before=2, after=2, source=None): + return [Turn(role="user", type="user", content="как чинили?", raw={})] + + def step(self, session_id, uuid, direction, count=1, source=None): + return [Turn(role="assistant", type="assistant", content="вот так", raw={})] + + def grep(self, pattern, session_id=None, **kw): + return [Anchor(session_id="s1", uuid="u9", role="user", snippet=pattern, + score=1.0, project="pr-review", when=1785000000)] + + def recent_sessions(self, **kw): + return [{"source": "claude", "session_id": "s1", "project": "pr-review", + "turns": 12, "last_activity_human": "вчера", "label": "работа"}] + + +@pytest.fixture +def hub(tmp_path): + root = tmp_path / "hub" + # ONE instance shared across threads: the real Recall is thread-local (a + # sqlite connection cannot be shared), so a per-thread factory would hand + # the assertions in this file a different object than the request used. + shared = FakeRecall(str(root)) + return Hub(root, recall_factory=lambda: shared, composer=None) + + +@pytest.fixture +def url(hub): + httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(hub)) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{httpd.server_address[1]}" + httpd.shutdown() + httpd.server_close() + + +@pytest.fixture +def remote(url, hub): + return RemoteRecall(HubConfig(url, hub.keys.issue("maxim"))) + + +# -- ask --------------------------------------------------------------------- + +def test_without_a_model_ask_still_answers_with_what_it_found(hub): + result = hub_ask.answer(hub, "как чинили CI?", composer=None) + assert result["composed"] is False + assert "запинили mcp<2" in result["answer"] + + +def test_with_a_model_the_answer_is_written(hub): + result = hub_ask.answer(hub, "как чинили CI?", + composer=lambda req, chunks: "Запинили mcp<2.") + assert (result["composed"], result["answer"]) == (True, "Запинили mcp<2.") + + +def test_sources_say_whose_history_the_answer_came_from(hub): + result = hub_ask.answer(hub, "как чинили CI?", composer=None) + assert result["sources"][0]["owner"] == "egor" + assert result["sources"][0]["project"] == "pr-review" + + +def test_sources_never_leak_server_paths(hub): + result = hub_ask.answer(hub, "как чинили CI?", composer=None) + assert "file_path" not in result["sources"][0] + assert "/transcripts/" not in json.dumps(result["sources"]) + + +def test_the_question_reaches_the_composer_with_the_fragments(hub): + seen = {} + + def composer(req, chunks): + seen.update(req=req, chunks=chunks) + return "ok" + + hub_ask.answer(hub, "как чинили CI?", composer=composer) + assert seen["req"]["question"] == "как чинили CI?" + assert seen["chunks"][0]["snippet"].startswith("запинили") + + +def test_privacy_rule_states_what_must_not_be_repeated(): + rule = hub_ask.PRIVACY_RULE + assert "health, family, money, private messages" in rule + # attribution stays allowed — knowing whose work it was is the point + assert "Saying whose work a finding came from is fine" in rule + + +def test_privacy_rule_is_attached_to_the_hub_composer(tmp_path, monkeypatch): + captured = {} + + def fake_make_composer(system_extra=""): + captured["extra"] = system_extra + return lambda req, chunks: "ok" + + monkeypatch.setattr("session_recall.share.compose.make_composer", + fake_make_composer) + hub = Hub(tmp_path / "h", recall_factory=lambda: FakeRecall()) + assert hub.composer is not None + assert "health, family, money" in captured["extra"] + + +def test_ask_over_http_requires_a_question(url, hub): + key = hub.keys.issue("egor") + request = urllib.request.Request( + url + "/v1/ask", data=json.dumps({}).encode(), + headers={"Authorization": f"Bearer {key}"}) + with pytest.raises(urllib.error.HTTPError) as raised: + urllib.request.urlopen(request, timeout=10) + assert raised.value.code == 400 + + +# -- MCP proxy --------------------------------------------------------------- + +def test_remote_search_returns_real_anchors(remote): + hits = remote.recall_search("как чинили", k=3) + assert isinstance(hits, SearchResult) + assert isinstance(hits[0], Anchor) + assert hits[0].snippet.startswith("запинили") + assert hits.degraded is None + + +def test_remote_expand_and_step_return_turns(remote): + turns = remote.expand_around("s1", "u1") + assert isinstance(turns[0], Turn) and turns[0].content == "как чинили?" + assert remote.step("s1", "u1", "next")[0].content == "вот так" + + +def test_remote_grep_and_recent(remote): + assert remote.grep("mcp<2")[0].snippet == "mcp<2" + assert remote.recent_sessions()[0]["project"] == "pr-review" + + +def test_remote_ask(remote): + assert "запинили" in remote.ask("как чинили CI?")["answer"] + + +def test_dates_are_resolved_client_side_and_sent_as_epochs(remote, hub): + """The server has no business guessing the user's timezone.""" + remote.recall_search("q", start_ts=1785000000, end_ts=1785600000) + _, kw = hub.recall.searched[-1] + assert (kw["start_ts"], kw["end_ts"]) == (1785000000, 1785600000) + + +def test_a_dead_hub_says_so_in_one_sentence(): + from session_recall.hub.client import HubError + dead = RemoteRecall(HubConfig("http://127.0.0.1:1", "sr_x_" + "0" * 32), + timeout=2) + with pytest.raises(HubError, match="unreachable"): + dead.recall_search("anything") + + +def test_mcp_picks_the_hub_only_when_joined(monkeypatch, tmp_path): + """A solo install must not change behaviour, and must not even import + the hub client's network path.""" + from session_recall import server as mcp_server + + monkeypatch.setattr(HubConfig, "load", staticmethod(lambda path=None: None)) + monkeypatch.setattr(mcp_server, "Recall", lambda *a, **k: "LOCAL") + monkeypatch.setattr(mcp_server, "Store", lambda *a, **k: None) + monkeypatch.setattr(mcp_server, "make_embedder", lambda: None) + monkeypatch.setattr(mcp_server, "make_reranker", lambda: None) + assert mcp_server.build_recall() == "LOCAL" + + monkeypatch.setattr(HubConfig, "load", + staticmethod(lambda path=None: HubConfig("http://h", "k"))) + assert isinstance(mcp_server.build_recall(), RemoteRecall) + + +def test_digest_dates_are_human_readable(hub): + """`when_human` is added at serialisation time, so an answer built straight + from Recall must humanise the epoch itself or print an empty pair of + brackets where the date belongs.""" + result = hub_ask.answer(hub, "как чинили CI?", composer=None) + assert result["sources"][0]["when_human"].strip() + assert "()" not in result["answer"] + + +def test_remask_rewrites_what_was_stored_before_the_map_was_fixed(tmp_path): + """Masking runs at ingest, so a map fixed afterwards must be applicable to + bytes already on disk — re-uploading gigabytes to fix a regex is absurd.""" + from session_recall.hub import storage + from session_recall.hub.indexer import remask_all + from session_recall.hub.masking import SecretMap + + hub = Hub(tmp_path / "hub", recall_factory=lambda: FakeRecall()) + secret = "Xk39dmPQ7wLz2vRt" + rel = "claude/-Users-egor-proj/s1.jsonl" + ledger = hub.ledger + storage.append(hub.transcripts, "egor", rel, 0, + f'{{"text":"pass {secret}"}}\n'.encode(), ledger) + before = ledger.read("egor")[rel] + + SecretMap.build({"servers/NETCUP_PASSWORD": secret}, + salt="s").save(hub.secrets_path) + result = remask_all(hub) + + stored = storage.resolve(hub.transcripts, "egor", rel).read_text() + assert result["files_rewritten"] == 1 and result["secrets_masked"] == 1 + assert secret not in stored and "${servers/NETCUP_PASSWORD}" in stored + # the ledger counts the CLIENT's bytes, which a rewrite does not change + assert ledger.read("egor")[rel] == before + + +def test_remask_without_a_map_does_nothing(tmp_path): + from session_recall.hub.indexer import remask_all + hub = Hub(tmp_path / "hub2", recall_factory=lambda: FakeRecall()) + assert "skipped" in remask_all(hub) diff --git a/tests/test_hub_auth.py b/tests/test_hub_auth.py new file mode 100644 index 0000000..0a21154 --- /dev/null +++ b/tests/test_hub_auth.py @@ -0,0 +1,85 @@ +"""Member keys: what they prove, what a leaked hub database does not give away.""" + +import json + +import pytest + +from session_recall.hub.auth import KeyStore, bearer, parse_owner + + +@pytest.fixture +def keys(tmp_path): + return KeyStore(tmp_path / "keys.json") + + +def test_issued_key_verifies_to_its_owner(keys): + key = keys.issue("egor", note="macbook") + assert keys.verify(f"Bearer {key}") == "egor" + + +def test_key_carries_the_owner_in_the_clear(keys): + assert parse_owner(keys.issue("egor")) == "egor" + + +@pytest.mark.parametrize("header", [ + None, "", "Bearer", "Bearer ", "Basic sr_egor_" + "0" * 32, + "Bearer sr_egor_nothex", "Bearer sr_egor_0000", "Bearer not-a-key", + "Bearer sr_EGOR_" + "0" * 32, # owner names are lowercase +]) +def test_malformed_credentials_are_refused(keys, header): + keys.issue("egor") + assert keys.verify(header) is None + + +def test_an_unissued_but_well_formed_key_is_refused(keys): + keys.issue("egor") + assert keys.verify("Bearer sr_egor_" + "a" * 32) is None + + +def test_revoking_by_owner_kills_every_device_at_once(keys): + laptop, desktop = keys.issue("egor", "laptop"), keys.issue("egor", "desktop") + other = keys.issue("maxim") + assert keys.revoke("egor") == 2 + assert keys.verify(f"Bearer {laptop}") is None + assert keys.verify(f"Bearer {desktop}") is None + assert keys.verify(f"Bearer {other}") == "maxim" + + +def test_revoking_by_key_id_kills_one_device(keys): + laptop, desktop = keys.issue("egor", "laptop"), keys.issue("egor", "desktop") + target = next(r["id"] for r in keys.listing() if r["note"] == "laptop") + assert keys.revoke(target) == 1 + assert keys.verify(f"Bearer {laptop}") is None + assert keys.verify(f"Bearer {desktop}") == "egor" + + +def test_revoking_twice_is_not_counted_again(keys): + keys.issue("egor") + assert keys.revoke("egor") == 1 + assert keys.revoke("egor") == 0 + + +def test_the_store_never_holds_a_usable_key(keys, tmp_path): + key = keys.issue("egor") + on_disk = (tmp_path / "keys.json").read_text() + assert key not in on_disk + assert key.split("_")[-1] not in on_disk + + +def test_listing_shows_members_without_leaking_credentials(keys): + key = keys.issue("egor", "macbook") + row, = keys.listing() + assert (row["owner"], row["note"]) == ("egor", "macbook") + assert key not in json.dumps(row) + + +def test_bad_owner_names_are_refused_at_issue(keys): + for bad in ("", "Egor", "e" * 33, "../maxim", "egor egor"): + with pytest.raises(ValueError): + keys.issue(bad) + + +def test_bearer_parsing_is_case_insensitive_on_the_scheme(): + assert bearer("bearer abc") == "abc" + assert bearer("BEARER abc") == "abc" + assert bearer("Bearerabc") is None diff --git a/tests/test_hub_client.py b/tests/test_hub_client.py new file mode 100644 index 0000000..823ca1b --- /dev/null +++ b/tests/test_hub_client.py @@ -0,0 +1,188 @@ +"""Client against a real hub: the two halves of the protocol only matter +together, so these run the actual server on an ephemeral port.""" + +import json +import stat +import threading +from http.server import ThreadingHTTPServer + +import pytest + +from session_recall.hub import storage +from session_recall.hub.app import Hub, make_handler +from session_recall.hub.client import (CONSENT, HubConfig, HubError, join, + local_files, push) +from session_recall.hub.masking import SecretMap + +ANTHROPIC = "sk-ant-api03-" + "B" * 40 +NETCUP = "Xk39dmPQ7wLz2vRt" + + +@pytest.fixture +def hub(tmp_path): + return Hub(tmp_path / "hub", recall_factory=lambda: None) + + +@pytest.fixture +def url(hub): + httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(hub)) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{httpd.server_address[1]}" + httpd.shutdown() + httpd.server_close() + + +@pytest.fixture +def roots(tmp_path): + """A miniature version of the three local transcript roots.""" + claude = tmp_path / "claude-projects" / "-Users-egor-proj" + claude.mkdir(parents=True) + (claude / "sess-a.jsonl").write_text('{"type":"user","text":"привет"}\n') + codex = tmp_path / "codex-sessions" / "2026" / "08" / "05" + codex.mkdir(parents=True) + (codex / "roll-1.jsonl").write_text('{"type":"message","text":"codex"}\n') + archive = tmp_path / "codex-archive" + archive.mkdir() + (archive / "old.jsonl").write_text('{"type":"message","text":"old"}\n') + return {"claude_root": tmp_path / "claude-projects", + "codex_sessions": tmp_path / "codex-sessions", + "codex_archive": archive} + + +@pytest.fixture +def cfg(url, hub, tmp_path): + key = hub.keys.issue("egor") + return join(url, key, path=tmp_path / "hub.json") + + +def test_local_files_covers_claude_codex_and_the_archive(roots): + rels = dict(local_files(**roots)) + assert set(rels) == { + "claude/-Users-egor-proj/sess-a.jsonl", + "codex/2026/08/05/roll-1.jsonl", + "codex/archived/old.jsonl", + } + + +def test_join_verifies_the_key_before_saving(url, hub, tmp_path): + path = tmp_path / "hub.json" + with pytest.raises(HubError, match="revoked|rejected"): + join(url, "sr_egor_" + "0" * 32, path=path) + assert not path.exists() # a failed join leaves nothing behind + + +def test_join_stores_the_key_readable_only_by_its_owner(url, hub, tmp_path): + path = tmp_path / "hub.json" + join(url, hub.keys.issue("egor"), path=path) + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + assert HubConfig.load(path).url == url + + +def test_consent_text_states_what_actually_happens(): + assert "ВСЕМ участникам" in CONSENT + assert "hub leave" in CONSENT + + +def test_push_uploads_everything_then_nothing(cfg, hub, roots): + first = push(cfg, roots=roots) + assert first["files"] == 3 and first["uploaded_bytes"] > 0 + assert push(cfg, roots=roots) == { + "files": 0, "uploaded_bytes": 0, "redacted": 0, + "skipped": 0, "failed": 0} + + +def test_push_sends_only_the_tail_of_a_grown_transcript(cfg, hub, roots): + push(cfg, roots=roots) + grown = roots["claude_root"] / "-Users-egor-proj" / "sess-a.jsonl" + addition = '{"type":"assistant","text":"ответ"}\n' + with open(grown, "a") as fh: + fh.write(addition) + + stats = push(cfg, roots=roots) + assert stats["files"] == 1 + assert stats["uploaded_bytes"] == len(addition.encode()) + stored = storage.resolve(hub.transcripts, "egor", + "claude/-Users-egor-proj/sess-a.jsonl") + assert stored.read_text().endswith(addition) + + +def test_a_rewritten_transcript_is_resent_whole(cfg, hub, roots): + push(cfg, roots=roots) + rewritten = roots["claude_root"] / "-Users-egor-proj" / "sess-a.jsonl" + rewritten.write_text('{"short":1}\n') # now SHORTER than the hub's copy + + push(cfg, roots=roots) + stored = storage.resolve(hub.transcripts, "egor", + "claude/-Users-egor-proj/sess-a.jsonl") + assert stored.read_text() == '{"short":1}\n' + + +def test_format_shaped_secrets_never_leave_the_machine(cfg, hub, roots): + """Client-side redaction is the first layer: an API key is cut before the + request is built, so it is not merely masked on arrival — it never travels.""" + leaky = roots["claude_root"] / "-Users-egor-proj" / "sess-a.jsonl" + leaky.write_text(json.dumps({"text": f"key is {ANTHROPIC}"}) + "\n") + + stats = push(cfg, roots=roots) + stored = storage.resolve(hub.transcripts, "egor", + "claude/-Users-egor-proj/sess-a.jsonl").read_text() + assert stats["redacted"] == 1 + assert ANTHROPIC not in stored and "[REDACTED:anthropic-key]" in stored + + +def test_redaction_keeps_the_resume_point_aligned(cfg, hub, roots): + """Redaction changes length, so the hub must count the CLIENT's bytes — + otherwise the next push re-sends from a wrong offset forever.""" + leaky = roots["claude_root"] / "-Users-egor-proj" / "sess-a.jsonl" + line = json.dumps({"text": f"key is {ANTHROPIC}"}) + "\n" + leaky.write_text(line) + push(cfg, roots=roots) + + rel = "claude/-Users-egor-proj/sess-a.jsonl" + assert hub.ledger.read("egor")[rel] == len(line.encode()) + assert push(cfg, roots=roots)["files"] == 0 # nothing left to send + + +def test_hub_side_masking_still_applies_to_what_the_client_missed( + cfg, hub, roots): + """A password has no format, so the client cannot catch it; the Doppler + map on the hub is what stops it.""" + SecretMap.build({"servers/NETCUP_PASSWORD": NETCUP}, + salt="test-salt").save(hub.secrets_path) + leaky = roots["claude_root"] / "-Users-egor-proj" / "sess-a.jsonl" + leaky.write_text(json.dumps({"text": f"sshpass -p {NETCUP}"}) + "\n") + + stats = push(cfg, roots=roots) + stored = storage.resolve(hub.transcripts, "egor", + "claude/-Users-egor-proj/sess-a.jsonl").read_text() + assert stats["redacted"] == 0 # the client saw nothing + assert NETCUP not in stored + assert "${servers/NETCUP_PASSWORD}" in stored + + +def test_unsupported_names_are_skipped_not_fatal(cfg, hub, roots): + odd = roots["claude_root"] / "project with spaces" + odd.mkdir() + (odd / "sess.jsonl").write_text("{}\n") + stats = push(cfg, roots=roots) + assert stats["skipped"] == 1 and stats["files"] == 3 + + +def test_a_revoked_key_stops_the_push_with_a_clear_message(cfg, hub, roots): + hub.keys.revoke("egor") + with pytest.raises(HubError, match="revoked"): + push(cfg, roots=roots) + + +def test_broad_assignment_pattern_does_not_eat_ordinary_text(cfg, hub, roots): + """`token: abc12345` appears in code, JSON and prose everywhere. Cutting it + automatically removed 27324 fragments of real work on the first corpus.""" + ordinary = roots["claude_root"] / "-Users-egor-proj" / "sess-a.jsonl" + ordinary.write_text(json.dumps( + {"text": "в конфиге поле token: abcdef123456 — разберись почему падает"}) + "\n") + + stats = push(cfg, roots=roots) + stored = storage.resolve(hub.transcripts, "egor", + "claude/-Users-egor-proj/sess-a.jsonl").read_text() + assert stats["redacted"] == 0 + assert "token: abcdef123456" in stored diff --git a/tests/test_hub_http.py b/tests/test_hub_http.py new file mode 100644 index 0000000..260519d --- /dev/null +++ b/tests/test_hub_http.py @@ -0,0 +1,179 @@ +"""The hub over real HTTP: ThreadingHTTPServer on an ephemeral port, urllib as +the client — the same shape as the relay suite.""" + +import base64 +import gzip +import json +import threading +import urllib.error +import urllib.request +from http.server import ThreadingHTTPServer + +import pytest + +from session_recall.hub import storage +from session_recall.hub.app import Hub, make_handler +from session_recall.hub.masking import SecretMap +from session_recall.models import Anchor + +GOOD = "claude/-Users-egor-proj/sess.jsonl" +NETCUP = "Xk39dmPQ7wLz2vRt" + + +class FakeRecall: + def __init__(self): + self.calls = [] + + def recall_search(self, query, **kw): + self.calls.append((query, kw)) + return [Anchor(session_id="s1", uuid="u1", role="assistant", + snippet="we pinned the version", score=0.9, + project="proj", when=1785000000, source="claude")] + + +@pytest.fixture +def hub(tmp_path): + return Hub(tmp_path / "hub", recall_factory=FakeRecall) + + +@pytest.fixture +def server(hub): + httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(hub)) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{httpd.server_address[1]}" + httpd.shutdown() + httpd.server_close() + + +def post(url, path, body, key=None, compress=False): + raw = json.dumps(body).encode() + headers = {"Content-Type": "application/json"} + if compress: + raw, headers["Content-Encoding"] = gzip.compress(raw), "gzip" + if key: + headers["Authorization"] = f"Bearer {key}" + request = urllib.request.Request(url + path, data=raw, headers=headers) + with urllib.request.urlopen(request, timeout=10) as response: + return response.status, json.loads(response.read()) + + +def upload(url, key, data: bytes, offset=0, path=GOOD, **extra): + return post(url, "/v1/ingest", + {"path": path, "offset": offset, + "data": base64.b64encode(data).decode(), **extra}, key=key) + + +def test_healthz_needs_no_key_and_says_nothing_about_content(server): + with urllib.request.urlopen(server + "/healthz", timeout=10) as response: + assert json.loads(response.read()) == {"ok": True} + + +def test_every_data_route_requires_a_key(server): + for path in ("/v1/ingest/manifest", "/v1/ingest", "/v1/search"): + with pytest.raises(urllib.error.HTTPError) as raised: + post(server, path, {}) + assert raised.value.code == 401 + + +def test_a_revoked_key_stops_working_immediately(server, hub): + key = hub.keys.issue("egor") + assert post(server, "/v1/ingest/manifest", {}, key=key)[0] == 200 + hub.keys.revoke("egor") + with pytest.raises(urllib.error.HTTPError) as raised: + post(server, "/v1/ingest/manifest", {}, key=key) + assert raised.value.code == 401 + + +def test_unknown_route_is_404(server, hub): + with pytest.raises(urllib.error.HTTPError) as raised: + post(server, "/v1/whatever", {}, key=hub.keys.issue("egor")) + assert raised.value.code == 404 + + +def test_upload_then_manifest_roundtrip(server, hub): + key = hub.keys.issue("egor") + assert upload(server, key, b"line1\n") == (200, {"size": 6, "masked": 0}) + assert upload(server, key, b"line2\n", offset=6)[1]["size"] == 12 + assert post(server, "/v1/ingest/manifest", {}, key=key)[1] == {"files": {GOOD: 12}} + stored = storage.resolve(hub.transcripts, "egor", GOOD) + assert stored.read_bytes() == b"line1\nline2\n" + + +def test_uploads_land_in_the_sender_s_own_tree(server, hub): + """The owner comes from the key, so a member cannot write into another + member's history no matter what the body says.""" + egor, maxim = hub.keys.issue("egor"), hub.keys.issue("maxim") + upload(server, egor, b"egor\n") + upload(server, maxim, b"maxim\n") + assert storage.resolve(hub.transcripts, "egor", GOOD).read_bytes() == b"egor\n" + assert storage.resolve(hub.transcripts, "maxim", GOOD).read_bytes() == b"maxim\n" + + +def test_traversal_is_refused(server, hub): + key = hub.keys.issue("egor") + with pytest.raises(urllib.error.HTTPError) as raised: + upload(server, key, b"x", path="claude/../../../etc/passwd.jsonl") + assert raised.value.code == 400 + + +def test_offset_mismatch_tells_the_client_where_to_resume(server, hub): + key = hub.keys.issue("egor") + upload(server, key, b"line1\n") + with pytest.raises(urllib.error.HTTPError) as raised: + upload(server, key, b"late\n", offset=99) + assert raised.value.code == 409 + assert json.loads(raised.value.read())["size"] == 6 + + +def test_gzipped_bodies_are_accepted(server, hub): + key = hub.keys.issue("egor") + status, body = post(server, "/v1/ingest", + {"path": GOOD, "offset": 0, + "data": base64.b64encode(b"compressed\n").decode()}, + key=key, compress=True) + assert (status, body["size"]) == (200, 11) + + +def test_known_secrets_never_reach_the_disk(server, hub): + """The masking map is live at ingest, so a credential in a transcript is + replaced before the bytes are stored — not after someone notices.""" + SecretMap.build({"servers/NETCUP_PASSWORD": NETCUP}, + salt="test-salt").save(hub.secrets_path) + key = hub.keys.issue("egor") + line = json.dumps({"text": f"ssh with {NETCUP}"}).encode() + b"\n" + _, body = upload(server, key, line) + + stored = storage.resolve(hub.transcripts, "egor", GOOD).read_bytes() + assert NETCUP.encode() not in stored + assert b"${servers/NETCUP_PASSWORD}" in stored + assert body["masked"] == 1 + # The manifest counts the CLIENT's bytes, so the next tail lines up even + # though masking changed the stored length. + assert body["size"] == len(line) != len(stored) + + +def test_search_returns_ranked_anchors(server, hub): + key = hub.keys.issue("egor") + status, body = post(server, "/v1/search", + {"query": "how did we pin it", "k": 3}, key=key) + assert status == 200 + assert body["anchors"][0]["snippet"] == "we pinned the version" + assert body["anchors"][0]["when_human"] # humanised for the agent + assert body["degraded"] is None + + +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")) + assert raised.value.code == 400 + + +def test_oversized_body_is_refused(server, hub): + key = hub.keys.issue("egor") + request = urllib.request.Request( + server + "/v1/ingest", data=b"{}", + headers={"Authorization": f"Bearer {key}", + "Content-Length": str(64 * 1024 * 1024)}) + with pytest.raises(urllib.error.HTTPError) as raised: + urllib.request.urlopen(request, timeout=10) + assert raised.value.code == 400 diff --git a/tests/test_hub_masking.py b/tests/test_hub_masking.py new file mode 100644 index 0000000..762bc94 --- /dev/null +++ b/tests/test_hub_masking.py @@ -0,0 +1,160 @@ +"""Masking known secrets by Doppler name. + +Two properties matter more than any single substitution: nothing recoverable +is stored in the map, and ordinary words are never masked — a map that eats +the string "dev" would quietly mangle the whole team's history.""" + +import json + +from session_recall.hub.masking import SecretMap, collect_from_doppler, maskable + +NETCUP = "Xk39dmPQ7wLz2vRt" # long, mixed case + digits +ANTHROPIC = "sk-ant-api03-" + "A" * 40 + + +def build(**entries) -> SecretMap: + return SecretMap.build(entries, salt="test-salt") + + +def test_masks_a_known_value_with_its_variable_name(): + smap = build(**{"servers/NETCUP_PASSWORD": NETCUP}) + masked, hits = smap.mask(f"sshpass -p {NETCUP} ssh root@host") + assert masked == "sshpass -p ${servers/NETCUP_PASSWORD} ssh root@host" + assert hits == 1 + + +def test_masks_a_value_glued_to_a_flag(): + """The widest tokenisation swallows `--key=…` whole; a narrower alphabet + is what recovers the value itself.""" + smap = build(**{"session-recall/ANTHROPIC_API_KEY": ANTHROPIC}) + masked, hits = smap.mask(f"claude --key={ANTHROPIC} -p hi") + assert "${session-recall/ANTHROPIC_API_KEY}" in masked + assert ANTHROPIC not in masked and hits >= 1 + + +def test_masks_every_occurrence(): + smap = build(**{"servers/NETCUP_PASSWORD": NETCUP}) + masked, hits = smap.mask(f"{NETCUP} and again {NETCUP}") + assert NETCUP not in masked and hits == 2 + + +def test_masks_inside_json_which_is_how_transcripts_actually_look(): + smap = build(**{"servers/NETCUP_PASSWORD": NETCUP}) + line = json.dumps({"role": "user", "text": f"пароль {NETCUP} для ssh"}) + masked, _ = smap.mask(line) + assert NETCUP not in masked + assert json.loads(masked)["text"] == "пароль ${servers/NETCUP_PASSWORD} для ssh" + + +def test_the_map_stores_no_recoverable_secret(tmp_path): + smap = build(**{"servers/NETCUP_PASSWORD": NETCUP}) + path = tmp_path / "secrets.json" + smap.save(path) + assert NETCUP not in path.read_text() + assert SecretMap.load(path).mask(NETCUP)[1] == 1 + + +def test_short_and_label_like_values_are_never_masked(): + assert not maskable("DOPPLER_CONFIG", "dev") + assert not maskable("DOPPLER_PROJECT", "session-recall") + assert not maskable("SHORT", "abc123") # under the length bar + assert not maskable("PEM", "-----BEGIN KEY-----\nabc") # whitespace: not a token + + +def test_a_long_lowercase_word_needs_more_than_length(): + # The name gate is applied first, so these all use credential-shaped names + # and vary only the VALUE. + assert not maskable("API_KEY", "configuration") # 13 chars, one class + assert maskable("API_KEY", "configurationmanagement") # 23 chars, still one + assert maskable("API_KEY", "configuration7") # two classes + + +def test_low_entropy_values_do_not_eat_ordinary_text(): + smap = build(**{"servers/DEPLOY_TOKEN": "dev", + "servers/NETCUP_PASSWORD": NETCUP}) + masked, hits = smap.mask("deploying to dev with the dev config") + assert masked == "deploying to dev with the dev config" and hits == 0 + + +def test_an_empty_map_is_a_no_op(): + smap = SecretMap(salt="", labels={}) + assert smap.mask("anything at all") == ("anything at all", 0) + + +def test_salt_is_reused_so_refreshes_stay_comparable(): + first = SecretMap.build({"a/B": NETCUP}, salt="fixed") + second = SecretMap.build({"a/B": NETCUP}, salt=first.salt) + assert first.labels == second.labels + + +def test_collect_walks_projects_configs_and_secrets(): + calls = [] + + def runner(args): + calls.append(args) + if args[1] == "projects": + return json.dumps([{"id": "servers"}, {"id": "locked"}]) + if args[1] == "configs": + if args[-1] == "locked": + raise RuntimeError("403") # a project the token cannot read + return json.dumps([{"name": "dev"}]) + return json.dumps({"NETCUP_PASSWORD": NETCUP, "PORT": 8080}) + + entries = collect_from_doppler(runner=runner) + assert entries == {"servers/NETCUP_PASSWORD": NETCUP} # non-string PORT dropped + assert any("locked" in " ".join(c) for c in calls) # tried, then skipped + + +def test_configuration_values_are_not_credentials(): + """The first real corpus masked 851 mentions of a model name and 2429 of a + hostname. Shape cannot separate `claude-opus-4` from a token — the variable + NAME can.""" + for name in ("CLAUDE_MODEL", "EMBED_MODEL", "LLM_MODEL", + "URBANTECH_GPU_HOSTNAME", "SERVER_IP", "R2_BUCKET", + "AWS_REGION", "POSTGRES_HOST"): + assert not maskable(name, "claude-opus-4-20260101"), name + + +def test_real_credential_names_still_mask(): + for name in ("NETCUP_PASSWORD", "VOYAGE_API_KEY", "JWT_SECRET", + "TG_APPROVAL_BOT_TOKEN", "R2_ACCESS_KEY_ID", + "POSTGRES_PASSWORD", "SIGNING_KEY"): + assert maskable(name, "Xk39dmPQ7wLz2vRt"), name + + +def test_model_names_survive_masking_end_to_end(): + smap = build(**{"chloe/CLAUDE_MODEL": "claude-opus-4-20260101", + "servers/NETCUP_PASSWORD": NETCUP}) + text = f"взяли claude-opus-4-20260101 и пароль {NETCUP}" + masked, hits = smap.mask(text) + assert "claude-opus-4-20260101" in masked # configuration stays readable + assert NETCUP not in masked and hits == 1 + + +def test_secret_right_after_an_escaped_newline_is_masked(): + """Found in production: transcripts are JSON, so a newline is the two + characters `\\` and `n`. The backslash ends a token but the `n` glues onto + what follows, so a key written after a line break tokenised as `n` + and matched nothing — one real API key survived masking this way.""" + smap = build(**{"session-recall/VOYAGE_API_KEY": ANTHROPIC}) + raw = json.dumps({"text": f"замени в doppler\n\n{ANTHROPIC}"}) + assert "\\n\\n" + ANTHROPIC in raw # the shape that leaked + masked, hits = smap.mask(raw) + assert ANTHROPIC not in masked and hits == 1 + assert "${session-recall/VOYAGE_API_KEY}" in masked + + +def test_other_json_escapes_glue_the_same_way(): + smap = build(**{"a/API_KEY": ANTHROPIC}) + for escape in ("\\t", "\\r", "\\b", "\\f"): + masked, hits = smap.mask(f'"{escape}{ANTHROPIC}"') + assert ANTHROPIC not in masked, escape + assert hits == 1, escape + + +def test_a_real_word_before_a_secret_is_not_stripped(): + """Only an escape letter after a backslash is peeled — an ordinary prefix + must not be, or the map would start matching things it should not.""" + smap = build(**{"a/API_KEY": ANTHROPIC}) + masked, hits = smap.mask(f"prefix{ANTHROPIC}") + assert hits == 0 and masked == f"prefix{ANTHROPIC}" diff --git a/tests/test_hub_storage.py b/tests/test_hub_storage.py new file mode 100644 index 0000000..61e74b7 --- /dev/null +++ b/tests/test_hub_storage.py @@ -0,0 +1,104 @@ +"""Ingest-side storage: paths arrive over the network, so `resolve` is a +security boundary and gets adversarial tests, not happy-path ones.""" + +import pytest + +from session_recall.hub import storage +from session_recall.hub.storage import Ledger, OffsetMismatch, UnsafePath + +GOOD = "claude/-Users-egor-proj/1234-abcd.jsonl" + + +@pytest.fixture +def root(tmp_path): + return tmp_path / "transcripts" + + +@pytest.fixture +def ledger(tmp_path): + return Ledger(tmp_path / "state") + + +def test_resolve_accepts_a_normal_transcript(root): + path = storage.resolve(root, "egor", GOOD) + assert path == root / "egor" / "claude" / "-Users-egor-proj" / "1234-abcd.jsonl" + + +@pytest.mark.parametrize("rel", [ + "claude/../../etc/passwd.jsonl", # classic traversal + "../maxim/claude/x.jsonl", # into another member's tree + "/etc/shadow.jsonl", # absolute + "claude\\..\\x.jsonl", # windows-style separator + "claude/x.jsonl\0.txt", # NUL truncation + "claude/.ssh/id_rsa.jsonl", # dot-leading segment + "secrets/x.jsonl", # unknown source root + "claude/proj/notes.txt", # not a transcript + "claude", # too shallow + "claude/" + "a/" * 10 + "x.jsonl", # too deep +]) +def test_resolve_rejects_hostile_paths(root, rel): + with pytest.raises(UnsafePath): + storage.resolve(root, "egor", rel) + + +def test_resolve_rejects_bad_owner(root): + with pytest.raises(UnsafePath): + storage.resolve(root, "../maxim", GOOD) + + +def test_resolve_refuses_to_follow_a_planted_symlink(root, tmp_path): + """Pattern matching cannot see a symlink, so the resolved path is checked + against the resolved owner root as an independent second gate.""" + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir(parents=True) + (root / "egor").mkdir(parents=True) + (root / "egor" / "claude").symlink_to(elsewhere, target_is_directory=True) + with pytest.raises(UnsafePath): + storage.resolve(root, "egor", GOOD) + + +def test_append_then_extend(root, ledger): + assert storage.append(root, "egor", GOOD, 0, b"line1\n", ledger) == 6 + assert storage.append(root, "egor", GOOD, 6, b"line2\n", ledger) == 12 + assert storage.resolve(root, "egor", GOOD).read_bytes() == b"line1\nline2\n" + + +def test_append_at_wrong_offset_reports_what_we_have(root, ledger): + storage.append(root, "egor", GOOD, 0, b"line1\n", ledger) + with pytest.raises(OffsetMismatch) as raised: + storage.append(root, "egor", GOOD, 99, b"late\n", ledger) + assert raised.value.actual == 6 + + +def test_offset_zero_replaces_a_rewritten_transcript(root, ledger): + storage.append(root, "egor", GOOD, 0, b"old content\n", ledger) + assert storage.append(root, "egor", GOOD, 0, b"new\n", ledger) == 4 + assert storage.resolve(root, "egor", GOOD).read_bytes() == b"new\n" + + +def test_ledger_counts_client_bytes_not_stored_bytes(root, ledger): + """Masking rewrites the text, so the manifest must track how much of the + CLIENT's file we consumed — otherwise every masked transcript looks + truncated and uploads forever.""" + storage.append(root, "egor", GOOD, 0, b"short\n", ledger, received_len=500) + assert storage.manifest(root, "egor", ledger) == {GOOD: 500} + + +def test_manifest_drops_entries_whose_file_is_gone(root, ledger): + storage.append(root, "egor", GOOD, 0, b"x\n", ledger) + storage.resolve(root, "egor", GOOD).unlink() + assert storage.manifest(root, "egor", ledger) == {} + + +def test_manifest_is_per_owner(root, ledger): + storage.append(root, "egor", GOOD, 0, b"x\n", ledger) + assert storage.manifest(root, "maxim", ledger) == {} + + +def test_owners_lists_only_valid_member_directories(root): + for name in ("egor", "maxim"): + (root / name).mkdir(parents=True) + (root / "NOT A NAME").mkdir() + (root / "loose.jsonl").parent.mkdir(exist_ok=True) + (root / "loose.jsonl").write_text("") + assert storage.owners(root) == ["egor", "maxim"] diff --git a/tests/test_share_compose.py b/tests/test_share_compose.py index 81cc0c0..ef75bfd 100644 --- a/tests/test_share_compose.py +++ b/tests/test_share_compose.py @@ -147,3 +147,102 @@ def test_no_chunks_means_no_call(): client = FakeClient(FakeResponse([FakeBlock("hallucinated")])) assert _composer(client)(REQ, []) is None assert client.beta.messages.create.__self__.calls == [] + + +# -- codex CLI composer ------------------------------------------------------ +class FakeRun: + """Stands in for `codex exec`: records argv and writes the answer where + the real CLI would (-o), while leaving noise on stdout like it does.""" + + def __init__(self, answer="вот как чинили: запинили mcp<2", returncode=0): + self.answer, self.returncode = answer, returncode + self.calls, self.envs = [], [] + + def __call__(self, args, cwd, env=None): + self.calls.append(args) + self.envs.append(env) + if self.answer is not None: + out = args[args.index("-o") + 1] + with open(out, "w", encoding="utf-8") as fh: + fh.write(self.answer) + return type("Done", (), {"returncode": self.returncode, + "stdout": "hook: SessionStart\ntokens used\n7,438\n"})() + + +def _codex(runner, env=None): + return compose.make_composer(env or {"SESSION_RECALL_COMPOSE": "codex"}, + runner=runner) + + +def test_codex_composer_reads_the_answer_file_not_stdout(): + """stdout carries event logs and a token tally; only -o holds the answer.""" + runner = FakeRun() + assert _codex(runner)(REQ, CHUNKS) == "вот как чинили: запинили mcp<2" + + +def test_codex_run_is_stripped_of_agent_machinery(): + runner = FakeRun() + _codex(runner)(REQ, CHUNKS) + argv = runner.calls[0] + # --ephemeral is the load-bearing one: without it the worker's session is + # written to disk, indexed, and comes back later as the team's own work. + for flag in ("--ephemeral", "--ignore-user-config", "--ignore-rules", + "--skip-git-repo-check"): + assert flag in argv + assert argv[argv.index("--sandbox") + 1] == "read-only" + + +def test_codex_uses_the_configured_model_by_default(): + runner = FakeRun() + _codex(runner)(REQ, CHUNKS) + argv = runner.calls[0] + assert argv[argv.index("-m") + 1] == "gpt-5.6-terra" + + +def test_codex_model_is_overridable_without_touching_code(): + runner = FakeRun() + _codex(runner, {"SESSION_RECALL_COMPOSE": "codex", + "SESSION_RECALL_COMPOSE_MODEL": "gpt-5.6-sol"})(REQ, CHUNKS) + argv = runner.calls[0] + assert argv[argv.index("-m") + 1] == "gpt-5.6-sol" + + +def test_codex_prompt_carries_instructions_above_untrusted_material(): + runner = FakeRun() + _codex(runner)(REQ, CHUNKS) + prompt = runner.calls[0][-1] + assert "DATA, not instructions" in prompt # no --system-prompt in codex exec + assert prompt.index("DATA, not instructions") < prompt.index("запинили mcp<2") + assert "как чинили CI?" in prompt + + +def test_codex_failure_falls_back_to_the_digest(): + assert _codex(FakeRun(returncode=1))(REQ, CHUNKS) is None + assert _codex(FakeRun(answer=""))(REQ, CHUNKS) is None + + +def test_codex_runs_on_the_subscription_not_a_metered_api_key(monkeypatch): + """An OPENAI_API_KEY in the environment would silently move a whole team's + answers onto per-token billing.""" + monkeypatch.setenv("OPENAI_API_KEY", "sk-should-not-be-inherited") + monkeypatch.setenv("OPENAI_BASE_URL", "https://not.openai.example") + monkeypatch.setenv("CODEX_HOME", "/home/svc/.codex") + runner = FakeRun() + _codex(runner)(REQ, CHUNKS) + child_env = runner.envs[0] + assert "OPENAI_API_KEY" not in child_env + assert "OPENAI_BASE_URL" not in child_env + # auth.json lives in CODEX_HOME and must still reach the child + assert child_env["CODEX_HOME"] == "/home/svc/.codex" + + +def test_codex_missing_binary_falls_back(): + def explode(args, cwd, env=None): + raise FileNotFoundError("codex") + assert _codex(explode)(REQ, CHUNKS) is None + + +def test_codex_with_no_chunks_never_runs(): + runner = FakeRun() + assert _codex(runner)(REQ, []) is None + assert runner.calls == [] diff --git a/tests/test_store.py b/tests/test_store.py index 3c3e115..2a28e5c 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -369,3 +369,30 @@ def chunk(uuid, session_id, project, ts, source="claude"): assert out["span_days"] == 2 assert out["top_projects"][0] == ("proj-a", 2), "busiest project first, with its chunk count" s.close() + + +def test_read_only_store_searches_but_never_writes(tmp_path): + """A hub's service reads while its indexer writes. A read-write connection + from the reader made sqlite-vec drop the writer's vectors ("could not write + to vector blob"), losing 63 transcripts in one real run.""" + import pytest + from session_recall.config import EMBED_DIM + + path = tmp_path / "index.db" + writer = Store(path) + chunk = Chunk(session_id="s1", uuid="u1", role="user", text="doppler маскировка", + project="p", cwd="/tmp/p", git_branch="main", ts=1785000000, + file_path=str(tmp_path / "t.jsonl"), byte_offset=0, byte_len=10, + turn_index=0, content_hash="h1", source="claude") + writer.add(chunk, [0.01] * EMBED_DIM) + writer.commit() + writer.close() + + reader = Store(path, read_only=True) + assert reader.knn([0.01] * EMBED_DIM, 3) # vector search works + assert reader.fts("doppler", 3) # keyword search works + assert reader.get_chunk(1).text == "doppler маскировка" + with pytest.raises(Exception): # and writes are refused + reader.add(chunk, [0.02] * EMBED_DIM) + reader.commit() + reader.db.close()