diff --git a/docs/agent.md b/docs/agent.md index b9d2048..d088e2a 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -162,7 +162,7 @@ ## Workers -Run sequentially after `StreamEnd`. Currently: `CompressionWorker`. Workers receive a `WorkerContext` carrying the active `profile`, so `CompressionWorker` applies per-profile compression overrides (`compression_keep_recent`, `compression_max_tokens`, `compression_prompt_file`). The worker gates on the real `context_tokens` from the last LLM call, then delegates to `compress_and_save_session(reason="postturn")` — the same retry / hard-truncate / safety-net / archiving pipeline as the pre- and mid-turn paths (with the mid-turn `keep_recent_messages`). It compresses unconditionally when `WorkerContext.context_truncated` is set: a truncated prompt reports only about half the window, so the threshold would never fire on it. +Run sequentially after `StreamEnd`. Currently: `CompressionWorker`. Workers receive a `WorkerContext` carrying the active `profile`, so `CompressionWorker` applies per-profile compression overrides (`compression_keep_recent`, `compression_max_tokens`, `compression_prompt_file`), and the `memory_store`, so the messages the pass demotes are indexed into `memory_chunks` (see [`memory.md`](memory.md)). The worker gates on the real `context_tokens` from the last LLM call, then delegates to `compress_and_save_session(reason="postturn")` — the same retry / hard-truncate / safety-net / archiving pipeline as the pre- and mid-turn paths (with the mid-turn `keep_recent_messages`). It compresses unconditionally when `WorkerContext.context_truncated` is set: a truncated prompt reports only about half the window, so the threshold would never fire on it. Pre-turn compression also runs at the start of `run_stream()`: it estimates tokens via `real_baseline_estimate(session.context, session.context)` (real `prompt_eval_count` from the last call + calibrated delta; the calibrated `estimate_tokens` only before the first LLM call of a session) plus the incoming user message — images included, since it is not in `session.context` yet — and, when over `window × threshold`, is guarded by `would_compress(current_tokens=…)` before emitting `CompressionStarted`. The window is `profile.context_window` or `settings.ollama_num_ctx` (`Agent._window_for`). See [`sessions.md`](sessions.md). @@ -172,8 +172,9 @@ Every LLM call receives: 1. System message: `persona + "---" + profile.system_prompt` (injected fresh, never stored). -2. Optional memory message: `"## What I remember about the user\n..."`. -3. `session.context` messages (system messages stripped to avoid duplication). +2. Optional memory message: `"## What I remember about the user\n..."`, plus the per-turn list of individual facts. +3. Optional `"## Recovered from earlier in this conversation"` message — up to two raw excerpts that `memory_search` matched to the current topic, only when the context holds a summary (`memory_recall_enabled`; delivered chunks deduped via `session_metadata["injected_chunk_ids"]`). See [`memory.md`](memory.md#automatic-recall). +4. `session.context` messages (system messages stripped to avoid duplication). Profile switches and persona changes take effect immediately. diff --git a/docs/config.md b/docs/config.md index 8f4fa3b..a807f3f 100644 --- a/docs/config.md +++ b/docs/config.md @@ -355,6 +355,17 @@ | `CONTEXT_MESSAGE_TOKEN_BUDGET` | int | `0` | Per-message token budget for the LLM context view. A single `tool`/`assistant` message whose estimated size exceeds this is head/tail-truncated in the *built* context only (stored history is never mutated) so one huge tool result cannot alone blow the window. `user` messages are capped too, at least at `OLLAMA_NUM_CTX // 3`, so an oversized paste cannot dead-end the session. `0` = auto (`OLLAMA_NUM_CTX // 6`). | | `CONTEXT_IMAGE_TOKENS` | int | `1024` | Tokens charged for one inline image whose dimensions cannot be read from its header. Images with a readable header are priced by area — roughly one token per 32×32 px, measured on qwen3.6 (1024×1024 ≈ 1026 tokens, 512×512 ≈ 258) — so the session total, the compression trigger and the per-turn image budget all see what the model actually gets. See `navi/core/image_tokens.py`. | +## Memory search and recall + +| Variable | Type | Default | Description | +|---|---|---|---| +| `MEMORY_SEARCH_ENABLED` | bool | `true` | Index the messages that context compression drops into `memory_chunks` (verbatim, append-only) with breadcrumbs, so `memory_search` can find them later | +| `MEMORY_RECALL_ENABLED` | bool | `true` | Automatically inject relevant raw excerpts when the context holds a summary | +| `MEMORY_RECALL_K` | int | `2` | How many raw excerpts may be injected per turn | +| `MEMORY_RECALL_MIN_SIMILARITY` | float | `0.55` | Vector-similarity floor for an injected excerpt (the FTS fallback requires 2 matched terms instead) | + +See [`docs/memory.md`](memory.md) for the three stores, the tiered merge and the freshness rules. + ## Gmail | Variable | Type | Default | Description | diff --git a/docs/mechanics.md b/docs/mechanics.md index 852e0c6..dd0c912 100644 --- a/docs/mechanics.md +++ b/docs/mechanics.md @@ -175,6 +175,7 @@ | **CodeExecTool** | Run Python in subprocess sandbox. Non-admin confined to `user_data//` or the session directory. | None | `code_exec.py` | ✅ | | **ImageViewTool** | Load image from path/URL → resize to 1024px, JPEG, return base64 for LLM. | None | `image_view.py` | ✅ | | **MemoryTool** | Save/search/forget/list user facts. Dual search: semantic (cosine) + ILIKE fallback. | None | `memory.py` | ✅ | +| **MemorySearchTool** | One query over facts + raw chunks dropped by compression + session sketches. Tiered merge, provenance, freshness labels. | `MEMORY_SEARCH_ENABLED`, `MEMORY_RECALL_*` | `memory_search.py` | ✅ | | **TodoTool** | Session-scoped task tracker. Set/view/update/clear. Auto-populated from planning. | None | `todo.py` | ✅ | | **ScratchpadTool** | Session-scoped working notes. Write/append/read/clear per section. | None | `scratchpad.py` | ✅ | | **SpawnAgentTool** | Spawn isolated subagent with own tool loop. Supports `inherit_system_prompt`, `briefing`, `profile_id`. | None | `spawn_agent.py` | ✅ | diff --git a/docs/memory.md b/docs/memory.md index baf52a8..89bce1b 100644 --- a/docs/memory.md +++ b/docs/memory.md @@ -73,16 +73,19 @@ ## Storage (`navi/memory/store.py`) -Three tables in the database: +Four tables in the database: | Table | Purpose | |---|---| | `memory_facts` | Individual facts: `(user_id, category, key, value)` — unique on `(user_id, category, key)` | | `memory_summary` | Per-user narrative summary (`user_id` scoped) | | `session_memory_state` | Tracks which sessions have been processed (by `extracted_at`) | +| `memory_chunks` | Verbatim messages dropped by context compression — append-only, unique on `(session_id, content_hash)` | `user_id` references `navi_users(id)` with `ON DELETE CASCADE`. Facts and summaries are scoped per user. Admin with `navi.memory.read_all` can pass `user_id=None` for global search. +`memory_chunks` deliberately has **no foreign key** on `session_id`: the memory DDL may run before `sessions` exists on a fresh database. Rows are deleted by `PgSessionStore.delete()` instead. + `merged_into` is a nullable pointer to the row that superseded this one — the extractor writes the same fact under a new spelling of the key, and the older row is hidden rather than deleted, so the merge can be undone with `UPDATE memory_facts SET merged_into = NULL`. Every read path filters `merged_into IS NULL`; `get_all_facts(include_merged=True)` and `fact_count(include_merged=True)` are the only ways to see a hidden row, and `GET /admin/memory?include_merged=true` exposes them. `MemoryStore` lazily creates tables on first async operation via `_get_pool()`. All operations are async via asyncpg (PostgreSQL). @@ -100,6 +103,11 @@ | `mark_session_extracted(session_id)` | Record extraction timestamp | | `get_extracted_at(session_id)` | Check if/when a session was processed | | `backfill_embeddings(batch_size=8)` | Generate embeddings for facts with `embedding IS NULL` (hidden rows skipped) | +| `add_chunks(chunks, user_id=None, session_id=None)` | Write raw chunks (`memory_chunks`, `ChunkMixin`). Batched embeddings, one `INSERT … ON CONFLICT (session_id, content_hash) DO NOTHING` per row, returns the number actually written | +| `search_chunks(query, user_id=None, session_id=None, limit=8)` | **Vector search first** (cosine distance < 0.45), then Postgres FTS (`plainto_tsquery` on `content`, plus `ILIKE ANY` on content and `breadcrumbs`) with Python `terms_hit` scoring. An empty normalized query returns the most recent chunks | +| `latest_summary_at(session_id)` | Timestamp of the session's last summary — the freshness boundary for raw excerpts. Failures return `None` | + +`KvStore.search_values(user_id, scope, terms, limit=10, session_id=None, prefer_session_id=None)` (`navi/store/__init__.py`) is the read path for scratchpad sketches: ILIKE against `key` and `value` inside a scope, current session ranked first. --- @@ -149,6 +157,76 @@ --- +## Raw chunks and breadcrumbs (`navi/memory/_chunks.py`, `navi/memory/breadcrumbs.py`) + +Every context compression is a lossy step: the messages it drops are gone from the model's window forever. `memory_chunks` keeps them verbatim. + +### What is written + +Compression collects the messages it demotes (`demoted` — exactly the rows it flips to `is_context=False`) and hands them to `build_chunks`: + +- messages are packed into chunks of up to `CHUNK_MAX_CHARS` (4000) chars; a single message is capped at `CHUNK_MESSAGE_MAX_CHARS` (12 000) with its middle elided; +- `user`, `assistant` and `tool` messages are included, each prefixed (`User:`, `Assistant:`, `Tool result (name):`); `system` messages, old summaries and compression markers are skipped; +- each chunk records `start_seq`/`end_seq`, `started_at`/`ended_at`, and a `sha256` of its content. + +The table is **append-only and immutable**: a correction is a new chunk, never an edit. `(session_id, content_hash)` is unique, so indexing the same demoted range twice is a no-op. + +### Breadcrumbs + +The breadcrumbs are the entities a later question will be phrased with — a file, a command, a port, an amount, a ticket, a link — and they are searched alongside the chunk text. Two sources, merged: + +1. **The summarizer's `KEYS:` line.** The summary prompt ends with *"End with ONE line `KEYS: item, item, …` — the breadcrumbs of this history … at most 30, most important first. … write `KEYS: NONE` when nothing is worth indexing."* `split_keys_line` removes that line from the summary text and returns the keys; the summary message keeps the rest. +2. **A deterministic extractor** (`extract_breadcrumbs`): URLs, inline code, paths, `SNAKE_CASE`, `CamelCase` and `CONSTANT` identifiers, amounts (`4,5 GB`, `12 ms`, `80%`), `#ticket` numbers and quoted phrases. It runs on the dropped messages themselves, so it also covers the hard-truncate fallback — the path that never calls a summarizer and therefore never produces a `KEYS:` line. + +The union (`MAX_BREADCRUMBS` = 40) is attached to every chunk produced by that compression. Indexing failures are logged (`compressor.chunk_index_failed`) and never fail the compression. + +Indexes: `(user_id, session_id, ended_at DESC)`, an expression GIN on `to_tsvector('simple', content)`, a GIN on `breadcrumbs`, and an HNSW index on `embedding` when pgvector is available. + +--- + +## Unified search (`navi/memory/search.py`) + +Three stores, one read path. Writing is deliberately separate; reading is not — an agent asked "what did we decide about the port" should not have to know which store holds the answer. + +| Store | Shape | Written by | +|---|---|---| +| `facts` | curated, rewritten by key, user lifetime | extractor, `memory` tool | +| `raw` | verbatim messages dropped by compression | `compress_and_save_session` | +| `sketches` | a session's working notes, updated in place | `ScratchpadTool` (`session_store` scope `scratchpad`) | + +`UnifiedMemorySearch.search(query, scope="all", k=5, user_id=None, session_id=None, current_session_id=None, summary_at=None)` returns `MemoryHit` objects. The merge is tiered, not a blind mix: + +1. **facts** — curated truth first; +2. **raw** — by relevance (vector similarity, else matched terms); +3. **sketches** — the current session's topic pad last. + +`scope` narrows to one store (`facts` / `raw` / `sketches`); `all` is the default and an unknown value falls back to it. `k` is per store, clamped to 1–20. + +Every hit carries provenance: `source_type`, `session_id`, `timestamp`, `key`/`category` for a fact, `start_seq`/`end_seq` for a raw excerpt, section for a sketch. + +**Freshness.** A raw excerpt is *reference*, a fact is *curated*. `_mark_freshness` compares them: a raw chunk that shares at least two content words with a **newer** fact is marked `superseded_by=` and rendered with a warning; a raw chunk older than the same session's last summary is marked `before_summary` and rendered as a verbatim record from before the current summary. The outdated item is still returned — "we changed our mind, the old value was X" is often exactly the question. + +`format_hits` renders the tool output, e.g.: + +``` +memory_search "server port" (scope=all): 2 hit(s) + +1. fact [technical] · session s1 · updated 2026-11-09 12:00 + server_port: 8099 +2. raw · session s1 · turn 12–14 · 2026-11-06 18:41 + ⚠ outdated: a newer curated fact overrides this + keys: server_port, 8098 + User: ...verbatim... +``` + +### Automatic recall + +`ContextBuilder._memory_recall_msg` injects up to `memory_recall_k` (2) raw excerpts **once per turn, only when the context already holds a summary** — that is the moment when dropped history matters. The query is built from the last four user/assistant messages plus the incoming one, so the search follows the current topic. Hits below `memory_recall_min_similarity` (0.55 vector, or 2 matched terms on the FTS path) and hits marked `superseded_by` are dropped. Delivered chunk ids are remembered in `session.session_metadata["injected_chunk_ids"]`, so the same excerpt does not ride in every prompt; the model can still call `memory_search` for anything wider. + +Both behaviours are switchable: `MEMORY_SEARCH_ENABLED` (indexing demoted messages) and `MEMORY_RECALL_ENABLED` (the automatic injection), see `docs/config.md`. + +--- + ## Memory tool (`navi/tools/memory.py`) Single unified `memory` tool with `action` parameter: @@ -166,13 +244,34 @@ --- +## Search tool (`navi/tools/memory_search.py`) + +`memory_search` is the read side of memory and the only tool that reads all three stores. + +| Parameter | Description | +|---|---| +| `query` (required) | What to look for — a name, a file, a command, a value, a topic | +| `scope` | `all` (default), `facts`, `raw`, `sketches` | +| `k` | Results per store, 1–20, default 5 | + +There is deliberately **no separate session-search tool**: one query over everything, tiered merge, provenance on every line. The tool resolves `session_id`/`user_id` from `ToolContext` (falling back to the request context vars) and asks `latest_summary_at(session_id)` for the freshness boundary. + +### `memory` vs `memory_search` + +- **`memory`** writes. `save` / `forget` / `list`, and `search` for a fact-only lookup when the agent specifically wants curated facts. +- **`memory_search`** reads across everything. Use it when the user refers back to something no longer in view — "as we discussed", "the port we picked", "that file", a name, a number, a link, a decision — before saying you do not remember. + +--- + ## Memory usage guidelines (from persona) Call `memory_search` when: -- The user mentions something personal (location, project, preference, recurring task). +- The user refers back to something you can no longer see — "as we discussed", "the port we picked", "that file", a name, a number, a link, a decision — before saying you do not remember. +- The user mentions something personal (location, project, preference, recurring task) and you need the fact behind it. - About to make an assumption about the user's environment or preferences — verify first. -- The user asks about something helped with before. -Do NOT call `memory_search` reflexively at the start of every session — only when context warrants it. +Prefer a newer curated fact over an older raw excerpt; when they disagree, `memory_search` says which one is outdated. -Call `memory_forget` only when the user explicitly asks, or when a stored fact is clearly wrong or outdated. +Do NOT call `memory_search` reflexively at the start of every session — only when context warrants it (automatic recall already covers the compressed-history case). + +Call `memory` with `action="forget"` only when the user explicitly asks, or when a stored fact is clearly wrong or outdated. diff --git a/docs/navi_code.md b/docs/navi_code.md index 227de88..3b6ec9c 100644 --- a/docs/navi_code.md +++ b/docs/navi_code.md @@ -65,7 +65,7 @@ - Расположение: `navi/profiles/navi_code/`. - База: `developer`, адаптирован под терминал. - Включённые инструменты: - - Native: `terminal`, `filesystem`, `code_exec`, `spawn_agent`, `todo`, `scratchpad`, `reflect`, `plan`, `image_view`, `ssh_exec`, `list_tools`, `tool_manual`, `switch_profile`, `list_profiles`, `memory`, `schedule_recall`, `manage_recall`. + - Native: `terminal`, `filesystem`, `code_exec`, `spawn_agent`, `todo`, `scratchpad`, `reflect`, `plan`, `image_view`, `ssh_exec`, `list_tools`, `tool_manual`, `switch_profile`, `list_profiles`, `memory`, `memory_search`, `schedule_recall`, `manage_recall`. - MCP: `navi-web` (`search`, `browse`, `request`) — веб-поиск и просмотр страниц. - Отключённые инструменты: `share_file`, `content_publish`, `gmail`. - Планирование: фазы 1 и 3 включены, вызываются агентом через инструмент `plan` — нет обязательного пре-тарн гейта. Для комплексных задач (COMPLEXITY=complex) агент излагает план и ждёт подтверждения. diff --git a/docs/profiles.md b/docs/profiles.md index 4762952..36e3b3d 100644 --- a/docs/profiles.md +++ b/docs/profiles.md @@ -216,7 +216,7 @@ Sub-agents are narrower than the agent, and for `designer_3d` they are now exactly what its prompt promises them — web search, page reading, filesystem reads, image viewing: -`scratchpad`, `reflect`, `filesystem`, `image_view`, `memory`, `list_tools`, `tool_manual` +`scratchpad`, `reflect`, `filesystem`, `image_view`, `memory`, `memory_search`, `list_tools`, `tool_manual` plus `navi-web` (`search`, `browse`). Until 2026-10-12 that set was the general one: it carried `terminal`, `code_exec`, `share_file`, `content_publish`, `todo` and the whole `navi-3d` server — running commands, publishing and the modelling tools its prompt forbids — @@ -316,7 +316,7 @@ Terminal-first local coding assistant. Designed for the Navi Code CLI and single-user local deployments: -- **Native tools:** `todo`, `scratchpad`, `reflect`, `plan`, `switch_profile`, `list_profiles`, `filesystem`, `code_exec`, `terminal`, `image_view`, `ssh_exec`, `memory`, `list_tools`, `tool_manual`, `spawn_agent`, `schedule_recall`, `manage_recall`. +- **Native tools:** `todo`, `scratchpad`, `reflect`, `plan`, `switch_profile`, `list_profiles`, `filesystem`, `code_exec`, `terminal`, `image_view`, `ssh_exec`, `memory`, `memory_search`, `list_tools`, `tool_manual`, `spawn_agent`, `schedule_recall`, `manage_recall`. - **MCP tools:** `navi-web` (`search`, `browse`, `request`) — web lookup and page browsing. - **Excluded:** `share_file`, `content_publish`, `gmail`. - **Planning:** Phase 1 and Phase 3 enabled; the `plan` tool is in the native tool set, so the agent plans deliberately when a task warrants it. Phase 1 classifies `COMPLEXITY: simple | medium | complex` — complex tasks get "present the plan and wait for confirmation". diff --git a/docs/sessions.md b/docs/sessions.md index 0171fcc..b9ab903 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -154,8 +154,8 @@ 2. **Meta-summary:** if `to_summarize` contains multiple existing summaries totaling > `_META_SUMMARY_THRESHOLD ≈ 10_600` chars, consolidate them into one via `_meta_summarize` first so old summaries don't crowd the summarizer input. 3. Format `to_summarize` as plain text. Tool calls are shown as compact previews (max 120 chars for args). **Critical** tool results (`is_compression_critical=True` or critical tool names) survive verbatim up to 4000 chars, then head+tail halves (2000+2000) instead of collapsing to the non-critical preview; non-critical results are capped at an 800-char preview. Base64 images are collected for vision models up to an image budget (`window × 0.25`, priced per image); images over it are left out of the call, while their `[+ N image(s)]` note still tells the summarizer they existed. Without the cap, a dozen screenshots made a summarization prompt that could not fit its own window, was silently truncated, and produced a summary written as if the history had been read. 4. Truncate formatted input to `_MAX_SUMMARY_INPUT_CHARS = 32_000` chars, keeping head (75%, oldest messages) + tail (25%, the messages closest to the kept window) — a head-only cut used to silently drop the newest summarized work. -5. Call `llm.complete()` with `think=False` to produce a bullet-point summary. -6. Replace `to_summarize` with a single summary message (`role=user`, `is_summary=True`). +5. Call `llm.complete()` with `think=False` to produce a bullet-point summary. The summary prompt asks the model to end with one `KEYS: item, item, …` line — the breadcrumbs of the compressed history (names, commands, paths, config keys, amounts, tickets, decisions, links). `split_keys_line` removes that line from the summary text and returns the keys; a summary without it comes back unchanged. +6. Replace `to_summarize` with a single summary message (`role=user`, `is_summary=True`), carrying the keys in `metadata["breadcrumbs"]`. 7. Return `system_msgs + [summary_msg] + to_keep`. `compress_session` wraps `compress_context` with retry + a **token-budget hard-truncate fallback** (`_hard_truncate`): if the LLM summarization fails twice, it drops oldest turns (keeping system + newest whole turns) until under `_HARD_TRUNCATE_TOKEN_FRAC = 0.5` of the window — a last resort with no LLM call. @@ -172,6 +172,14 @@ - The last `context_keep_recent` conversational turns (or the intra-turn recent window). - System messages (never stored in context anyway). +### Raw history index + +Compression removes messages from the model's window — it does not have to lose them. On every path through `compress_and_save_session` (pre-turn, mid-turn, post-turn, forced `/compact`) the exact messages that were demoted are also written to `memory_chunks` as append-only verbatim chunks with their turn range, timestamps and breadcrumbs (`navi/memory/_chunks.py`, `navi/memory/breadcrumbs.py`; see `docs/memory.md`). Breadcrumbs are the summary's `KEYS:` line merged with entities extracted from the dropped text, which also covers the hard-truncate fallback that never calls a summarizer. + +The index is idempotent via `UNIQUE(session_id, content_hash)` and failures never fail a compression. `PgSessionStore.delete()` removes a session's chunks along with the session. Indexing is gated by `memory_search_enabled`. + +The agent then reaches the dropped history through `memory_search(scope="raw")` — or automatically: when the context holds a summary, the agent injects up to two relevant excerpts per turn (`memory_recall_enabled`, `memory_recall_k`, `memory_recall_min_similarity`) and remembers what it has already injected in `session_metadata["injected_chunk_ids"]`. + ### Events - `CompressionStarted` — status signal (`context_tokens`, `max_context_tokens`). diff --git a/docs/store.md b/docs/store.md index 61c999b..6e3522a 100644 --- a/docs/store.md +++ b/docs/store.md @@ -44,6 +44,7 @@ await store.get_all(user_id, session_id, scope) # dict[str, str] await store.delete(user_id, session_id, scope, key) # None await store.clear_scope(user_id, session_id, scope) # None +await store.search_values(user_id, scope, terms, limit=10) # list[dict] ``` - `user_id` — nullable for legacy single-user mode. @@ -51,6 +52,8 @@ - `scope` — namespace: `scratchpad`, or any custom string. - `key` — arbitrary string identifier within the scope. +`search_values(user_id, scope, terms, limit=10, session_id=None, prefer_session_id=None)` is the read path behind `memory_search(scope="sketches")`: ILIKE against key and value inside a scope, the preferred session ranked first and then newest first; no terms returns the most recent rows. + ## Wiring `KvStore` is created as a singleton in `navi/core/container.py` (`get_kv_store()` exposes it to routes) and injected into `build_default_registries()`, which passes it to `ScratchpadTool` on startup. `TodoTool` gets the dedicated `PgTodoStore` from the same container. diff --git a/docs/tools.md b/docs/tools.md index cffcf47..9f45999 100644 --- a/docs/tools.md +++ b/docs/tools.md @@ -50,6 +50,7 @@ | `ListToolsTool` | `list_tools` | Tools a profile can actually use, grouped by source (`native`, then one section per `mcp____`). Defaults to the profile the run executes as (`profile_id` overrides it); `scope` selects that profile's `agent` or `subagent` toolset. Names only by default; `query` narrows by substring over names and descriptions, `verbose` adds descriptions. Declared-but-unregistered tools (server not connected) are listed separately as "Not registered" instead of being advertised | | `ToolManualTool` | `tool_manual` | Return manuals/{name}.md or auto-generate from schema | | `MemoryTool` | `memory` | Unified memory tool: save, search, and forget facts | +| `MemorySearchTool` | `memory_search` | One query over everything Navi remembers: curated facts, session sketches, and verbatim excerpts of turns that context compression dropped. Tiered merge (facts → raw → sketches), provenance on every hit, newer curated fact beats older raw excerpt. See [`memory.md`](memory.md#unified-search-navimemorysearchpy) | | `SpawnAgentTool` | `spawn_agent` | Spawn an isolated subagent (blocking). Optional `profile_id` selects another profile; omitted means parent profile. `inherit_system_prompt=true` prepends the parent profile's full system prompt as a base layer for the subagent | | `SwitchProfileTool` | `switch_profile` | Switch the active profile for a session | | `ListProfilesTool` | `list_profiles` | List all available profiles | diff --git a/navi/core/agent.py b/navi/core/agent.py index c20d9e7..8fe1f14 100644 --- a/navi/core/agent.py +++ b/navi/core/agent.py @@ -191,6 +191,7 @@ backend_registry: BackendRegistry, workers: list["Worker"] | None = None, memory_store: "MemoryStore | None" = None, + memory_search=None, cp_registry: "ContextProviderRegistry | None" = None, mcp_manager=None, ) -> None: @@ -200,11 +201,13 @@ self._backends = backend_registry self._workers: list["Worker"] = workers or [] self._memory = memory_store + self._memory_search = memory_search self._cp_registry = cp_registry self._mcp_manager = mcp_manager self._ctx_builder = ContextBuilder( profile_registry=profile_registry, memory_store=memory_store, + memory_search=memory_search, cp_registry=cp_registry, mcp_manager=mcp_manager, ) @@ -362,6 +365,7 @@ temperature=settings.context_summary_temperature, session_id=session_id, reason="forced", + memory_store=self._memory, keep_recent=settings.context_keep_recent, max_tokens=settings.context_summary_max_tokens, # Enable the intra-turn fallback so a single long autonomous turn @@ -559,6 +563,24 @@ session_cwd=(session.session_metadata or {}).get("cwd"), ) ) + # Raw excerpts compression dropped. Deduped across turns through the + # session's ephemeral metadata: the same chunk should not ride along in + # every prompt of a long session just because the topic persists. + if session.session_metadata is None: + session.session_metadata = {} + # A list, not a set: session_metadata is JSON-encoded into the + # sessions row on every save, and a set is not serializable. + injected_chunks = session.session_metadata.setdefault( + "injected_chunk_ids", [] + ) + mem_recall_task = asyncio.create_task( + self._ctx_builder._memory_recall_msg( + session, + user_message, + user_id=session.user_id, + injected_ids=injected_chunks, + ) + ) ctx_injections = await ctx_task mem_facts = await mem_facts_task if mem_facts: @@ -570,6 +592,9 @@ ) else: log.debug("agent.memory_facts_none", session_id=session_id) + mem_recall = await mem_recall_task + if mem_recall: + ctx_injections.append(mem_recall) # Drain background-task completion notes into the session context # (persisted so compression keeps them; not displayed — the client @@ -988,6 +1013,7 @@ # is over the window — compress now instead of waiting for a # threshold that the truncated count will never reach. context_truncated=self._compressor.pop_context_truncated(), + memory_store=self._memory, ) events: list[AgentEvent] = [] for worker in self._workers: @@ -1112,6 +1138,7 @@ temperature=settings.context_summary_temperature, session_id=session_id, reason="preturn", + memory_store=self._memory, keep_recent=settings.context_keep_recent, max_tokens=settings.context_summary_max_tokens, window=window, @@ -1167,6 +1194,7 @@ temperature=settings.context_summary_temperature, session_id=session_id, reason="midturn", + memory_store=self._memory, keep_recent=settings.context_keep_recent, max_tokens=settings.context_summary_max_tokens, keep_recent_messages=max(12, settings.context_keep_recent * 2), diff --git a/navi/core/compressor.py b/navi/core/compressor.py index 36822f8..ac68257 100644 --- a/navi/core/compressor.py +++ b/navi/core/compressor.py @@ -24,6 +24,8 @@ from navi.llm.base import LLMBackend, Message from navi.config import settings +from navi.memory._chunks import build_chunks +from navi.memory.breadcrumbs import MAX_BREADCRUMBS, extract_breadcrumbs, split_keys_line from .events import ContextCompressed from .token_calibration import calibration_for from .context_builder import truncate_oversized @@ -65,7 +67,12 @@ "signatures, one-line fixes the user approved, key config lines. Do NOT paste long blocks, " "patches, or full command output — keep only the short line that matters.\n" "- Do not paraphrase values that must stay precise.\n" - "- Use Markdown headers exactly as shown." + "- Use Markdown headers exactly as shown.\n" + "- End with ONE line `KEYS: item, item, …` — the breadcrumbs of this history: " + "names, commands, file paths, config keys, amounts, ticket numbers, decisions and " + "links that a later turn may ask about by name. Plain items, no explanations, at " + "most 30, most important first. This line is an index, not prose; write " + "`KEYS: NONE` when nothing is worth indexing." ) @@ -670,6 +677,13 @@ prompt, tools=None, temperature=temperature, model=model, think=False, max_tokens=effective_max_tokens ) summary_text = (response.content or "").strip() or "(summary unavailable)" + # The summarizer's trailing `KEYS:` line is the breadcrumb index for the + # messages this pass drops. It is stripped from the summary the agent reads + # (it is an index, not prose) and carried on the summary message so the + # caller can write it next to the raw chunks. + summary_text, summary_keys = split_keys_line(summary_text) + if not summary_text.strip(): + summary_text = "(summary unavailable)" summary_msg = Message( role="user", @@ -677,6 +691,7 @@ is_summary=True, is_display=False, created_at=datetime.now(timezone.utc), + metadata={"breadcrumbs": summary_keys} if summary_keys else {}, ) return system_msgs + [summary_msg] + to_keep, summary_text @@ -1152,6 +1167,7 @@ max_tokens: int | None = None, keep_recent_messages: int | None = None, window: int | None = None, + memory_store=None, ) -> ContextCompressed | None: """Compresses the session context and persists the changes to the session store.""" count_before = len(session.context) @@ -1206,6 +1222,12 @@ summary_msg.is_display = False session.messages.append(summary_msg) + # Keep the dropped turns searchable before they leave the context for + # good. This must happen while `demoted` is still in hand: after the + # save there is no object that says "these were the turns compression + # removed", only the display history. + await self._index_demoted(memory_store, session, session_id, demoted, summary_msg) + # Add a system message with is_compression=True and content=summary_text session.messages.append( Message( @@ -1257,3 +1279,66 @@ context_tokens=session.context_token_count, max_context_tokens=window_tokens, ) + + async def _index_demoted( + self, + memory_store, + session, + session_id: str, + demoted: list[Message], + summary_msg: Message | None, + ) -> None: + """Write the turns compression just dropped into the raw-chunk index. + + Append-only: re-running compression on the same history hits the + `(session_id, content_hash)` uniqueness and writes nothing new. + + Breadcrumbs come from the summary's `KEYS:` line and, always, from the + deterministic extractor over the chunk text. The hard-truncate fallback + never calls a summarizer, so its index would otherwise be empty; and the + extractor also catches names the summarizer chose not to list. Every + chunk of this pass carries the union — the keys describe the batch, and + a searcher does not know which chunk of the batch holds the name. + """ + if memory_store is None or not demoted or not settings.memory_search_enabled: + return + try: + chunks = build_chunks(demoted) + if not chunks: + return + summary_keys = [] + if summary_msg is not None: + summary_keys = list( + (summary_msg.metadata or {}).get("breadcrumbs") or [] + ) + keys = _merge_breadcrumbs( + summary_keys, + extract_breadcrumbs( + [c["content"] for c in chunks], + limit=max(0, MAX_BREADCRUMBS - len(summary_keys)), + ), + ) + for chunk in chunks: + chunk["breadcrumbs"] = keys + await memory_store.add_chunks( + chunks, user_id=getattr(session, "user_id", None), session_id=session_id + ) + except Exception: + # Indexing must never break compression — a missing chunk is a lost + # search result, a raised exception here is a lost turn. + log.warning("compressor.chunk_index_failed", session_id=session_id, exc_info=True) + + +def _merge_breadcrumbs(*groups: list[str]) -> list[str]: + merged: list[str] = [] + seen: set[str] = set() + for group in groups: + for item in group: + key = " ".join(str(item).lower().split()) + if not key or key in seen: + continue + seen.add(key) + merged.append(str(item)) + if len(merged) >= MAX_BREADCRUMBS: + return merged + return merged diff --git a/navi/core/container.py b/navi/core/container.py index 11beb71..3f92add 100644 --- a/navi/core/container.py +++ b/navi/core/container.py @@ -18,6 +18,7 @@ from navi.core.scheduler import RecallScheduler from navi.db import Database from navi.memory import MemoryStore + from navi.memory.search import UnifiedMemorySearch from navi.mcp import McpManager from navi.skills import ProtocolRunStore, SkillStore from navi.store import KvStore @@ -34,6 +35,7 @@ memory_store: "MemoryStore" = None # type: ignore[assignment] session_store: "SessionStore" = None # type: ignore[assignment] kv_store: "KvStore" = None # type: ignore[assignment] + memory_search: "UnifiedMemorySearch" = None # type: ignore[assignment] todo_store: "PgTodoStore" = None # type: ignore[assignment] skill_store: "SkillStore" = None # type: ignore[assignment] protocol_store: "ProtocolRunStore" = None # type: ignore[assignment] @@ -64,6 +66,7 @@ backend_registry=self.backend_registry, workers=self.workers, memory_store=self.memory_store, + memory_search=self.memory_search, cp_registry=self.cp_registry, mcp_manager=self.mcp_manager, ) @@ -108,6 +111,7 @@ from navi.core.pg_session_store import PgSessionStore from navi.db import Database from navi.memory import MemoryStore + from navi.memory.search import UnifiedMemorySearch from navi.mcp import McpManager from navi.mcp.tools import McpTool from navi.skills import ProtocolRunStore, SkillStore @@ -124,6 +128,10 @@ session_store = PgSessionStore(pool) memory_store = MemoryStore(pool) kv_store = KvStore(pool) + # Read side over facts + sketches + raw chunks. The compressor writes chunks + # through memory_store directly; this facade is what the tool and the + # automatic recall use, so there is exactly one search implementation. + memory_search = UnifiedMemorySearch(memory_store, kv_store) todo_store = PgTodoStore(pool) skill_store = SkillStore(pool) protocol_store = ProtocolRunStore(pool) @@ -157,6 +165,7 @@ todo_store=todo_store, skill_store=skill_store, protocol_store=protocol_store, + memory_search=memory_search, ) # Wire embedding backend into memory store @@ -255,6 +264,7 @@ workers=workers, mcp_manager=mcp_manager, terminal_manager=terminal_manager, + memory_search=memory_search, ) # Start the LLM health probe. It only does work while a target is # degraded, and it is what lets the next turn move back to the primary diff --git a/navi/core/context_builder.py b/navi/core/context_builder.py index b89c603..fa73857 100644 --- a/navi/core/context_builder.py +++ b/navi/core/context_builder.py @@ -121,6 +121,27 @@ return out +def _recall_query(context: list, user_message: str, max_chars: int = 1200) -> str: + """A search query made of the recent turn window, not just the last message. + + The user's "and what about the port?" is unsearchable on its own; in the + company of the previous two exchanges it names the topic. Assistant replies + are included in shortened form because the concrete names (a file, a value, a + plan) usually appear there, not in the user's own wording. + """ + parts: list[str] = [] + turns = [m for m in context if m.role in ("user", "assistant") and not m.is_summary] + for msg in turns[-4:]: + text = (msg.content or "").strip() + if text: + parts.append(text[:300]) + tail = (user_message or "").strip() + if tail: + parts.append(tail[:300]) + query = "\n".join(parts) + return query[-max_chars:].strip() + + def _fact_value_is_abs_path_outside(value: str, cwd: str) -> bool: """True when a fact's value is an absolute filesystem path outside `cwd`. @@ -248,11 +269,13 @@ self, profile_registry, memory_store: "MemoryStore | None" = None, + memory_search=None, cp_registry: "ContextProviderRegistry | None" = None, mcp_manager=None, ) -> None: self._profiles = profile_registry self._memory = memory_store + self._memory_search = memory_search self._cp_registry = cp_registry self._mcp_manager = mcp_manager # Keyed by (profile id, user role): the "Available profiles" block lists @@ -402,6 +425,96 @@ log.info("memory_facts_msg.injected", lines=len(new_facts), length=len(result)) return Message(role="system", content=result) + async def _memory_recall_msg( + self, + session, + user_message: str, + user_id: str | None = None, + injected_ids: list[str] | None = None, + ) -> "Message | None": + """Restore a few verbatim excerpts that compression dropped. + + Only fires once the session actually carries a summary: before that, the + history is still in the context and a search would just duplicate it. + The query is built from the window of recent turns, not the last message, + because "and what about the other one?" is only meaningful next to what + came before it. + + This is the recall path that does not wait for the model to call a tool. + The model can still call `memory_search` for anything wider. + """ + if self._memory_search is None or not _config.settings.memory_recall_enabled: + return None + context = getattr(session, "context", None) or [] + if not any(m.is_summary for m in context): + return None + session_id = getattr(session, "id", None) + query = _recall_query(context, user_message) + if not query: + return None + try: + hits = await self._memory_search.search( + query, + scope="raw", + k=_config.settings.memory_recall_k, + user_id=user_id, + session_id=session_id, + current_session_id=session_id, + summary_at=await self._memory_search.latest_summary_at(session_id), + ) + except Exception: + log.debug("memory_recall.search_failed", exc_info=True) + return None + + selected = [] + for hit in hits: + if hit.source_type != "raw" or not hit.chunk_id: + continue + if injected_ids is not None and hit.chunk_id in injected_ids: + continue + if hit.score is not None: + if hit.score < _config.settings.memory_recall_min_similarity: + continue + elif hit.terms_hit < 2: + # No vector score (lexical fallback): demand a real keyword + # overlap, otherwise a one-word coincidence injects a turn. + continue + if hit.superseded_by: + continue + selected.append(hit) + if len(selected) >= _config.settings.memory_recall_k: + break + if not selected: + return None + if injected_ids is not None: + for hit in selected: + injected_ids.append(hit.chunk_id) + + lines = [ + "## Recovered from earlier in this conversation", + "These are verbatim excerpts of turns that were compressed out of context. " + "The summary is the curated version; these carry the exact details — names, " + "commands, values, links.", + ] + for hit in selected: + turn = ( + f"turn {hit.start_seq}" if hit.start_seq == hit.end_seq + else f"turns {hit.start_seq}–{hit.end_seq}" + ) + if hit.end_seq < 0: + turn = "turn unknown" + when = hit.timestamp.strftime("%Y-%m-%d %H:%M") if hit.timestamp else "unknown time" + lines.append(f"\n• session {hit.session_id}, {turn}, {when}") + lines.append(hit.text[:1500].strip()) + result = "\n".join(lines) + log.info( + "memory_recall.injected", + session_id=session_id, + chunks=len(selected), + length=len(result), + ) + return Message(role="system", content=result) + async def _collect_context_injections(self, profile: "AgentProfile") -> list[Message]: if self._cp_registry is None: return [] diff --git a/navi/core/orchestrator.py b/navi/core/orchestrator.py index e94930e..03af54b 100644 --- a/navi/core/orchestrator.py +++ b/navi/core/orchestrator.py @@ -422,6 +422,7 @@ self._container.backend_registry, workers=self._container.workers, memory_store=self._container.memory_store, + memory_search=getattr(self._container, "memory_search", None), cp_registry=self._container.cp_registry, mcp_manager=mcp_manager, ) diff --git a/navi/core/pg_session_store.py b/navi/core/pg_session_store.py index 40077fb..e1e221d 100644 --- a/navi/core/pg_session_store.py +++ b/navi/core/pg_session_store.py @@ -1005,6 +1005,15 @@ pool = await self._get_pool() async with pool.acquire() as conn: result = await conn.execute("DELETE FROM sessions WHERE id = $1", session_id) + # Raw chunks have no FK on purpose (memory DDL may run before the + # sessions table exists), so deleting a chat must clean them here — + # otherwise a deleted conversation stays searchable. + try: + await conn.execute( + "DELETE FROM memory_chunks WHERE session_id = $1", session_id + ) + except asyncpg.UndefinedTableError: + pass return pending is not None or result == "DELETE 1" async def get_or_create_share(self, session_id: str, owner_id: str | None) -> dict | None: diff --git a/navi/core/registry.py b/navi/core/registry.py index 0ee55c5..161fb9b 100644 --- a/navi/core/registry.py +++ b/navi/core/registry.py @@ -23,6 +23,7 @@ ListProfilesTool, ManageRecallTool, MemoryTool, + MemorySearchTool, NotifyTool, PeerTool, ReflectTool, @@ -229,6 +230,7 @@ todo_store=None, skill_store=None, protocol_store=None, + memory_search=None, ) -> tuple[ToolRegistry, ProfileRegistry, BackendRegistry, ContextProviderRegistry]: """Build and populate registries with all built-in components.""" from navi.core.ai_helper import AIHelper @@ -270,6 +272,7 @@ registry=tools, profile_registry=profiles, mcp_manager=mcp_manager ) memory_tool = MemoryTool(memory_store) if memory_store else None + memory_search_tool = MemorySearchTool(memory_search) if memory_search else None mcp_status_tool = McpStatusTool(mcp_manager=mcp_manager) create_mcp_server_tool = CreateMcpServerTool() test_mcp_tool_tool = TestMcpToolTool(mcp_manager=mcp_manager) @@ -310,6 +313,8 @@ TasksTool()] if memory_tool: builtins.append(memory_tool) + if memory_search_tool: + builtins.append(memory_search_tool) if skill_store is not None: from navi.tools.skills import SkillTool diff --git a/navi/memory/_chunks.py b/navi/memory/_chunks.py new file mode 100644 index 0000000..47088da --- /dev/null +++ b/navi/memory/_chunks.py @@ -0,0 +1,414 @@ +"""Raw conversation chunks — the append-only record of what compression dropped. + +Long-term memory used to have three shapes: curated facts, per-session sketches, +and the live context. When the context was compressed, everything that did not +fit the summary was simply gone — the message rows stay in the database, but +nothing searched them by meaning, and after a reload the session carried only the +summary. "We discussed this earlier" was unanswerable. + +This table is the missing store: every message that compression demotes is +written here once, verbatim (a very large single message is elided in the +middle), with the turn range and timestamps it came from and the breadcrumb keys +extracted for it. Three properties are load-bearing: + +* **Append-only and immutable.** A row is never updated. A correction is a new + chunk written later; `UNIQUE(session_id, content_hash)` only prevents the same + dropped run from being indexed twice, which is what makes a compression replay + idempotent. +* **Read together with facts and sketches**, through `UnifiedMemorySearch`, so an + agent has one place to ask. Writing stays separate: chunks are written only by + the compressor, never by a tool. +* **Provenance is part of the record** — session, turn range, timestamps — because + an answer built on a raw excerpt must say when it was said, and a curated fact + from last week must win over a raw line from last month. +""" + +from __future__ import annotations + +import hashlib +import uuid +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +import structlog + +from ._embeddings import _vector_to_str +from ._facts import _normalize_query + +if TYPE_CHECKING: + from navi.llm.base import Message + +log = structlog.get_logger() + +# One chunk is one searchable unit. 4000 chars is roughly a long tool result or +# a dozen chat messages — small enough that a hit is about one topic, large +# enough that the answer's context is not split across three rows. +CHUNK_MAX_CHARS = 4000 +# A single message can be megabytes (a full file read). Keep its head and tail: +# the middle of a dump is the part nobody asks for later, while the command line +# at the top and the error at the bottom are usually the breadcrumbs. +CHUNK_MESSAGE_MAX_CHARS = 12_000 + +# Cosine distance cutoff for a chunk match. Facts use 0.3 because a fact value is +# one short sentence; a 4 KB passage against a short query naturally sits farther +# away, and 0.45 is where the pilot's paraphrased queries still landed. Anything +# past it goes to the lexical fallback instead — a wrong vector hit is worse than +# a keyword hit for a long passage. +VECTOR_CUTOFF_DISTANCE = 0.45 + +_ROLE_PREFIX = { + "user": "User", + "assistant": "Assistant", + "tool": "Tool result", +} + + +def build_chunks( + messages: list[Message], + breadcrumbs: list[str] | None = None, + max_chars: int = CHUNK_MAX_CHARS, +) -> list[dict]: + """Group dropped messages into immutable chunks. + + Consecutive messages are packed together until `max_chars`, so a turn's user + message, tool calls and the assistant reply land in one chunk. Summary and + compression marker messages are skipped — a summary is not raw history, and + indexing old summaries again would make each compression index its own past. + """ + chunks: list[dict] = [] + current: list[str] = [] + current_len = 0 + start_seq: int | None = None + end_seq: int | None = None + started_at: datetime | None = None + ended_at: datetime | None = None + keys = list(breadcrumbs or []) + + def flush() -> None: + nonlocal current, current_len, start_seq, end_seq, started_at, ended_at + if not current: + return + text = "\n\n".join(current).strip() + if text: + chunks.append({ + "content": text, + "content_hash": hashlib.sha256(text.encode("utf-8", "replace")).hexdigest(), + "start_seq": start_seq if start_seq is not None else -1, + "end_seq": end_seq if end_seq is not None else -1, + "started_at": started_at, + "ended_at": ended_at, + "breadcrumbs": keys, + }) + current, current_len = [], 0 + start_seq = end_seq = None + started_at = ended_at = None + + for msg in messages: + if msg.role == "system" or msg.is_summary or msg.is_compression: + continue + text = _format_message(msg) + if not text: + continue + if current and current_len + len(text) > max_chars: + flush() + current.append(text) + current_len += len(text) + 2 + if start_seq is None: + start_seq = msg.sequence_number + started_at = msg.created_at + if msg.sequence_number >= 0: + end_seq = msg.sequence_number + if msg.created_at is not None: + ended_at = msg.created_at + flush() + return chunks + + +def _format_message(msg: Message) -> str: + content = (msg.content or "").strip() + if not content: + # An assistant message that only requested tool calls still records what + # was asked for; the result follows in its own message. + if msg.tool_calls: + names = ", ".join(tc.name for tc in msg.tool_calls if getattr(tc, "name", None)) + return f"Assistant requested tools: {names}" if names else "" + return "" + if len(content) > CHUNK_MESSAGE_MAX_CHARS: + head = CHUNK_MESSAGE_MAX_CHARS * 3 // 4 + tail = CHUNK_MESSAGE_MAX_CHARS - head - 40 + content = f"{content[:head]}\n…[middle elided]…\n{content[-tail:]}" + prefix = _ROLE_PREFIX.get(msg.role, msg.role.title()) + if msg.role == "tool" and msg.name: + prefix = f"Tool result ({msg.name})" + return f"{prefix}: {content}" + + +class ChunkMixin: + """Storage and search for dropped-raw-history chunks. + + Expected on the composite class: + _get_pool() -> asyncpg.Pool + _generate_embedding(text) -> list[float] | None + _generate_embeddings(texts) -> list[list[float] | None] + _has_pgvector() -> bool + """ + + async def add_chunks( + self, + chunks: list[dict], + user_id: str | None = None, + session_id: str | None = None, + ) -> int: + """Write chunks, skipping ones already indexed for this session.""" + if not chunks or not session_id: + return 0 + # The embedding column exists only where pgvector does; on a plain Postgres + # the `::vector` cast in the INSERT would fail the whole write, so the row + # is written text-only and found through FTS. + use_vectors = await self._has_pgvector() + embeddings = ( + await self._generate_embeddings([c["content"] for c in chunks]) + if use_vectors else [None] * len(chunks) + ) + now = datetime.now(timezone.utc) + pool = await self._get_pool() + written = 0 + async with pool.acquire() as conn: + for chunk, embedding in zip(chunks, embeddings): + vec_str = _vector_to_str(embedding) if embedding else None + if use_vectors: + result = await conn.execute( + """INSERT INTO memory_chunks + (id, user_id, session_id, start_seq, end_seq, started_at, ended_at, + content, breadcrumbs, content_hash, created_at, embedding) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::vector) + ON CONFLICT (session_id, content_hash) DO NOTHING""", + str(uuid.uuid4()), user_id, session_id, + chunk["start_seq"], chunk["end_seq"], + chunk.get("started_at"), chunk.get("ended_at"), + chunk["content"], list(chunk.get("breadcrumbs") or []), + chunk["content_hash"], now, vec_str, + ) + else: + result = await conn.execute( + """INSERT INTO memory_chunks + (id, user_id, session_id, start_seq, end_seq, started_at, ended_at, + content, breadcrumbs, content_hash, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + ON CONFLICT (session_id, content_hash) DO NOTHING""", + str(uuid.uuid4()), user_id, session_id, + chunk["start_seq"], chunk["end_seq"], + chunk.get("started_at"), chunk.get("ended_at"), + chunk["content"], list(chunk.get("breadcrumbs") or []), + chunk["content_hash"], now, + ) + if result == "INSERT 0 1": + written += 1 + if written: + log.info( + "memory.chunks_written", + session_id=session_id, + written=written, + skipped=len(chunks) - written, + ) + return written + + async def search_chunks( + self, + query: str, + user_id: str | None = None, + session_id: str | None = None, + limit: int = 8, + ) -> list[dict]: + """Search raw chunks: vector first, lexical fallback when that misses. + + `session_id` narrows to one conversation; omitting it searches every + session of the user. Anonymous rows (`user_id IS NULL`) are their own + scope, exactly like facts. + """ + if limit <= 0: + return [] + rows: list[dict] = [] + if self._embedding_backend and await self._has_pgvector(): + query_embedding = await self._generate_embedding(query) + vec_str = _vector_to_str(query_embedding) if query_embedding else None + if vec_str: + try: + rows = await self._vector_chunks( + vec_str, user_id, session_id, limit, + ) + except Exception: + log.warning("memory.chunk_vector_failed", query=query[:40], exc_info=True) + if rows: + log.debug("memory.chunk_vector_search", hits=len(rows), query=query[:40]) + return rows + rows = await self._lexical_chunks(query, user_id, session_id, limit) + if rows: + log.debug("memory.chunk_lexical_search", hits=len(rows), query=query[:40]) + return rows + + async def _vector_chunks( + self, + vec_str: str, + user_id: str | None, + session_id: str | None, + limit: int, + ) -> list[dict]: + where, params = _chunk_filter(user_id, session_id) + params.append(vec_str) + vec_idx = len(params) + params.append(limit * 3) + lim_idx = len(params) + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + f"""SELECT id, session_id, start_seq, end_seq, started_at, ended_at, + content, breadcrumbs, created_at, + embedding <=> ${vec_idx}::vector AS distance + FROM memory_chunks + WHERE {where} + ORDER BY embedding <=> ${vec_idx}::vector + LIMIT ${lim_idx}""", + *params, + ) + hits: list[dict] = [] + for row in rows: + distance = row["distance"] + if distance is None or distance >= VECTOR_CUTOFF_DISTANCE: + continue + hit = _row_to_chunk(row) + hit["score"] = max(0.0, 1.0 - float(distance)) + hits.append(hit) + if len(hits) >= limit: + break + return hits + + async def _lexical_chunks( + self, + query: str, + user_id: str | None, + session_id: str | None, + limit: int, + ) -> list[dict]: + terms = _normalize_query(query) + where, params = _chunk_filter(user_id, session_id) + if not terms: + return await self._recent_chunks(where, params, limit) + ts_query = " ".join(terms) + like_patterns = [f"%{term}%" for term in terms] + params.append(ts_query) + ts_idx = len(params) + params.append(like_patterns) + like_idx = len(params) + params.append(limit * 4) + lim_idx = len(params) + match_clause = ( + f"(to_tsvector('simple', content) @@ plainto_tsquery('simple', ${ts_idx}) " + f"OR content ILIKE ANY(${like_idx}::text[]) " + f"OR array_to_string(breadcrumbs, ' ') ILIKE ANY(${like_idx}::text[]))" + ) + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + f"""SELECT id, session_id, start_seq, end_seq, started_at, ended_at, + content, breadcrumbs, created_at, + ts_rank(to_tsvector('simple', content), + plainto_tsquery('simple', ${ts_idx})) AS rank + FROM memory_chunks + WHERE {where} AND {match_clause} + ORDER BY rank DESC, ended_at DESC NULLS LAST + LIMIT ${lim_idx}""", + *params, + ) + hits = [] + normalized_terms = {t.lower() for t in terms if len(t) > 2} + for row in rows: + hit = _row_to_chunk(row) + haystack = f"{hit['content']} {' '.join(hit['breadcrumbs'])}".lower() + hit["terms_hit"] = sum(1 for term in normalized_terms if term in haystack) + hit["score"] = None + hits.append(hit) + # A row matched one long term by accident; a row matching several query + # terms is the one the caller meant. Rank on that before the DB order. + hits.sort(key=lambda h: (-h["terms_hit"], -_ts(h.get("rank")))) + return hits[:limit] + async def _recent_chunks(self, where: str, params: list, limit: int) -> list[dict]: + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + f"""SELECT id, session_id, start_seq, end_seq, started_at, ended_at, + content, breadcrumbs, created_at + FROM memory_chunks + WHERE {where} + ORDER BY ended_at DESC NULLS LAST + LIMIT ${len(params) + 1}""", + *params, limit, + ) + hits = [] + for index, row in enumerate(rows): + hit = _row_to_chunk(row) + hit["score"] = 1.0 - index * 0.05 + hit["terms_hit"] = 0 + hits.append(hit) + return hits + + async def latest_summary_at(self, session_id: str) -> datetime | None: + """When this session was last compressed — raw chunks before it are verbatim + history that the current summary has already curated.""" + pool = await self._get_pool() + async with pool.acquire() as conn: + try: + return await conn.fetchval( + "SELECT max(created_at) FROM session_messages " + "WHERE session_id = $1 AND is_summary = TRUE", + session_id, + ) + except Exception: + log.debug("memory.latest_summary_failed", session_id=session_id, exc_info=True) + return None + + async def chunk_count(self, session_id: str | None = None, user_id: str | None = None) -> int: + where, params = _chunk_filter(user_id, session_id) + pool = await self._get_pool() + async with pool.acquire() as conn: + return int(await conn.fetchval( + f"SELECT count(*) FROM memory_chunks WHERE {where}", *params + ) or 0) + + +def _chunk_filter(user_id: str | None, session_id: str | None) -> tuple[str, list]: + """WHERE clause for the (user, session) scope. Anonymous rows are NULL-owned.""" + clauses = [] + params: list = [] + if user_id is None: + clauses.append("user_id IS NULL") + else: + params.append(user_id) + clauses.append(f"user_id = ${len(params)}") + if session_id is not None: + params.append(session_id) + clauses.append(f"session_id = ${len(params)}") + return " AND ".join(clauses) or "TRUE", params + + +def _row_to_chunk(row) -> dict: + return { + "rank": row["rank"] if "rank" in row.keys() else None, + "id": row["id"], + "session_id": row["session_id"], + "start_seq": row["start_seq"], + "end_seq": row["end_seq"], + "started_at": row["started_at"], + "ended_at": row["ended_at"], + "content": row["content"], + "breadcrumbs": list(row["breadcrumbs"] or []), + "created_at": row["created_at"], + "terms_hit": 0, + } + + +def _ts(rank) -> float: + try: + return float(rank or 0.0) + except (TypeError, ValueError): + return 0.0 diff --git a/navi/memory/_ddl.py b/navi/memory/_ddl.py index 22798c0..857c3ac 100644 --- a/navi/memory/_ddl.py +++ b/navi/memory/_ddl.py @@ -10,6 +10,9 @@ "CREATE INDEX IF NOT EXISTS idx_memory_facts_embedding ON memory_facts USING hnsw (embedding vector_cosine_ops)" if pgvector_available else "" ) + chunk_embedding_col = ( + f"embedding vector({settings.embedding_dimensions})," if pgvector_available else "" + ) stmts = [ # Migration: add user_id to existing tables (nullable for legacy) "ALTER TABLE memory_facts ADD COLUMN IF NOT EXISTS user_id TEXT REFERENCES navi_users(id) ON DELETE CASCADE", @@ -64,7 +67,50 @@ session_id TEXT PRIMARY KEY, extracted_at TIMESTAMPTZ NOT NULL )""", + # Raw history dropped by compression: append-only, never updated. No FK to + # `sessions` on purpose — this DDL can run before the sessions table exists + # on a fresh database, and a failed CREATE here would silently disable the + # whole store for the process. PgSessionStore.delete clears the rows. + f"""CREATE TABLE IF NOT EXISTS memory_chunks ( + id TEXT PRIMARY KEY, + user_id TEXT, + session_id TEXT NOT NULL, + start_seq INTEGER NOT NULL DEFAULT -1, + end_seq INTEGER NOT NULL DEFAULT -1, + started_at TIMESTAMPTZ, + ended_at TIMESTAMPTZ, + content TEXT NOT NULL, + breadcrumbs TEXT[] NOT NULL DEFAULT '{{}}', + content_hash TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + {chunk_embedding_col} + UNIQUE(session_id, content_hash) + )""", + "CREATE INDEX IF NOT EXISTS idx_memory_chunks_scope " + "ON memory_chunks (user_id, session_id, ended_at DESC)", + # `to_tsvector('simple', content)` has no stemming and no stopwords, which is + # what a mixed-language corpus needs; the expression here must match the + # query's exactly or the planner cannot use this index. + "CREATE INDEX IF NOT EXISTS idx_memory_chunks_fts " + "ON memory_chunks USING gin (to_tsvector('simple', content))", + "CREATE INDEX IF NOT EXISTS idx_memory_chunks_breadcrumbs " + "ON memory_chunks USING gin (breadcrumbs)", ] + if pgvector_available: + # An older database may hold a `memory_chunks` created before pgvector was + # available: CREATE TABLE IF NOT EXISTS leaves it without the column, and + # every INSERT then dies on "column embedding does not exist". Add it before + # the HNSW index so the table heals itself on the first run that can use + # vectors. + stmts.append( + "ALTER TABLE memory_chunks ADD COLUMN IF NOT EXISTS " + "embedding vector(%d)" % settings.embedding_dimensions + ) + if pgvector_available: + stmts.append( + "CREATE INDEX IF NOT EXISTS idx_memory_chunks_embedding " + "ON memory_chunks USING hnsw (embedding vector_cosine_ops)" + ) if embedding_idx: stmts.insert(2, embedding_idx) return stmts diff --git a/navi/memory/breadcrumbs.py b/navi/memory/breadcrumbs.py new file mode 100644 index 0000000..ed8c259 --- /dev/null +++ b/navi/memory/breadcrumbs.py @@ -0,0 +1,146 @@ +"""Breadcrumb keys — the names a later search will ask by. + +When compression drops a run of messages, the summary is the only thing left in +the model's context. A summary is good at the shape of what happened and bad at +the exact things a person later says out loud: a file name, a command, a port, an +amount, a ticket number. Those are exactly the strings a search query is made of. + +So every compression also stores a breadcrumb list — the entities that were +dropped — and the breadcrumb index is searched alongside the chunk text. The list +comes from two places: the summarizer writes a `KEYS:` line (the prompt asks for +it), and when that line is missing — the hard-truncate fallback never sees a +summarizer at all — a deterministic extractor reads the dropped messages directly. + +The extractor errs toward including a name: a breadcrumb that nobody searches +for costs a few bytes, one that is missing costs the detail. +""" + +from __future__ import annotations + +import re + +# Enough names to find a chunk by, few enough that the index stays a list and not +# a copy of the conversation. +MAX_BREADCRUMBS = 40 +_MAX_ITEM_CHARS = 120 + +_KEYS_LINE = re.compile(r"^\s*(?:\*\*)?KEYS(?:\*\*)?\s*[:：]\s*(.*)$", re.IGNORECASE) +_KEYS_NOISE = frozenset({"", "none", "n/a", "—", "-", "ничего", "нет"}) + +# Inline code first: commands, flags, paths and identifiers the agent wrote +# deliberately are the strongest breadcrumbs in a transcript. +_BACKTICK = re.compile(r"`([^`\n]{2,80})`") +_URL = re.compile(r"https?://[^\s<>\"')\]]{4,200}") +# A path has at least one slash and at least one dotted segment or a leading +# slash/dot — `foo/bar` alone is too often prose ("and/or"). +_PATH = re.compile( + r"(? tuple[str, list[str]]: + """Split the summarizer's trailing `KEYS:` line off the summary. + + Returns the summary without that line and the parsed keys. A summary with no + `KEYS:` line comes back unchanged with an empty list — the caller then falls + back to :func:`extract_breadcrumbs` on the dropped messages. + """ + if not text: + return text, [] + lines = text.splitlines() + for index in range(len(lines) - 1, -1, -1): + match = _KEYS_LINE.match(lines[index]) + if match is None: + continue + keys = _parse_keys(match.group(1)) + remaining = lines[:index] + lines[index + 1:] + return "\n".join(remaining).rstrip() + ("\n" if text.endswith("\n") else ""), keys + return text, [] + + +def _parse_keys(raw: str) -> list[str]: + keys: list[str] = [] + seen: set[str] = set() + for part in re.split(r"[,;]", raw): + item = re.sub(r"^[-*•]\s*", "", part.strip()) + item = item.strip("`\"'*").strip() + if item.lower() in _KEYS_NOISE: + continue + if not item or len(item) > _MAX_ITEM_CHARS: + continue + if item.lower() in seen: + continue + seen.add(item.lower()) + keys.append(item) + if len(keys) >= MAX_BREADCRUMBS: + break + return keys + + +def extract_breadcrumbs(texts: list[str], limit: int = MAX_BREADCRUMBS) -> list[str]: + """Pull the entities worth indexing out of dropped message text. + + Deterministic on purpose: the same conversation must produce the same index + keys on every replay, otherwise a chunk written twice gets two different + breadcrumb sets. + """ + found: list[str] = [] + seen: set[str] = set() + if limit <= 0: + return found + for text in texts: + if not text: + continue + for pattern in _PATTERNS: + for match in pattern.finditer(text): + item = match.group(1) if match.groups() else match.group(0) + item = item.strip().strip("`\"'").strip() + if not _keep(item): + continue + key = " ".join(item.lower().split()) + if key in seen: + continue + seen.add(key) + found.append(item) + if len(found) >= limit: + return found + return found + + +_GENERIC = frozenset({ + "true", "false", "none", "null", "and/or", "etc", "e.g", "i.e", +}) + + +def _keep(item: str) -> bool: + if len(item) < 2 or len(item) > _MAX_ITEM_CHARS: + return False + if item.lower() in _GENERIC: + return False + # A breadcrumb must carry at least one letter or digit; pure punctuation + # shows up in backticks and quoted strings often enough to matter. + return any(ch.isalnum() for ch in item) diff --git a/navi/memory/search.py b/navi/memory/search.py new file mode 100644 index 0000000..db650b3 --- /dev/null +++ b/navi/memory/search.py @@ -0,0 +1,331 @@ +"""One search across everything Navi remembers. + +Memory is stored in three different shapes, and they are different on purpose: + +* **facts** — curated, rewritten by key, user lifetime. "What is true about the + user" (``memory_facts``). +* **sketches** — a session's working notes, updated in place while a task runs + (``session_store`` scope ``scratchpad``). +* **raw** — verbatim messages that context compression dropped + (``memory_chunks``), append-only and immutable. + +Writing them is deliberately separate. Reading them is not: an agent asked "what +did we decide about the port" should not have to know which of the three holds +the answer, or that it is even a different store. This module merges the three +into one ranked answer with provenance attached to every item. + +The merge is tiered, not a blind mix: curated facts first, then raw excerpts by +relevance, then sketches from the current session. A fact that was rewritten +after a raw excerpt wins on freshness — the raw line is still returned, labelled +as outdated, because "we changed our mind, the old value was X" is often exactly +the question. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime + +import structlog + +from ._facts import _normalize_query + +log = structlog.get_logger() + +SCOPES = ("facts", "raw", "sketches", "all") +_MAX_K = 20 +_HIT_TEXT_CHARS = 1200 +_STOPWORDS = frozenset({ + "the", "and", "for", "was", "were", "that", "this", "with", "from", "have", + "has", "had", "not", "but", "you", "your", "our", "are", "what", "when", + "where", "which", "how", "why", "did", "does", "can", "could", "would", + "should", "about", "there", "then", "than", "into", "they", "them", "its", + "это", "что", "как", "где", "когда", "был", "была", "были", "для", "или", + "там", "тут", "уже", "ещё", "еще", "если", "чтобы", "только", "можно", +}) + + +@dataclass +class MemoryHit: + source_type: str # "fact" | "raw" | "sketch" + text: str + session_id: str | None = None + timestamp: datetime | None = None + key: str | None = None # fact key / sketch section + category: str | None = None # fact category + start_seq: int = -1 # raw: first dropped turn + end_seq: int = -1 # raw: last dropped turn + chunk_id: str | None = None + score: float | None = None # vector similarity, 0..1 when available + terms_hit: int = 0 # lexical overlap, for the FTS fallback + superseded_by: str | None = None # newer curated fact that overrides this + before_summary: bool = False # raw older than the session's last summary + breadcrumbs: tuple[str, ...] = field(default_factory=tuple) + + +class UnifiedMemorySearch: + """Read-side facade over facts, raw chunks and sketches.""" + + def __init__(self, memory_store=None, kv_store=None) -> None: + self._memory = memory_store + self._kv = kv_store + + async def latest_summary_at(self, session_id: str) -> datetime | None: + """When `session_id` was last compressed, for the freshness label.""" + if self._memory is None or not session_id: + return None + try: + return await self._memory.latest_summary_at(session_id) + except Exception: + return None + + async def search( + self, + query: str, + *, + scope: str = "all", + k: int = 5, + user_id: str | None = None, + session_id: str | None = None, + current_session_id: str | None = None, + summary_at: datetime | None = None, + ) -> list[MemoryHit]: + """Search the requested stores and return a tiered, provenance-labelled list. + + `session_id` narrows raw chunks to one conversation; `current_session_id` + only orders sketches (the current session's pad first). `summary_at` is the + timestamp of the session's latest summary — raw chunks older than it are + the verbatim record of history the summary already curated. + """ + query = (query or "").strip() + if not query: + return [] + scope = scope if scope in SCOPES else "all" + k = max(1, min(int(k or 5), _MAX_K)) + wanted = {"facts", "raw", "sketches"} if scope == "all" else {scope} + + facts: list[MemoryHit] = [] + raw: list[MemoryHit] = [] + sketches: list[MemoryHit] = [] + + if "facts" in wanted: + facts = await self._facts(query, user_id, k) + if "raw" in wanted: + raw = await self._raw(query, user_id, session_id, k) + if "sketches" in wanted: + sketches = await self._sketches(query, user_id, k, current_session_id or session_id) + + _mark_freshness(facts, raw, summary_at, current_session_id or session_id) + hits = facts + raw + sketches + log.info( + "memory_search.done", + query=query[:60], + scope=scope, + facts=len(facts), + raw=len(raw), + sketches=len(sketches), + ) + return hits + + async def _facts(self, query: str, user_id: str | None, k: int) -> list[MemoryHit]: + if self._memory is None: + return [] + try: + rows = await self._memory.search_facts(query, user_id=user_id, limit=k) + except Exception: + log.warning("memory_search.facts_failed", query=query[:40], exc_info=True) + return [] + hits = [] + for row in rows: + hits.append(MemoryHit( + source_type="fact", + text=f"{row.get('key')}: {row.get('value')}", + session_id=row.get("source_session_id") or None, + timestamp=_as_datetime(row.get("updated_at")), + key=row.get("key"), + category=row.get("category"), + score=None, + )) + return hits + + async def _raw( + self, query: str, user_id: str | None, session_id: str | None, k: int, + ) -> list[MemoryHit]: + if self._memory is None: + return [] + try: + rows = await self._memory.search_chunks( + query, user_id=user_id, session_id=session_id, limit=k, + ) + except Exception: + log.warning("memory_search.raw_failed", query=query[:40], exc_info=True) + return [] + hits = [] + for row in rows: + hits.append(MemoryHit( + source_type="raw", + text=row.get("content") or "", + session_id=row.get("session_id"), + timestamp=row.get("ended_at") or row.get("created_at"), + start_seq=row.get("start_seq", -1), + end_seq=row.get("end_seq", -1), + chunk_id=row.get("id"), + score=row.get("score"), + terms_hit=row.get("terms_hit", 0), + breadcrumbs=tuple(row.get("breadcrumbs") or ()), + )) + return hits + + async def _sketches( + self, query: str, user_id: str | None, k: int, prefer_session_id: str | None, + ) -> list[MemoryHit]: + if self._kv is None: + return [] + terms = _normalize_query(query) + try: + rows = await self._kv.search_values( + user_id, + "scratchpad", + terms, + limit=k, + prefer_session_id=prefer_session_id, + ) + except Exception: + log.warning("memory_search.sketches_failed", query=query[:40], exc_info=True) + return [] + hits = [] + for row in rows: + value = (row.get("value") or "").strip() + if not value: + continue + hits.append(MemoryHit( + source_type="sketch", + text=value, + session_id=row.get("session_id"), + timestamp=_as_datetime(row.get("updated_at")), + key=row.get("key"), + )) + return hits + + +def _as_datetime(value) -> datetime | None: + """Timestamps come back in two shapes: asyncpg returns `datetime`, the facts + store hands out ISO strings (`_facts._row_to_fact` formats for the API). The + freshness comparison runs across both, so normalize at the boundary rather + than letting `str <= datetime` blow up the whole search.""" + if value is None or isinstance(value, datetime): + return value + try: + return datetime.fromisoformat(str(value)) + except (TypeError, ValueError): + return None + + +def _mark_freshness( + facts: list[MemoryHit], + raw: list[MemoryHit], + summary_at: datetime | None, + summary_session_id: str | None, +) -> None: + """Label raw excerpts that a newer curated fact has already overridden. + + Only facts *newer* than the excerpt can supersede it — an old fact next to a + newer verbatim line is the lossy copy, not the truth. Overlap is judged on + content words, not on the user's query: the conflict has to be between the + two records themselves. + + `summary_at` belongs to one session's summary; a chunk from another session + says nothing about it, so the comparison is scoped to that session. + """ + newest: MemoryHit | None = None + for fact in facts: + if fact.timestamp is None: + continue + if newest is None or fact.timestamp > newest.timestamp: + newest = fact + for hit in raw: + if ( + summary_at is not None + and summary_session_id is not None + and hit.session_id == summary_session_id + and hit.timestamp is not None + ): + hit.before_summary = hit.timestamp < summary_at + if newest is None or newest.timestamp is None or hit.timestamp is None: + continue + if newest.timestamp <= hit.timestamp: + continue + if _shared_terms(newest.text, hit.text) >= 2: + label = newest.key or "a newer fact" + hit.superseded_by = f"{label} (updated {_fmt_time(newest.timestamp)})" + + +def _shared_terms(left: str, right: str) -> int: + left_terms = { + t for t in _normalize_query(left) + if len(t) > 2 and t not in _STOPWORDS + } + right_terms = set(_normalize_query(right)) + return len(left_terms & right_terms) + + +def _fmt_time(value: datetime | None) -> str: + if value is None: + return "" + try: + return value.strftime("%Y-%m-%d %H:%M") + except (AttributeError, ValueError): + return str(value)[:16] + + +def format_hits(hits: list[MemoryHit], query: str, scope: str) -> str: + """Render search results for a tool result — provenance on every line.""" + header = f'memory_search "{query}" (scope={scope}): {len(hits)} hit(s)' + if not hits: + return header + "\nNo matches in any store. Try fewer, different words." + lines = [header] + for hit in hits: + if hit.source_type == "fact": + where = f"fact [{hit.category or 'other'}]" + if hit.session_id: + where += f" · session {hit.session_id}" + if hit.timestamp is not None: + where += f" · updated {_fmt_time(hit.timestamp)}" + lines.append(f"\n• {where}\n {hit.text}") + continue + if hit.source_type == "raw": + turn = ( + f"turn {hit.start_seq}" if hit.start_seq == hit.end_seq + else f"turns {hit.start_seq}–{hit.end_seq}" + ) + if hit.end_seq < 0: + turn = "turn unknown" + where = f"raw · session {hit.session_id} · {turn}" + if hit.timestamp is not None: + where += f" · {_fmt_time(hit.timestamp)}" + lines.append(f"\n• {where}\n {_excerpt(hit.text)}") + if hit.superseded_by: + lines.append( + f" ⚠ outdated: a newer curated fact overrides this — {hit.superseded_by}. " + "Trust the fact; quote this only as history." + ) + elif hit.before_summary: + lines.append( + " (verbatim record from before the current summary — exact details, " + "but the summary is the curated version.)" + ) + if hit.breadcrumbs: + lines.append(f" keys: {', '.join(hit.breadcrumbs[:12])}") + continue + where = f"sketch · session {hit.session_id} · {hit.key or 'main'}" + if hit.timestamp is not None: + where += f" · updated {_fmt_time(hit.timestamp)}" + lines.append(f"\n• {where}\n {_excerpt(hit.text)}") + return "\n".join(lines) + + +def _excerpt(text: str) -> str: + text = (text or "").strip() + if len(text) <= _HIT_TEXT_CHARS: + return text + return text[:_HIT_TEXT_CHARS].rstrip() + " …" diff --git a/navi/memory/store.py b/navi/memory/store.py index 0053662..808047f 100644 --- a/navi/memory/store.py +++ b/navi/memory/store.py @@ -3,6 +3,7 @@ MemoryStore is a composite of mixins: - EmbeddingMixin (pgvector checks, embedding generation, backfill) - FactMixin (CRUD + search with vector/ILIKE fallback) + - ChunkMixin (append-only raw history dropped by compression + search) - SummaryMixin (conversation summary) - SessionStateMixin (per-session extraction tracking) """ @@ -13,6 +14,7 @@ import asyncpg import structlog +from ._chunks import ChunkMixin from ._ddl import _build_ddl from ._embeddings import EmbeddingMixin from ._facts import FactMixin @@ -25,7 +27,7 @@ log = structlog.get_logger() -class MemoryStore(EmbeddingMixin, FactMixin, SummaryMixin, SessionStateMixin): +class MemoryStore(EmbeddingMixin, FactMixin, ChunkMixin, SummaryMixin, SessionStateMixin): def __init__(self, pool: asyncpg.Pool, embedding_backend: "LLMBackend | None" = None) -> None: self._pool = pool self._initialized = False diff --git a/navi/profiles/assistant/config.json b/navi/profiles/assistant/config.json index df0969e..16fae59 100644 --- a/navi/profiles/assistant/config.json +++ b/navi/profiles/assistant/config.json @@ -48,6 +48,7 @@ "code_exec", "terminal", "memory", + "memory_search", "list_tools", "tasks", "tool_manual", @@ -85,6 +86,7 @@ "code_exec", "terminal", "memory", + "memory_search", "list_tools", "tool_manual", "share_file", diff --git a/navi/profiles/designer_3d/config.json b/navi/profiles/designer_3d/config.json index c906859..984c769 100644 --- a/navi/profiles/designer_3d/config.json +++ b/navi/profiles/designer_3d/config.json @@ -48,6 +48,7 @@ "code_exec", "terminal", "memory", + "memory_search", "list_tools", "tasks", "tool_manual", @@ -82,6 +83,7 @@ "filesystem", "image_view", "memory", + "memory_search", "list_tools", "tool_manual" ], diff --git a/navi/profiles/developer/config.json b/navi/profiles/developer/config.json index dbf10be..066c5eb 100644 --- a/navi/profiles/developer/config.json +++ b/navi/profiles/developer/config.json @@ -51,6 +51,7 @@ "terminal", "image_view", "memory", + "memory_search", "reload_tools", "list_tools", "tasks", diff --git a/navi/profiles/discuss/config.json b/navi/profiles/discuss/config.json index 40f6610..e10d310 100644 --- a/navi/profiles/discuss/config.json +++ b/navi/profiles/discuss/config.json @@ -35,6 +35,7 @@ "scratchpad", "reflect", "memory", + "memory_search", "image_view", "todo", "skills", diff --git a/navi/profiles/modeler_3d/config.json b/navi/profiles/modeler_3d/config.json index 7a7d143..0388834 100644 --- a/navi/profiles/modeler_3d/config.json +++ b/navi/profiles/modeler_3d/config.json @@ -52,6 +52,7 @@ "terminal", "image_view", "memory", + "memory_search", "list_tools", "tasks", "tool_manual", diff --git a/navi/profiles/navi_code/config.json b/navi/profiles/navi_code/config.json index 943c085..39a8a80 100644 --- a/navi/profiles/navi_code/config.json +++ b/navi/profiles/navi_code/config.json @@ -55,6 +55,7 @@ "terminal", "image_view", "memory", + "memory_search", "list_tools", "tasks", "tool_manual", diff --git a/navi/profiles/server_admin/config.json b/navi/profiles/server_admin/config.json index af058ae..4d16c2d 100644 --- a/navi/profiles/server_admin/config.json +++ b/navi/profiles/server_admin/config.json @@ -52,6 +52,7 @@ "ssh_exec", "image_view", "memory", + "memory_search", "reload_tools", "list_tools", "tasks", diff --git a/navi/settings_model.py b/navi/settings_model.py index 1802d9d..fbd4a0b 100644 --- a/navi/settings_model.py +++ b/navi/settings_model.py @@ -31,6 +31,22 @@ # 0 disables the cap. embedding_max_chars: int = 6000 + # Unified memory search (`memory_search`) — one query over curated facts, the + # session sketches and the raw turns context compression dropped. The raw + # index is written by the compressor; with this off nothing new is indexed, + # and the tool still answers from facts and sketches. + memory_search_enabled: bool = True + # Automatic recall: when the session already carries a compression summary, + # the top few raw excerpts relevant to the recent turns are injected into the + # next prompt (provenance included). This is the "we discussed this earlier" + # path that does not wait for the model to call a tool. + memory_recall_enabled: bool = True + memory_recall_k: int = 2 + # Floor for a vector-similarity match on an automatic injection. Lexical + # (full-text) matches are judged by how many query terms they contain + # instead, so a cold pgvector install still recalls — see navi/memory/search.py. + memory_recall_min_similarity: float = 0.55 + openai_api_key: str = "" openai_model: str = "gpt-4" openai_base_url: str | None = None diff --git a/navi/store/__init__.py b/navi/store/__init__.py index 32f932c..eb5bc25 100644 --- a/navi/store/__init__.py +++ b/navi/store/__init__.py @@ -98,3 +98,47 @@ "DELETE FROM session_store WHERE user_id = $1 AND session_id = $2 AND scope = $3", uid, session_id, scope, ) + + async def search_values( + self, + user_id: str | None, + scope: str, + terms: list[str], + limit: int = 10, + session_id: str | None = None, + prefer_session_id: str | None = None, + ) -> list[dict]: + """Find scoped values whose key or text mentions any of `terms`. + + Scope-wide (one section per key), so this is the read side the unified + memory search needs for sketches: sections are small and updated in + place, and a session's own section is almost always the relevant one, + hence `prefer_session_id` ordering before recency. + + With no terms the most recently updated rows are returned — a search with + an empty query should list, not find nothing. + """ + uid = _norm_uid(user_id) + params: list[Any] = [uid, scope] + clauses = ["user_id = $1", "scope = $2"] + if session_id is not None: + params.append(session_id) + clauses.append(f"session_id = ${len(params)}") + if terms: + params.append([f"%{t}%" for t in terms]) + idx = len(params) + clauses.append(f"(key ILIKE ANY(${idx}::text[]) OR value ILIKE ANY(${idx}::text[]))") + if prefer_session_id is not None: + params.append(prefer_session_id) + order = f"(session_id = ${len(params)}) DESC, updated_at DESC" + else: + order = "updated_at DESC" + params.append(limit) + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + f"SELECT session_id, key, value, updated_at FROM session_store " + f"WHERE {' AND '.join(clauses)} ORDER BY {order} LIMIT ${len(params)}", + *params, + ) + return [dict(row) for row in rows] diff --git a/navi/tools/__init__.py b/navi/tools/__init__.py index 485f807..aaf0eb2 100644 --- a/navi/tools/__init__.py +++ b/navi/tools/__init__.py @@ -9,6 +9,7 @@ from .spawn_agent import SpawnAgentTool from .terminal import TerminalTool from .memory import MemoryTool +from .memory_search import MemorySearchTool from .test_mcp_tool import TestMcpToolTool from .todo import TodoTool from .scratchpad import ScratchpadTool @@ -32,6 +33,7 @@ "ImageViewTool", "ManageRecallTool", "MemoryTool", + "MemorySearchTool", "ScheduleRecallTool", "TestMcpToolTool", "SpawnAgentTool", diff --git a/navi/tools/memory_search.py b/navi/tools/memory_search.py new file mode 100644 index 0000000..9412f5b --- /dev/null +++ b/navi/tools/memory_search.py @@ -0,0 +1,85 @@ +"""`memory_search` — one query over facts, session sketches and raw history.""" + +from __future__ import annotations + +from navi.memory.search import UnifiedMemorySearch, format_hits +from navi.tools._internal.base import current_session_id, current_user_id + +from ._internal.base import Tool, ToolContext, ToolResult + + +class MemorySearchTool(Tool): + name = "memory_search" + description = ( + "Search everything this Navi remembers, in one call: curated facts about the user, " + "the session scratchpads, and verbatim excerpts of earlier turns that context " + "compression dropped. Use it when the user refers back to something you can no " + "longer see — 'as we discussed', 'the port we picked', 'that file', a name, a " + "number, a link, a decision — before saying you do not remember. " + "Every result carries its source: a fact (curated, rewritten by key), a raw excerpt " + "('session X, turn Y'), or a sketch. Prefer a newer fact over an older raw excerpt; " + "when the two disagree the result says which one is outdated." + ) + parameters = { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": ( + "What to look for, in the words of the thing itself — a name, a file, " + "a command, a value, the topic. Full sentences work too." + ), + }, + "scope": { + "type": "string", + "enum": ["all", "facts", "raw", "sketches"], + "description": ( + "Which store to read. 'all' (default) searches all three and returns " + "facts first, then relevant raw excerpts, then sketches. " + "'raw' = verbatim history, including turns compressed out of view." + ), + }, + "k": { + "type": "integer", + "description": "How many results per store to return (1-20, default 5).", + "minimum": 1, + "maximum": 20, + }, + }, + "required": ["query"], + } + + def __init__(self, search: UnifiedMemorySearch) -> None: + self._search = search + + async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult: + query = (params.get("query") or "").strip() + if not query: + return ToolResult( + success=False, output="query is required.", error="missing query" + ) + scope = (params.get("scope") or "all").strip().lower() + if scope not in ("all", "facts", "raw", "sketches"): + scope = "all" + try: + k = int(params.get("k") or 5) + except (TypeError, ValueError): + k = 5 + k = max(1, min(k, 20)) + + session_id = (ctx.session_id if ctx else None) or current_session_id.get() + user_id = (ctx.user_id if ctx else None) + if user_id is None: + user_id = current_user_id.get(None) + + summary_at = await self._search.latest_summary_at(session_id) if session_id else None + + hits = await self._search.search( + query, + scope=scope, + k=k, + user_id=user_id, + current_session_id=session_id, + summary_at=summary_at, + ) + return ToolResult(success=True, output=format_hits(hits, query, scope)) diff --git a/navi/workers/base.py b/navi/workers/base.py index 625f063..87de60a 100644 --- a/navi/workers/base.py +++ b/navi/workers/base.py @@ -32,6 +32,10 @@ # the prompt. ``context_tokens`` is then the truncated size, not the # context's, so threshold gates must not trust it. context_truncated: bool = False + # Memory store the compressor writes dropped turns into, so the raw-chunk + # index is filled by every compression path (pre/mid/post-turn), not just + # the ones the agent runs itself. + memory_store: object | None = None @dataclass diff --git a/navi/workers/compressor.py b/navi/workers/compressor.py index b5e0c38..312cbfb 100644 --- a/navi/workers/compressor.py +++ b/navi/workers/compressor.py @@ -63,6 +63,7 @@ # The window the gate above just fired against — the profile's # own model window, not the global default. window=ctx.max_context_tokens, + memory_store=ctx.memory_store, ) except Exception: # compress_and_save_session already retries and hard-truncates diff --git a/tests/unit/core/test_agent.py b/tests/unit/core/test_agent.py index e06a757..432a1b4 100644 --- a/tests/unit/core/test_agent.py +++ b/tests/unit/core/test_agent.py @@ -6,6 +6,7 @@ import asyncio import copy +import json from datetime import datetime, timezone from unittest.mock import MagicMock @@ -1456,3 +1457,60 @@ profile = make_profile("test", context_window=None) assert agent._window_for(profile) == agent_mod.settings.ollama_num_ctx + + +class TestMemoryRecallInjection: + """The compression-dropped excerpt path reaches the built context.""" + + @pytest.mark.asyncio + async def test_recalled_excerpt_is_injected_before_the_turn(self, monkeypatch): + from datetime import datetime, timezone + + from navi.memory.search import MemoryHit + + class FakeSearch: + async def search(self, query, **kwargs): + return [MemoryHit( + source_type="raw", text="the deploy port is 8099", + session_id="s1", start_seq=3, end_seq=4, chunk_id="c1", score=0.9, + timestamp=datetime(2026, 11, 9, tzinfo=timezone.utc), + )] + + async def latest_summary_at(self, session_id): + return None + + sessions = InMemorySessionStore() + profiles = ProfileRegistry() + profiles.register(make_profile("test")) + backends = BackendRegistry() + backends.register("ollama", FakeLLMBackend(responses=["hello"])) + agent = Agent( + session_store=sessions, + profile_registry=profiles, + tool_registry=make_registry_with_tools(), + backend_registry=backends, + memory_search=FakeSearch(), + ) + session = await sessions.create(profile_id="test") + session.id = "s1" + session.context.append(Message(role="user", content="summary", is_summary=True)) + await sessions.save(session) + + seen: dict = {} + original = agent._ctx_builder.build + + def spy(*args, **kwargs): + seen["extra"] = kwargs.get("extra_system") + return original(*args, **kwargs) + + monkeypatch.setattr(agent._ctx_builder, "build", spy) + await agent.run(session.id, "what was the port?") + + extra = seen.get("extra") or [] + recall = [m for m in extra if m.content and m.content.startswith("## Recovered")] + assert recall, "the raw excerpt was not injected into the built context" + assert "the deploy port is 8099" in recall[0].content + # The dedupe ledger lives in session_metadata, which PgSessionStore + # json-encodes on every save — a set there would break persistence. + assert session.session_metadata["injected_chunk_ids"] == ["c1"] + json.dumps(session.session_metadata) diff --git a/tests/unit/core/test_compression_breadcrumbs.py b/tests/unit/core/test_compression_breadcrumbs.py new file mode 100644 index 0000000..43c9606 --- /dev/null +++ b/tests/unit/core/test_compression_breadcrumbs.py @@ -0,0 +1,124 @@ +"""Compression breadcrumbs: the KEYS line and the raw-chunk index it feeds.""" + +from types import SimpleNamespace + +import navi.config as _config +import navi.core.compressor as _cm +from navi.config import Settings +from navi.core.compressor import ContextCompressor, compress_context +from navi.llm.base import Message +from tests.conftest_factory import FakeLLMBackend + + +def _conversation() -> list[Message]: + return [ + Message(role="system", content="sys"), + Message(role="user", content="use `uv run pytest` in navi/core/compressor.py", sequence_number=1), + Message(role="assistant", content="ok, port 8099", sequence_number=2), + Message(role="user", content="next", sequence_number=3), + Message(role="assistant", content="next answer", sequence_number=4), + Message(role="user", content="latest", sequence_number=5), + Message(role="assistant", content="latest answer", sequence_number=6), + ] + + +class TestKeysLine: + async def test_the_keys_line_is_stripped_and_stored_on_the_summary(self): + backend = FakeLLMBackend(responses=[ + "## Goal\ndo it\n\nKEYS: `uv run pytest`, navi/core/compressor.py, port 8099" + ]) + new_context, summary = await compress_context( + context=_conversation(), llm=backend, model="test", temperature=0.3, keep_recent=2, + ) + assert "KEYS" not in summary + assert summary.startswith("## Goal") + summary_msg = next(m for m in new_context if m.is_summary) + assert summary_msg.metadata["breadcrumbs"] == [ + "uv run pytest", "navi/core/compressor.py", "port 8099", + ] + + async def test_a_summary_without_keys_keeps_empty_metadata(self): + backend = FakeLLMBackend(responses=["plain summary"]) + new_context, summary = await compress_context( + context=_conversation(), llm=backend, model="test", temperature=0.3, keep_recent=2, + ) + assert summary == "plain summary" + assert next(m for m in new_context if m.is_summary).metadata == {} + + async def test_a_keys_only_response_does_not_become_an_empty_summary(self): + backend = FakeLLMBackend(responses=["KEYS: alpha, beta"]) + _, summary = await compress_context( + context=_conversation(), llm=backend, model="test", temperature=0.3, keep_recent=2, + ) + assert summary == "(summary unavailable)" + + +class FakeChunkStore: + def __init__(self): + self.calls = [] + + async def add_chunks(self, chunks, user_id=None, session_id=None): + self.calls.append((chunks, user_id, session_id)) + return len(chunks) + + +class TestIndexDemoted: + def _demoted(self) -> list[Message]: + return [ + Message(role="user", content="run `navi_search --port 8099`", sequence_number=1), + Message(role="assistant", content="done, see navi/memory/_chunks.py", sequence_number=2), + Message(role="user", content="old summary text", sequence_number=3, is_summary=True), + ] + + async def test_chunks_are_written_with_the_summary_keys_and_extracted_names(self): + store = FakeChunkStore() + summary_msg = Message( + role="user", content="summary", is_summary=True, + metadata={"breadcrumbs": ["from_summary_key"]}, + ) + session = SimpleNamespace(user_id="u1") + compressor = ContextCompressor() + + await compressor._index_demoted(store, session, "s1", self._demoted(), summary_msg) + + chunks, user_id, session_id = store.calls[0] + assert (user_id, session_id) == ("u1", "s1") + assert len(chunks) == 1 + keys = chunks[0]["breadcrumbs"] + assert "from_summary_key" in keys + assert "navi/memory/_chunks.py" in keys + assert "navi_search --port 8099" in keys + assert "old summary text" not in chunks[0]["content"] + + async def test_the_hard_truncate_path_falls_back_to_the_extractor(self): + store = FakeChunkStore() + await ContextCompressor()._index_demoted( + store, SimpleNamespace(user_id=None), "s1", self._demoted(), None, + ) + keys = store.calls[0][0][0]["breadcrumbs"] + assert "navi/memory/_chunks.py" in keys + + async def test_indexing_can_be_switched_off(self, monkeypatch): + new = Settings(_env_file=None, memory_search_enabled=False) + monkeypatch.setattr(_config, "settings", new) + monkeypatch.setattr(_cm, "settings", new) + store = FakeChunkStore() + await ContextCompressor()._index_demoted( + store, SimpleNamespace(user_id="u1"), "s1", self._demoted(), None, + ) + assert store.calls == [] + + async def test_a_broken_store_never_breaks_compression(self): + class Broken(FakeChunkStore): + async def add_chunks(self, chunks, user_id=None, session_id=None): + raise RuntimeError("db down") + + # No exception escapes: losing an index entry is cheaper than losing a turn. + await ContextCompressor()._index_demoted( + Broken(), SimpleNamespace(user_id="u1"), "s1", self._demoted(), None, + ) + + async def test_no_messages_means_no_call(self): + store = FakeChunkStore() + await ContextCompressor()._index_demoted(store, SimpleNamespace(user_id="u1"), "s1", [], None) + assert store.calls == [] diff --git a/tests/unit/core/test_memory_recall.py b/tests/unit/core/test_memory_recall.py new file mode 100644 index 0000000..39ee9c8 --- /dev/null +++ b/tests/unit/core/test_memory_recall.py @@ -0,0 +1,154 @@ +"""Automatic recall of raw excerpts — the read side of the compression index.""" + +from datetime import datetime, timezone +from types import SimpleNamespace + +import navi.config as _config +import navi.core.context_builder as _cb +from navi.config import Settings +from navi.core.context_builder import ContextBuilder, _recall_query +from navi.llm.base import Message +from navi.memory.search import MemoryHit +from tests.conftest_factory import make_profile_registry + +NOW = datetime(2026, 11, 9, 12, 0, tzinfo=timezone.utc) + + +class FakeSearch: + def __init__(self, hits=None): + self.hits = hits or [] + self.calls = [] + + async def search(self, query, **kwargs): + self.calls.append((query, kwargs)) + return self.hits + + async def latest_summary_at(self, session_id): + return NOW + + +def _session(*, with_summary: bool, context=None): + messages = context or [ + Message(role="user", content="we discussed the deploy port"), + Message(role="assistant", content="the deploy port is 8099"), + ] + if with_summary: + messages = [Message(role="user", content="[Context Summary]", is_summary=True), *messages] + return SimpleNamespace(id="s1", context=messages, session_metadata={}) + + +def _hit(text="the deploy port is 8099", **kwargs): + defaults = dict( + source_type="raw", text=text, session_id="s1", + start_seq=3, end_seq=4, timestamp=NOW, chunk_id="c1", score=0.9, + ) + defaults.update(kwargs) + return MemoryHit(**defaults) + + +class TestQueryWindow: + def test_the_query_uses_the_window_not_only_the_last_message(self): + context = [ + Message(role="user", content="we discussed the deploy port"), + Message(role="assistant", content="the deploy port is 8099"), + ] + query = _recall_query(context, "and what about it?") + assert "deploy port" in query + assert "8099" in query + assert "and what about it?" in query + + def test_summaries_are_kept_out_of_the_query(self): + context = [ + Message(role="user", content="a giant old summary", is_summary=True), + Message(role="user", content="fresh question"), + ] + assert "giant old summary" not in _recall_query(context, "next") + + def test_the_query_is_bounded(self): + context = [Message(role="user", content="x" * 10_000)] + assert len(_recall_query(context, "y" * 10_000)) <= 1200 + + +class TestRecallsOnlyAfterACompression: + async def test_no_summary_means_no_search(self): + search = FakeSearch([_hit()]) + builder = ContextBuilder(make_profile_registry(), memory_search=search) + msg = await builder._memory_recall_msg(_session(with_summary=False), "the port") + assert msg is None + assert search.calls == [] + + async def test_with_a_summary_the_session_is_searched(self): + search = FakeSearch([_hit()]) + builder = ContextBuilder(make_profile_registry(), memory_search=search) + msg = await builder._memory_recall_msg(_session(with_summary=True), "the port", user_id="u1") + assert msg is not None + _, kwargs = search.calls[0] + assert kwargs["scope"] == "raw" + assert kwargs["session_id"] == "s1" + assert kwargs["user_id"] == "u1" + + async def test_recall_can_be_switched_off(self, monkeypatch): + new = Settings(_env_file=None, memory_recall_enabled=False) + monkeypatch.setattr(_config, "settings", new) + monkeypatch.setattr(_cb, "_config", SimpleNamespace(settings=new)) + search = FakeSearch([_hit()]) + builder = ContextBuilder(make_profile_registry(), memory_search=search) + assert await builder._memory_recall_msg(_session(with_summary=True), "port") is None + assert search.calls == [] + + +class TestSelection: + async def test_an_excerpt_is_rendered_with_its_source(self): + builder = ContextBuilder(make_profile_registry(), memory_search=FakeSearch([_hit()])) + msg = await builder._memory_recall_msg(_session(with_summary=True), "port") + assert msg.role == "system" + assert "session s1, turns 3–4" in msg.content + assert "the deploy port is 8099" in msg.content + assert "verbatim" in msg.content + + async def test_the_same_chunk_is_not_injected_twice(self): + injected: list[str] = [] + builder = ContextBuilder(make_profile_registry(), memory_search=FakeSearch([_hit()])) + first = await builder._memory_recall_msg(_session(with_summary=True), "port", injected_ids=injected) + second = await builder._memory_recall_msg(_session(with_summary=True), "port", injected_ids=injected) + assert first is not None + assert second is None + assert injected == ["c1"] + + async def test_a_weak_vector_match_is_dropped(self, monkeypatch): + new = Settings(_env_file=None, memory_recall_min_similarity=0.6) + monkeypatch.setattr(_cb, "_config", SimpleNamespace(settings=new)) + builder = ContextBuilder( + make_profile_registry(), memory_search=FakeSearch([_hit(score=0.4)]) + ) + assert await builder._memory_recall_msg(_session(with_summary=True), "port") is None + + async def test_the_lexical_fallback_needs_two_terms(self): + weak = _hit(score=None, terms_hit=1, chunk_id="c1") + strong = _hit(score=None, terms_hit=2, chunk_id="c2", text="the port is in the deploy notes") + builder = ContextBuilder(make_profile_registry(), memory_search=FakeSearch([weak, strong])) + msg = await builder._memory_recall_msg(_session(with_summary=True), "port") + assert "deploy notes" in msg.content + assert msg.content.count("•") == 1 + + async def test_an_outdated_excerpt_is_not_injected(self): + builder = ContextBuilder(make_profile_registry(), memory_search=FakeSearch( + [_hit(superseded_by="server_port (updated later)")] + )) + assert await builder._memory_recall_msg(_session(with_summary=True), "port") is None + + async def test_at_most_k_excerpts(self, monkeypatch): + new = Settings(_env_file=None, memory_recall_k=1) + monkeypatch.setattr(_cb, "_config", SimpleNamespace(settings=new)) + hits = [_hit(chunk_id="c1"), _hit("second", chunk_id="c2")] + builder = ContextBuilder(make_profile_registry(), memory_search=FakeSearch(hits)) + msg = await builder._memory_recall_msg(_session(with_summary=True), "port") + assert "second" not in msg.content + + async def test_a_search_failure_is_swallowed(self): + class Broken(FakeSearch): + async def search(self, query, **kwargs): + raise RuntimeError("db down") + + builder = ContextBuilder(make_profile_registry(), memory_search=Broken()) + assert await builder._memory_recall_msg(_session(with_summary=True), "port") is None diff --git a/tests/unit/memory/test_breadcrumbs.py b/tests/unit/memory/test_breadcrumbs.py new file mode 100644 index 0000000..decbaaa --- /dev/null +++ b/tests/unit/memory/test_breadcrumbs.py @@ -0,0 +1,84 @@ +"""Breadcrumb keys — the index a later search asks by.""" + +from navi.memory.breadcrumbs import ( + MAX_BREADCRUMBS, + extract_breadcrumbs, + split_keys_line, +) + + +class TestSplitKeysLine: + def test_the_trailing_keys_line_is_parsed_and_removed(self): + text = "## Goal\nDo the thing.\n\nKEYS: navi/server.py, pytest, port 8099" + clean, keys = split_keys_line(text) + assert clean == "## Goal\nDo the thing." + assert keys == ["navi/server.py", "pytest", "port 8099"] + + def test_bold_and_lowercase_spellings_are_recognised(self): + for line in ("**KEYS**: a, b", "keys: a, b", "KEYS: a; b"): + _, keys = split_keys_line(f"summary\n{line}") + assert keys == ["a", "b"], line + + def test_none_yields_no_keys(self): + clean, keys = split_keys_line("summary\nKEYS: NONE") + assert keys == [] + assert clean == "summary" + + def test_backticks_and_bullets_are_stripped(self): + _, keys = split_keys_line("s\nKEYS: `memory_search`, - `navi/memory/_chunks.py`, " + "x" * 200) + assert keys == ["memory_search", "navi/memory/_chunks.py"] + + def test_a_summary_without_the_line_is_untouched(self): + text = "## Goal\nNothing to index here." + assert split_keys_line(text) == (text, []) + + def test_only_the_last_line_counts(self): + # The model may mention KEYS in prose; the index is the line at the end. + text = "KEYS: not-this-one\n## Goal\nbody\nKEYS: real_key" + clean, keys = split_keys_line(text) + assert keys == ["real_key"] + assert "not-this-one" in clean + + +class TestExtractBreadcrumbs: + def test_names_commands_paths_and_amounts_are_found(self): + text = ( + "We ran `uv run pytest -q` in navi/core/compressor.py and it failed:\n" + "AssertionError: memory_search returned 0 hits\n" + "The budget is 4,5 GB and ticket #522 is closed." + ) + found = extract_breadcrumbs([text]) + joined = " | ".join(found) + assert "uv run pytest -q" in joined + assert "navi/core/compressor.py" in joined + assert "memory_search" in joined + assert "#522" in joined + assert any("4,5 GB" in item for item in found) + + def test_identifiers_and_urls(self): + found = extract_breadcrumbs( + ["See https://tasks.gnexus.space/tasks/522 and call MemorySearchTool.execute."] + ) + assert "https://tasks.gnexus.space/tasks/522" in found + assert "MemorySearchTool" in found + + def test_duplicates_collapse_case_insensitively(self): + found = extract_breadcrumbs(["`navi/config.py` and NAVI/CONFIG.PY"]) + assert len([f for f in found if f.lower() == "navi/config.py"]) == 1 + + def test_the_limit_is_respected(self): + text = " ".join(f"`thing_{index}`" for index in range(60)) + assert len(extract_breadcrumbs([text], limit=5)) == 5 + assert extract_breadcrumbs([text], limit=0) == [] + + def test_the_cap_is_not_exceeded_by_default(self): + text = " ".join(f"`name_{index}`" for index in range(MAX_BREADCRUMBS + 20)) + assert len(extract_breadcrumbs([text])) == MAX_BREADCRUMBS + + def test_empty_text_produces_nothing(self): + assert extract_breadcrumbs([]) == [] + assert extract_breadcrumbs(["", None]) == [] # type: ignore[list-item] + + def test_extraction_is_deterministic(self): + text = "`alpha`, `/tmp/x.py`, snake_case_name, 12 ms" + assert extract_breadcrumbs([text]) == extract_breadcrumbs([text]) diff --git a/tests/unit/memory/test_chunks.py b/tests/unit/memory/test_chunks.py new file mode 100644 index 0000000..8570182 --- /dev/null +++ b/tests/unit/memory/test_chunks.py @@ -0,0 +1,193 @@ +"""Raw chunks: building them from dropped messages, storing them once, finding them.""" + +from datetime import datetime, timezone + +from navi.llm.base import Message +from navi.memory._chunks import build_chunks +from tests.conftest_factory import FakeConnection, make_store_with_pool + + +def _msg(role: str, content: str, seq: int = 0, **kwargs) -> Message: + return Message(role=role, content=content, sequence_number=seq, **kwargs) + + +class TestBuildChunks: + def test_marks_summary_and_compression_rows_are_not_indexed(self): + messages = [ + _msg("user", "hello", 1), + _msg("user", "old summary", 2, is_summary=True), + _msg("system", "compression marker", 3, is_compression=True), + _msg("system", "plain system", 4), + _msg("assistant", "hi", 5), + ] + chunks = build_chunks(messages) + assert len(chunks) == 1 + assert "old summary" not in chunks[0]["content"] + assert "compression marker" not in chunks[0]["content"] + assert "hello" in chunks[0]["content"] + assert "hi" in chunks[0]["content"] + + def test_turns_are_packed_until_the_size_cap(self): + messages = [_msg("user", "x" * 60, index) for index in range(10)] + chunks = build_chunks(messages, max_chars=150) + assert len(chunks) == 5 # two messages per chunk + assert all(len(c["content"]) <= 150 for c in chunks) + assert [c["start_seq"] for c in chunks] == [0, 2, 4, 6, 8] + + def test_the_turn_range_tracks_the_packed_messages(self): + messages = [_msg("user", "a", 7), _msg("assistant", "b", 8), _msg("tool", "c", 9, name="grep")] + chunk = build_chunks(messages)[0] + assert chunk["start_seq"] == 7 + assert chunk["end_seq"] == 9 + assert "Tool result (grep): c" in chunk["content"] + + def test_a_huge_message_is_elided_in_the_middle(self): + body = "HEAD " + "x" * 20_000 + " TAIL" + chunk = build_chunks([_msg("tool", body, 1)])[0] + assert len(chunk["content"]) < 13_000 + assert chunk["content"].startswith("Tool result: HEAD") + assert chunk["content"].endswith("TAIL") + assert "middle elided" in chunk["content"] + + def test_timestamps_come_from_the_messages(self): + early = datetime(2026, 1, 1, tzinfo=timezone.utc) + late = datetime(2026, 1, 2, tzinfo=timezone.utc) + chunk = build_chunks([ + _msg("user", "a", 1, created_at=early), + _msg("assistant", "b", 2, created_at=late), + ])[0] + assert chunk["started_at"] == early + assert chunk["ended_at"] == late + + def test_the_hash_is_stable_and_content_addressed(self): + first = build_chunks([_msg("user", "same", 1)])[0] + second = build_chunks([_msg("user", "same", 99)])[0] + assert first["content_hash"] == second["content_hash"] + + def test_an_assistant_tool_request_without_text_is_recorded(self): + from navi.llm.base import ToolCallRequest + + msg = Message( + role="assistant", + content="", + sequence_number=1, + tool_calls=[ToolCallRequest(id="1", name="grep", arguments={})], + ) + chunks = build_chunks([msg]) + assert chunks and "grep" in chunks[0]["content"] + + +class TestAddChunks: + async def test_chunks_are_inserted_with_scope_and_breadcrumbs(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + store._pgvector_checked = True + store._pgvector_available = True + conn.enqueue("INSERT 0 1") + conn.enqueue("INSERT 0 1") + chunks = build_chunks( + [_msg("user", "first", 1), _msg("user", "second", 2)], max_chars=10 + ) + for chunk in chunks: + chunk["breadcrumbs"] = ["port 8099"] + written = await store.add_chunks(chunks, user_id="u1", session_id="s1") + assert written == 2 + assert all("memory_chunks" in call[1] for call in conn.calls) + assert "ON CONFLICT (session_id, content_hash) DO NOTHING" in conn.calls[0][1] + assert conn.calls[0][2][1] == "u1" + assert conn.calls[0][2][2] == "s1" + assert conn.calls[0][2][8] == ["port 8099"] + + async def test_without_pgvector_the_write_is_text_only(self): + # `::vector` would fail on a plain Postgres; the row must not name the column. + conn = FakeConnection() + store = make_store_with_pool(conn) + store._pgvector_checked = True + store._pgvector_available = False + conn.enqueue("INSERT 0 1") + written = await store.add_chunks( + build_chunks([_msg("user", "no vectors here", 1)]), session_id="s1" + ) + assert written == 1 + assert "embedding" not in conn.calls[0][1] + + async def test_a_conflict_is_not_counted_as_written(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + store._pgvector_checked = True + store._pgvector_available = True + conn.enqueue("INSERT 0 0") + written = await store.add_chunks( + build_chunks([_msg("user", "again", 1)]), session_id="s1" + ) + assert written == 0 + + async def test_no_session_means_nothing_is_written(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + assert await store.add_chunks(build_chunks([_msg("user", "x", 1)])) == 0 + assert conn.calls == [] + + +class TestSearchChunks: + async def test_lexical_fallback_ranks_by_matched_terms(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + conn.enqueue([ + { + "rank": 0.5, "id": "a", "session_id": "s1", "start_seq": 1, "end_seq": 2, + "started_at": None, "ended_at": datetime(2026, 1, 1), + "content": "the port is 8099 and the host is localhost", + "breadcrumbs": ["port"], "created_at": datetime(2026, 1, 1), + }, + { + "rank": 0.1, "id": "b", "session_id": "s1", "start_seq": 3, "end_seq": 4, + "started_at": None, "ended_at": datetime(2026, 1, 1), + "content": "some port talk without the value", + "breadcrumbs": [], "created_at": datetime(2026, 1, 1), + }, + ]) + hits = await store.search_chunks("port host localhost", user_id="u1", limit=5) + assert [h["id"] for h in hits] == ["a", "b"] + assert hits[0]["terms_hit"] == 3 + assert hits[1]["terms_hit"] == 1 + assert hits[0]["score"] is None + + async def test_the_scope_filter_is_applied(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + conn.enqueue([]) + await store.search_chunks("q", user_id=None, session_id="s1", limit=3) + query, args = conn.calls[0][1], conn.calls[0][2] + assert "user_id IS NULL" in query + assert "session_id = $1" in query + assert args[0] == "s1" + + async def test_an_empty_term_list_falls_back_to_recent(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + conn.enqueue([ + { + "id": "x", "session_id": "s1", "start_seq": 1, "end_seq": 1, + "started_at": None, "ended_at": None, "content": "body", + "breadcrumbs": [], "created_at": None, + }, + ]) + hits = await store.search_chunks("!!!", user_id="u1", limit=2) + assert hits[0]["id"] == "x" + + +class TestLatestSummaryAt: + async def test_returns_the_timestamp(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + when = datetime(2026, 2, 3, tzinfo=timezone.utc) + conn.enqueue(when) + assert await store.latest_summary_at("s1") == when + assert "is_summary = TRUE" in conn.calls[0][1] + + async def test_a_failure_is_swallowed(self): + conn = FakeConnection() + store = make_store_with_pool(conn) + conn.enqueue(RuntimeError("boom")) + assert await store.latest_summary_at("s1") is None diff --git a/tests/unit/memory/test_memory_ddl.py b/tests/unit/memory/test_memory_ddl.py index f9d31e1..3d0e7d0 100644 --- a/tests/unit/memory/test_memory_ddl.py +++ b/tests/unit/memory/test_memory_ddl.py @@ -38,3 +38,26 @@ for stmt in _build_ddl(pgvector): if "CREATE INDEX" in stmt: assert "merged_into" not in stmt, f"an index mentions merged_into: {stmt[:80]}" + + +def test_memory_chunks_embedding_is_migrated_before_its_index(): + """A database that first ran without pgvector holds a `memory_chunks` table + with no `embedding` column; CREATE TABLE IF NOT EXISTS never adds it, and + every chunk INSERT then fails on the missing column. The ALTER has to come + before the HNSW index, which would otherwise fail on the same table. + """ + ddl = _build_ddl(True) + alter = next( + (i for i, stmt in enumerate(ddl) + if "ALTER TABLE memory_chunks ADD COLUMN IF NOT EXISTS embedding" in stmt), + None, + ) + index = next( + (i for i, stmt in enumerate(ddl) if "idx_memory_chunks_embedding" in stmt), + None, + ) + assert alter is not None, "the chunks embedding column is never migrated" + assert index is not None and alter < index + assert not any( + "ALTER TABLE memory_chunks ADD COLUMN" in stmt for stmt in _build_ddl(False) + ), "a database without pgvector must not get a vector column" diff --git a/tests/unit/memory/test_memory_search.py b/tests/unit/memory/test_memory_search.py new file mode 100644 index 0000000..465a812 --- /dev/null +++ b/tests/unit/memory/test_memory_search.py @@ -0,0 +1,214 @@ +"""Unified memory search: tiered merge, provenance, freshness.""" + +from datetime import datetime, timedelta, timezone + +from navi.memory.search import UnifiedMemorySearch, format_hits + +NOW = datetime(2026, 11, 9, 12, 0, tzinfo=timezone.utc) + + +class FakeMemory: + def __init__(self, facts=None, chunks=None, summary_at=None): + self.facts = facts or [] + self.chunks = chunks or [] + self.summary_at = summary_at + self.calls = [] + + async def search_facts(self, query, user_id=None, limit=15): + self.calls.append(("facts", user_id, limit)) + return self.facts[:limit] + + async def search_chunks(self, query, user_id=None, session_id=None, limit=8): + self.calls.append(("raw", user_id, session_id, limit)) + return self.chunks[:limit] + + async def latest_summary_at(self, session_id): + return self.summary_at + + +class FakeKv: + def __init__(self, rows=None): + self.rows = rows or [] + self.calls = [] + + async def search_values(self, user_id, scope, terms, limit=10, session_id=None, prefer_session_id=None): + self.calls.append((user_id, scope, tuple(terms), limit, prefer_session_id)) + return self.rows[:limit] + + +def _fact(key, value, updated=NOW, category="technical"): + return { + "id": key, "category": category, "key": key, "value": value, + "updated_at": updated, "source_session_id": "sess-fact", + } + + +def _chunk(content, session="s1", start=1, end=2, ended=NOW, score=0.9): + return { + "id": f"chunk-{content[:6]}", "session_id": session, + "start_seq": start, "end_seq": end, "started_at": None, + "ended_at": ended, "content": content, "breadcrumbs": ["port"], + "created_at": ended, "score": score, "terms_hit": 0, + } + + +class TestTiering: + async def test_facts_come_first_then_raw_then_sketches(self): + memory = FakeMemory(facts=[_fact("server_port", "8099")], chunks=[_chunk("port was 8098")]) + kv = FakeKv([{"session_id": "s1", "key": "findings", "value": "port 8098", "updated_at": NOW}]) + search = UnifiedMemorySearch(memory, kv) + + hits = await search.search("port", user_id="u1") + + assert [h.source_type for h in hits] == ["fact", "raw", "sketch"] + + async def test_each_scope_reads_only_its_store(self): + memory = FakeMemory(facts=[_fact("k", "v")], chunks=[_chunk("c")]) + kv = FakeKv([{"session_id": "s1", "key": "k", "value": "v", "updated_at": NOW}]) + search = UnifiedMemorySearch(memory, kv) + + assert [h.source_type for h in await search.search("q", scope="facts")] == ["fact"] + assert [h.source_type for h in await search.search("q", scope="raw")] == ["raw"] + assert [h.source_type for h in await search.search("q", scope="sketches")] == ["sketch"] + + async def test_an_unknown_scope_behaves_like_all(self): + memory = FakeMemory(facts=[_fact("k", "v")]) + search = UnifiedMemorySearch(memory, FakeKv()) + hits = await search.search("q", scope="nonsense") + assert [h.source_type for h in hits] == ["fact"] + + async def test_an_empty_query_returns_nothing_without_touching_stores(self): + memory = FakeMemory(facts=[_fact("k", "v")]) + search = UnifiedMemorySearch(memory, FakeKv()) + assert await search.search(" ") == [] + assert memory.calls == [] + + async def test_k_is_clamped(self): + memory = FakeMemory(facts=[_fact("k", "v")]) + kv = FakeKv() + search = UnifiedMemorySearch(memory, kv) + await search.search("q", k=999) + await search.search("q", k=0) # falsy k falls back to the default + fact_calls = [c for c in memory.calls if c[0] == "facts"] + assert fact_calls[0][2] == 20 + assert fact_calls[1][2] == 5 + + async def test_a_failing_store_does_not_sink_the_others(self): + class BrokenMemory(FakeMemory): + async def search_facts(self, query, user_id=None, limit=15): + raise RuntimeError("db down") + + memory = BrokenMemory(chunks=[_chunk("recoverable")]) + search = UnifiedMemorySearch(memory, FakeKv()) + hits = await search.search("q") + assert [h.source_type for h in hits] == ["raw"] + + +class TestProvenance: + async def test_a_fact_carries_category_key_and_update_time(self): + search = UnifiedMemorySearch(FakeMemory(facts=[_fact("server_port", "8099")]), FakeKv()) + hit = (await search.search("port"))[0] + assert hit.category == "technical" + assert hit.key == "server_port" + assert hit.timestamp == NOW + assert hit.session_id == "sess-fact" + + async def test_raw_carries_session_and_turn_range(self): + search = UnifiedMemorySearch(FakeMemory(chunks=[_chunk("body", start=12, end=14)]), FakeKv()) + hit = (await search.search("body"))[0] + assert hit.session_id == "s1" + assert (hit.start_seq, hit.end_seq) == (12, 14) + assert hit.breadcrumbs == ("port",) + + async def test_a_sketch_carries_its_section(self): + kv = FakeKv([{"session_id": "s9", "key": "findings", "value": "x", "updated_at": NOW}]) + hit = (await UnifiedMemorySearch(FakeMemory(), kv).search("x"))[0] + assert (hit.session_id, hit.key) == ("s9", "findings") + + +class TestFreshness: + async def test_a_newer_fact_supersedes_an_older_raw_excerpt(self): + memory = FakeMemory( + facts=[_fact("server_port", "8099 is the server port", updated=NOW)], + chunks=[_chunk("the server port was 8098", ended=NOW - timedelta(days=3))], + ) + hit = (await UnifiedMemorySearch(memory, FakeKv()).search("server port"))[1] + assert hit.superseded_by is not None + assert "server_port" in hit.superseded_by + + async def test_an_older_fact_does_not_supersede_a_newer_excerpt(self): + memory = FakeMemory( + facts=[_fact("server_port", "8099 is the server port", updated=NOW - timedelta(days=3))], + chunks=[_chunk("the server port is 8100 now", ended=NOW)], + ) + hit = (await UnifiedMemorySearch(memory, FakeKv()).search("server port"))[1] + assert hit.superseded_by is None + + async def test_unrelated_facts_do_not_supersede(self): + memory = FakeMemory( + facts=[_fact("editor", "neovim", updated=NOW)], + chunks=[_chunk("the server port was 8098", ended=NOW - timedelta(days=1))], + ) + hit = (await UnifiedMemorySearch(memory, FakeKv()).search("server port"))[1] + assert hit.superseded_by is None + + async def test_raw_older_than_the_current_summary_is_marked(self): + memory = FakeMemory( + chunks=[_chunk("old turn", ended=NOW - timedelta(days=2))], + summary_at=NOW, + ) + hit = (await UnifiedMemorySearch(memory, FakeKv()).search( + "old", current_session_id="s1", summary_at=NOW, + ))[0] + assert hit.before_summary is True + + async def test_iso_string_timestamps_are_normalized(self): + # `search_facts` returns `updated_at` as an ISO string (the facts store + # formats it for the API), while chunks carry real datetimes. Comparing + # the two used to raise TypeError and take the whole search down. + memory = FakeMemory( + facts=[_fact("server_port", "8099 is the server port", updated=NOW.isoformat())], + chunks=[_chunk("the server port was 8098", ended=NOW - timedelta(days=3))], + ) + hits = await UnifiedMemorySearch(memory, FakeKv()).search("server port") + fact, raw = hits[0], hits[1] + assert isinstance(fact.timestamp, datetime) + assert raw.superseded_by is not None + + async def test_another_sessions_summary_does_not_mark_this_chunk(self): + memory = FakeMemory(chunks=[_chunk("old turn", session="s2", ended=NOW - timedelta(days=2))]) + hit = (await UnifiedMemorySearch(memory, FakeKv()).search( + "old", current_session_id="s1", summary_at=NOW, + ))[0] + assert hit.before_summary is False + + +class TestFormatting: + async def test_every_line_names_its_source(self): + memory = FakeMemory( + facts=[_fact("server_port", "8099")], + chunks=[_chunk("verbatim body", start=4, end=4)], + ) + kv = FakeKv([{"session_id": "s3", "key": "findings", "value": "sketch body", "updated_at": NOW}]) + hits = await UnifiedMemorySearch(memory, kv).search("body port") + text = format_hits(hits, "body port", "all") + + assert 'memory_search "body port" (scope=all): 3 hit(s)' in text + assert "fact [technical]" in text + assert "raw · session s1 · turn 4" in text + assert "sketch · session s3 · findings" in text + assert "keys: port" in text + + async def test_an_outdated_raw_excerpt_says_so(self): + memory = FakeMemory( + facts=[_fact("server_port", "8099 is the server port", updated=NOW)], + chunks=[_chunk("the server port was 8098", ended=NOW - timedelta(days=3))], + ) + hits = await UnifiedMemorySearch(memory, FakeKv()).search("server port") + text = format_hits(hits, "server port", "all") + assert "outdated: a newer curated fact overrides this" in text + + def test_no_hits_says_so(self): + text = format_hits([], "nothing", "all") + assert "0 hit(s)" in text + assert "No matches" in text diff --git a/tests/unit/tools/test_memory_search_tool.py b/tests/unit/tools/test_memory_search_tool.py new file mode 100644 index 0000000..bc09414 --- /dev/null +++ b/tests/unit/tools/test_memory_search_tool.py @@ -0,0 +1,73 @@ +"""The `memory_search` tool — argument handling and output shape.""" + +from datetime import datetime, timezone + +from navi.memory.search import MemoryHit +from navi.tools._internal.base import ToolContext +from navi.tools.memory_search import MemorySearchTool + +NOW = datetime(2026, 11, 9, 12, 0, tzinfo=timezone.utc) + + +class FakeSearch: + def __init__(self, hits=None): + self.hits = hits or [] + self.calls = [] + + async def search(self, query, **kwargs): + self.calls.append((query, kwargs)) + return self.hits + + async def latest_summary_at(self, session_id): + return NOW if session_id == "s1" else None + + +class TestArguments: + async def test_a_missing_query_is_an_error(self): + result = await MemorySearchTool(FakeSearch()).execute({}) + assert result.success is False + assert result.error == "missing query" + + async def test_defaults_are_scope_all_and_k_five(self): + search = FakeSearch() + await MemorySearchTool(search).execute({"query": "port"}) + query, kwargs = search.calls[0] + assert query == "port" + assert kwargs["scope"] == "all" + assert kwargs["k"] == 5 + + async def test_an_invalid_scope_falls_back_to_all(self): + search = FakeSearch() + await MemorySearchTool(search).execute({"query": "port", "scope": "everything"}) + assert search.calls[0][1]["scope"] == "all" + + async def test_k_is_clamped(self): + search = FakeSearch() + tool = MemorySearchTool(search) + await tool.execute({"query": "x", "k": 500}) + await tool.execute({"query": "x", "k": "nope"}) + assert search.calls[0][1]["k"] == 20 + assert search.calls[1][1]["k"] == 5 + + async def test_user_and_session_come_from_the_tool_context(self): + search = FakeSearch() + ctx = ToolContext(session_id="s1", user_id="u1") + await MemorySearchTool(search).execute({"query": "x"}, ctx) + kwargs = search.calls[0][1] + assert kwargs["user_id"] == "u1" + assert kwargs["current_session_id"] == "s1" + assert kwargs["summary_at"] == NOW + + +class TestOutput: + async def test_hits_are_rendered_with_provenance(self): + hits = [ + MemoryHit(source_type="fact", text="server_port: 8099", key="server_port", + category="technical", timestamp=NOW), + MemoryHit(source_type="raw", text="we set the port to 8099", + session_id="s1", start_seq=3, end_seq=4, timestamp=NOW), + ] + result = await MemorySearchTool(FakeSearch(hits)).execute({"query": "port"}) + assert result.success is True + assert "fact [technical]" in result.output + assert "raw · session s1 · turns 3–4" in result.output