diff --git a/docs/PLAN-v0.31-10tb-scale-operating-layer.md b/docs/PLAN-v0.31-10tb-scale-operating-layer.md new file mode 100644 index 0000000..8108d4f --- /dev/null +++ b/docs/PLAN-v0.31-10tb-scale-operating-layer.md @@ -0,0 +1,405 @@ +# Synaptic Memory v0.31 - 10TB Scale Operating Layer Plan + +## Summary + +This plan turns the current large-corpus retrieval work into a 10TB-class +document memory architecture. The current `main` already has the important +search intelligence: + +- deterministic default search path +- `agent_search` / `deep_search` with bounded deterministic rewrites +- EvidenceSearch with FTS, optional vector seeds, graph expansion, rerank/MMR +- memory event ledger, retrieval feedback, scope-aware reinforcement, health + reporting, and pollution signals +- CompositeBackend shape for Kuzu + Qdrant + MinIO + +The part that is not yet 10TB-ready is the storage/index/ingestion operating +layer. A single SQLite/Kuzu-style backend can prove retrieval logic, but 10TB +requires specialized index services, incremental ingestion workers, ACL-aware +candidate routing, and health/lag observability. + +## Current Evidence + +Current public-scale assets: + +- `tests/benchmark/data/msmarco_full.db`: 11GB SQLite DB +- `tests/benchmark/data/msmarco_passage_full.corpus.jsonl`: 8,841,823 passages +- DeepSeek 50-query baseline before recent rewrite work: `23/50` +- Post rewrite 50-query follow-up: `25/50` +- Current `main` scale-plan baseline: `31/50` + (`examples/ablation/diagnostics/agent_loop_20260703_000851.md`) +- Targeted recovered cases: + - `237373` soil/rocks process wording + - `54544` blood-borne/sexually transmitted wording + - `319564` carrot fiber serving-size wording + - `155234` tire size/gas mileage wording + - `208145` bicycle tube sizing wording + +Current code strengths: + +- `run_agent_loop()` supports multi-turn tool exploration, forced first tool, + sufficiency gating, history compaction, and trace logging. +- `deep_search_tool()` runs bounded deterministic rewrites internally and merges + rewrite evidence before returning to the LLM. +- `EvidenceSearch.search()` provides the core candidate flow: FTS seed, optional + vector seed, graph expansion, rerank/MMR aggregation, diagnostics. +- `SynapticGraph.search(record=True)` can persist retrieval events and connect + results to feedback and scope-aware score updates. +- `memory_health()` already exposes feedback outcomes, memory event counts, + score scopes, suspect targets, signal kinds, and semantic extraction failures. + +Current scale blockers: + +- `KuzuBackend.search_fts()` currently fetches all nodes and scores in Python for + IR parity. That is useful for tests and small/medium graphs, but not viable at + 10TB. +- `CompositeBackend.search_vector()` fetches nodes from graph storage one by one + after Qdrant returns ids. At 10TB this needs batch fetch, payload filters, and + shard-aware routing. +- The `StorageBackend` protocol has no native query object for ACL, tenant, + source, time range, document type, index generation, cursor, or score metadata. +- Ingestion is library-local. 10TB needs durable queues, idempotent jobs, + retry/dead-letter handling, parser/OCR/embedding/index stages, and index lag + reporting. +- Agent tools currently accept broad query strings. At 10TB every tool call must + carry scope/ACL/domain filters and budgeted routing constraints. + +## Target Architecture + +```mermaid +flowchart LR + U["User / Agent"] --> A["SynapticGraph / Agent Loop"] + A --> R["IndexRouter"] + R --> L["Lexical Index\nOpenSearch/Elastic"] + R --> V["Vector Index\nQdrant/Milvus"] + R --> G["Graph Store\nKuzu/Postgres/Graph service"] + R --> M["Memory Ledger\nPostgres"] + G --> B["Blob Store\nS3/MinIO/NAS"] + I["Ingestion Workers"] --> B + I --> L + I --> V + I --> G + I --> M + M --> H["Memory Health / Feedback Aggregates"] + H --> R +``` + +The core rule is: no single store owns everything. + +- Blob/object store owns original files and large extracted text. +- Lexical index owns BM25/fielded filtering. +- Vector index owns ANN retrieval. +- Graph store owns node/edge topology and provenance links. +- Ledger store owns events, feedback, scope scores, health summaries, and lag. +- `IndexRouter` owns candidate fan-out, merge, score normalization, and + fallback rules. + +## Data Model Layers + +### Document Object + +The original document is immutable content addressed by: + +- `document_id` +- `source_uri` +- `content_hash` +- `version` +- `workspace_id` +- `acl_policy_id` +- `mime_type` +- `created_at` +- `updated_at` + +The blob store keeps raw files and large extracted text. Graph/search stores +only compact content, pointers, and metadata required for retrieval. + +### Chunk / Section + +Searchable unit: + +- `chunk_id` +- `document_id` +- `section_path` +- `page` +- `offset_start` +- `offset_end` +- `text_preview` +- `content_ref` +- `embedding_ref` +- `language` +- `domain` +- `source` +- `acl_policy_id` + +Chunk ids must be deterministic across re-ingest when the same text and section +path survive. This is required for feedback and memory scores to remain useful +after incremental updates. + +### Memory Metadata + +Current `MemoryEvent`, `RetrievalEvent`, `MemoryScope`, and `MemoryScore` stay +as the operating ledger. At scale they move to a durable partitioned store. + +Partition keys: + +- `workspace_id` +- `domain` +- time bucket +- event kind +- scope key + +## Retrieval Flow + +1. Normalize query and scope. +2. Resolve ACL/domain/source/time filters. +3. Generate bounded deterministic rewrites. +4. Route each query variant through `IndexRouter`. +5. Fetch lexical candidates with scores and index metadata. +6. Fetch vector candidates with payload filters. +7. Merge by normalized score, query variant, source diversity, and recency. +8. Fetch graph neighborhoods only for bounded candidate ids. +9. Apply memory boosts/penalties within scope. +10. Rerank and MMR aggregate. +11. Return compact evidence with provenance pointers. +12. If `record=True`, persist retrieval event and diagnostics. + +Important invariant: + +- LLM prompts never receive raw provenance/event rows. +- Prompts receive selected evidence plus compact source/title/chunk/page + summaries. + +## New Runtime Contracts + +The current `StorageBackend` remains the compatibility layer. The 10TB path adds +optional protocols. + +### CandidateProvider + +Returns scored candidate ids without materializing full nodes. + +Fields required per candidate: + +- `node_id` +- `document_id` +- `score` +- `score_source` +- `rank` +- `query_variant` +- `metadata` +- `index_generation` + +### IndexRouter + +Coordinates multiple candidate providers: + +- lexical provider +- vector provider +- graph provider +- memory provider + +Responsibilities: + +- apply scope/ACL filters before provider search +- fan out query variants +- normalize provider scores +- dedupe candidates by node/document +- enforce per-provider and global budgets +- expose diagnostics and index lag + +### IngestionJobStore + +Tracks durable indexing work: + +- `job_id` +- `document_id` +- `version` +- `stage` +- `status` +- `attempt` +- `error` +- `created_at` +- `updated_at` + +Stages: + +- `discover` +- `parse` +- `ocr` +- `chunk` +- `embed` +- `lexical_index` +- `vector_index` +- `graph_index` +- `semantic_extract` +- `ledger_commit` + +### IndexHealthBackend + +Reports operational lag: + +- documents discovered vs indexed +- chunks parsed vs embedded +- lexical generation +- vector generation +- graph generation +- failed jobs by stage/source/model +- average and p95 indexing latency + +## 10TB Storage Choices + +Recommended first production-like PoC: + +- Blob: MinIO or S3 +- Lexical: OpenSearch or Elastic +- Vector: Qdrant cluster +- Ledger/metadata: Postgres +- Graph: + - Kuzu for embedded/single-workspace PoC + - Postgres edge tables or graph service for multi-tenant production + +Do not force Kuzu to be the global 10TB graph store before measuring write and +multi-reader limits. Keep Kuzu as an excellent local graph/PoC backend and add a +router-compatible graph provider for production. + +## Implementation Order + +### Phase 0 - Baseline Lock + +Goal: know whether later scale changes preserve current quality. + +- Re-run DeepSeek 50-query baseline on current `main`. + - Current result: `31/50`, mean first relevant turn `1.16`, empty calls `6`, + duplicate calls `0`. +- If stable, run 100-query follow-up next. +- Report reach, first relevant turn/call, tool calls, empty calls, latency, + prompt/completion tokens, and recovered/regressed qids. +- Keep raw JSONL as diagnostic artifact only when small and useful; summarize in + curated markdown. + +### Phase 1 - Contract Scaffolding + +Goal: make the scale boundary explicit without changing default behavior. + +- Add optional `CandidateProvider`, `IndexRouter`, `IngestionJobStore`, and + `IndexHealthBackend` protocols. +- Add small dataclasses for candidate request/result/diagnostics. +- Add unit tests that these types serialize and keep default code unaffected. +- Document how `EvidenceSearch` will consume router candidates in Phase 2. + +### Phase 2 - Router Prototype + +Goal: prove the new boundary on the existing SQLite full DB. + +- Implement an in-process router wrapping the current `StorageBackend`. +- Make `EvidenceSearch` optionally accept a router. +- Ensure default path remains identical when router is absent. +- Add diagnostics for provider counts, score ranges, dedupe counts, and lag. + +Current implementation status: + +- `StorageCandidateProvider` wraps legacy `StorageBackend.search_fts()` and + returns scored candidate ids. +- `InProcessIndexRouter` merges provider results, dedupes by best score, and + reports provider diagnostics. +- `EvidenceSearch(index_router=...)` can use routed candidate ids for the seed + stage and hydrate only those ids. +- `SynapticGraph(index_router=...)` wires the router into the public search + facade while keeping the default path unchanged when omitted. + +### Phase 3 - External Lexical Provider + +Goal: remove the 10TB-incompatible FTS bottleneck. + +- Add OpenSearch/Elastic provider behind `CandidateProvider`. +- Support filters: workspace, ACL, domain, source, date, doc type, language. +- Return scored ids only; fetch full nodes in batches from metadata/graph store. +- Add 100GB to 1TB PoC benchmark. + +Current implementation status: + +- `OpenSearchCandidateProvider` accepts an injected OpenSearch/Elasticsearch + client without adding a mandatory runtime dependency. +- It converts `IndexFilter` into provider-side bool filters for workspace, ACL, + document id, source, language, mime type, tags, time ranges, and property + filters. +- It returns deduped `ScoredCandidate` ids with normalized lexical scores and + compact metadata, leaving node hydration to `EvidenceSearch` / graph storage. + +### Phase 4 - Vector Provider Upgrade + +Goal: make vector retrieval payload-filtered and batch-friendly. + +- Extend Qdrant helper to return scored candidates and payload metadata. +- Add payload filters matching lexical filters. +- Add batch graph/metadata hydration for vector results. +- Add shard/collection strategy by workspace/domain/language. + +Current implementation status: + +- `QdrantCandidateProvider` accepts an injected Qdrant-compatible client + without adding a mandatory runtime dependency. +- It uses `CandidateSearchRequest.embedding` and converts `IndexFilter` into + Qdrant payload filters for workspace, ACL, document id, source, language, + mime type, tags, time ranges, and property filters. +- It returns deduped vector `ScoredCandidate` ids with compact payload metadata, + leaving node hydration to `EvidenceSearch` / graph storage. + +### Phase 5 - Ingestion Pipeline + +Goal: make new/changed data manageable as memory, not bulk reloads. + +- Add durable ingestion job schema. +- Add idempotent parse/chunk/embed/index stages. +- Record every document version change as `MemoryEvent`. +- Add delete/update tombstone flow. +- Add index lag and failed-stage health report. + +Current implementation status: + +- `IngestionJobStore` now has Memory/SQLite persistence for stage, status, + attempt, error, version, and job properties. +- `IndexHealthBackend` now has Memory/SQLite persistence for provider lag + reports, pending counts, failed jobs, p95 latency, and provider properties. +- This records operational state only; idempotent parse/chunk/embed/index + orchestration and tombstones remain the next pipeline step. + +### Phase 6 - Memory Health at Scale + +Goal: keep feedback and pollution management useful with increasing data. + +- Move high-volume event aggregation to summary tables. +- Partition memory/retrieval events by workspace/time. +- Compute repeated failure, stale, conflict, and source/model drift from changed + windows instead of full scans. +- Feed high-confidence penalties back into `IndexRouter`. + +### Phase 7 - Production Gates + +Goal: prove it can run on real business documents. + +- Build 500-query internal eval set. +- Include single-hop, multi-hop, temporal, ACL, synonym, table/form, and + policy-conflict queries. +- Measure recall@k, answer groundedness, first relevant turn/call, latency, + token cost, index lag, and failure recovery. + +## Acceptance Criteria + +The system is 10TB-ready only when all are true: + +- 1TB PoC shows no quality regression against current baseline. +- Retrieval p95 is within the agreed SLA for filtered queries. +- Incremental update/delete does not require full reindex. +- ACL filters are enforced before candidate materialization. +- Event/feedback/provenance remain metadata, not prompt bloat. +- Health report exposes index lag, failed stages, scope score distribution, and + suspect memory targets. +- Agent tools cannot bypass scope/ACL/domain filters. +- Rollback/replay restores index state from document version and event ledger. + +## Near-Term TODO + +1. Run DeepSeek 100-query follow-up against the `31/50` baseline. +2. Add idempotent ingestion stage orchestration and tombstone/replay flow. +3. Run a 100GB or 1TB corpus rehearsal before any 10TB ingestion work. diff --git a/examples/ablation/diagnostics/agent_loop_20260703_000851.md b/examples/ablation/diagnostics/agent_loop_20260703_000851.md new file mode 100644 index 0000000..15b3e95 --- /dev/null +++ b/examples/ablation/diagnostics/agent_loop_20260703_000851.md @@ -0,0 +1,91 @@ +# Agent Loop Retrieval Benchmark — Synaptic + +- Run at: 2026-07-03 00:08:51 KST +- Dataset path: tests/benchmark/data/msmarco_passage_full.json +- SQLite DB path: tests/benchmark/data/msmarco_full.db +- Subset: 50 +- Corpus limit: 8841823 +- LLM base URL: https://api.deepseek.com/v1 +- Model: deepseek-v4-flash +- Max turns: 5 +- Sufficiency gate: yes +- Force first tool: yes +- Incremental JSONL: examples/ablation/diagnostics/agent_loop_deepseek_v4_flash_50_post_scale_plan.jsonl +- SQLite FTS AND-first threshold: 20 +- SQLite FTS lexical rerank pool: 500 + +This measures LLM-planned exploration. The agent can change follow-up queries and tool choices based on evidence from earlier turns. The main metric is document reach, not ranked MRR, because the agent loop returns a cumulative evidence set. + +## Summary + +- Reach: 31/50 (0.620) +- Mean turns: 4.18 +- Mean tool calls: 6.18 +- Mean first relevant turn: 1.16 +- Mean first relevant tool calls: 1.32 +- Mean elapsed: 57.2s +- P50/P90 elapsed: 57.7s / 74.3s +- Mean prompt tokens: 21487 +- Mean completion tokens: 1122 +- Mean unique tools: 2.38 +- Mean unique search targets: 5.84 +- Mean query rewrites: 4.48 +- Queries with >1 tool type: 48/50 +- Queries with query rewrites: 46/50 +- Duplicate tool calls: 0 +- Empty tool calls: 6 + +## Per Query + +| QID | Reach | Turns | Calls | Tools | Targets | Rewrites | First Rel Turn | First Rel Calls | Found Relevant | Elapsed | Query | +|-----|:-----:|------:|------:|------:|--------:|---------:|---------------:|----------------:|----------------|--------:|-------| +| 300674 | yes | 3 | 3 | 2 | 2 | 2 | 1 | 1 | 7067032 | 37.3s | how many years did william bradford serve as governor of plymouth colony? | +| 125705 | no | 5 | 7 | 2 | 7 | 7 | - | - | - | 54.6s | define preventive | +| 94798 | yes | 4 | 6 | 2 | 5 | 4 | 1 | 1 | 7067181 | 56.6s | color overlay photoshop | +| 9083 | yes | 2 | 1 | 1 | 1 | 1 | 1 | 1 | 7067274 | 30.2s | ____________________ is considered the father of modern medicine. | +| 174249 | no | 5 | 4 | 3 | 4 | 4 | - | - | - | 48.1s | does xpress bet charge to deposit money in your account | +| 320792 | yes | 5 | 7 | 3 | 7 | 7 | 2 | 3 | 7067677 | 70.7s | how much is a cost to run disneyland | +| 1090270 | yes | 3 | 3 | 2 | 3 | 0 | 1 | 1 | 7067796 | 39.1s | botulinum definition | +| 1101279 | no | 5 | 12 | 3 | 12 | 12 | - | - | - | 78.4s | do physicians pay for insurance from their salaries? | +| 201376 | yes | 4 | 6 | 3 | 6 | 1 | 1 | 1 | 7068066 | 56.4s | here there be dragons comic | +| 54544 | yes | 5 | 8 | 3 | 8 | 2 | 1 | 1 | 7068203 | 63.2s | blood diseases that are sexually transmitted | +| 118457 | no | 3 | 2 | 2 | 2 | 1 | - | - | - | 37.0s | define bona fides | +| 178627 | no | 5 | 10 | 3 | 10 | 10 | - | - | - | 77.4s | effects of detox juice cleanse | +| 1101278 | yes | 4 | 6 | 2 | 6 | 1 | 1 | 1 | 7068907 | 55.6s | do prince harry and william have last names | +| 68095 | yes | 4 | 6 | 2 | 6 | 5 | 1 | 1 | 7069266 | 54.8s | can hives be a sign of pregnancy | +| 87892 | yes | 5 | 9 | 3 | 8 | 7 | 1 | 1 | 7069601 | 65.1s | causes of petechial hemorrhage | +| 257309 | no | 4 | 5 | 2 | 5 | 5 | - | - | - | 59.2s | how long does it take to get your bsrn if you already have a bachelors degree | +| 1090242 | yes | 5 | 12 | 3 | 12 | 11 | 1 | 1 | 7070556 | 76.1s | symptoms of ptsd in vietnam veterans | +| 211691 | no | 5 | 5 | 3 | 5 | 3 | - | - | - | 53.1s | how coffee works quote | +| 165002 | yes | 3 | 3 | 2 | 3 | 3 | 1 | 1 | 7070877 | 49.4s | does contraction of the ciliary muscles shorten the lens | +| 1101276 | yes | 2 | 1 | 1 | 1 | 0 | 1 | 1 | 7070950 | 31.9s | do spiders eat other animals | +| 264827 | yes | 3 | 2 | 2 | 2 | 2 | 1 | 1 | 7071066 | 35.6s | how long is the flight from chicago to cairo | +| 342285 | no | 5 | 11 | 2 | 11 | 1 | - | - | - | 65.4s | how titanic facts | +| 372586 | no | 4 | 9 | 2 | 8 | 7 | - | - | - | 64.6s | how to play blu ray discs | +| 89786 | yes | 3 | 3 | 3 | 3 | 2 | 1 | 1 | 7071501 | 44.8s | central city definition | +| 118448 | no | 3 | 3 | 2 | 2 | 1 | - | - | - | 43.7s | define body muscular endurance | +| 92542 | yes | 3 | 4 | 2 | 4 | 3 | 1 | 1 | 7072003 | 49.0s | circulation money definition | +| 206117 | yes | 5 | 11 | 3 | 8 | 8 | 1 | 1 | 7072155, 7072160 | 64.0s | hotels in thornton co | +| 141472 | yes | 4 | 4 | 2 | 4 | 0 | 1 | 1 | 7072290 | 48.3s | derriere definition | +| 293992 | no | 5 | 6 | 3 | 5 | 5 | - | - | - | 57.5s | how many product lines does coca cola have | +| 196232 | no | 5 | 8 | 3 | 8 | 8 | - | - | - | 75.5s | government does do | +| 352818 | no | 5 | 7 | 3 | 6 | 5 | - | - | - | 60.1s | how to cook string beans | +| 45924 | no | 5 | 8 | 3 | 7 | 6 | - | - | - | 71.5s | average temperatures las vegas by month | +| 208145 | yes | 5 | 9 | 3 | 9 | 9 | 1 | 1 | 7072843 | 82.0s | how bicycle tire tubes are sized | +| 79891 | no | 4 | 5 | 2 | 5 | 5 | - | - | - | 57.8s | can you substitute chocolate chips for semi-sweet | +| 208494 | yes | 4 | 5 | 3 | 5 | 5 | 1 | 1 | 7073272 | 57.1s | how big do newfypoo's get | +| 319564 | yes | 4 | 6 | 2 | 6 | 6 | 1 | 1 | 7073381 | 63.8s | how much fiber is in carrots | +| 155234 | yes | 3 | 4 | 2 | 4 | 3 | 1 | 1 | 502713 | 57.1s | do bigger tires affect gas mileage | +| 14151 | yes | 5 | 8 | 2 | 8 | 7 | 1 | 1 | 7073772 | 69.5s | age requirements for name change | +| 67802 | yes | 3 | 6 | 2 | 6 | 5 | 1 | 1 | 7074071 | 54.4s | can green tea cause stomach problems | +| 1090184 | yes | 5 | 4 | 2 | 4 | 3 | 1 | 1 | 7074235 | 42.9s | synonym of subordinate | +| 323382 | no | 5 | 10 | 2 | 10 | 10 | - | - | - | 67.5s | how much is the stamp to send a card | +| 323998 | yes | 3 | 3 | 2 | 2 | 2 | 1 | 1 | 7074377 | 41.4s | how much magnesium in kidney beans | +| 91711 | yes | 5 | 7 | 2 | 4 | 3 | 1 | 1 | 1956185 | 63.9s | child psychiatrist salary 2016 | +| 125898 | yes | 5 | 8 | 3 | 6 | 6 | 3 | 6 | 7074710 | 62.3s | define prosthetic device | +| 289812 | no | 3 | 3 | 2 | 3 | 1 | - | - | - | 34.1s | how many mm is a nickel coin | +| 333486 | yes | 5 | 6 | 3 | 6 | 6 | 3 | 4 | 7075218 | 63.9s | how old do you have to be to get a job in idaho | +| 1090171 | yes | 5 | 8 | 2 | 8 | 0 | 1 | 1 | 7075317 | 58.6s | synonyms for the word discipline | +| 73257 | no | 5 | 8 | 3 | 8 | 3 | - | - | - | 70.6s | can seizure meds cause low sodium? | +| 1090170 | no | 5 | 11 | 3 | 11 | 11 | - | - | - | 74.3s | synonyms for the word, decline | +| 237373 | yes | 4 | 6 | 2 | 6 | 5 | 1 | 1 | 7075449 | 64.2s | how is soil created from rocks | diff --git a/examples/ablation/diagnostics/agent_loop_deepseek_auto_rewrites_20260702.md b/examples/ablation/diagnostics/agent_loop_deepseek_auto_rewrites_20260702.md index 744e775..b15dca0 100644 --- a/examples/ablation/diagnostics/agent_loop_deepseek_auto_rewrites_20260702.md +++ b/examples/ablation/diagnostics/agent_loop_deepseek_auto_rewrites_20260702.md @@ -68,6 +68,43 @@ Current deterministic `deep_search` also showed that several misses were caused | 155234 | do bigger tires affect gas mileage | tire size factors influence gas mileage; tire width versus gas mileage | 1 | reach=yes, first relevant turn 1 / call 1 | | 208145 | how bicycle tire tubes are sized | bicycle tire tube size sidewall ETRTO metric imperial; bicycle tire sidewall tube size printed raised numbers | 1 | reach=yes, first relevant turn 1 / call 1 | +## Current Main 50-Query Baseline + +After merging the process cleanup, blood-borne transmission rewrite, and answer-shaped rewrite hints, the current `main` DeepSeek 50-query run reached `31/50`, up from the original `23/50` and the earlier post-rewrite `25/50`. + +| Metric | Original 50-query run | Current main run | +| --- | ---: | ---: | +| Reach | 23/50 | 31/50 | +| Mean turns | 4.14 | 4.18 | +| Mean tool calls | 5.78 | 6.18 | +| Mean first relevant turn | 1.70 | 1.16 | +| Mean first relevant tool calls | 2.22 | 1.32 | +| Empty tool calls | 11 | 6 | +| Duplicate tool calls | 0 | 0 | + +Net new hits versus the original 50-query run: + +- `1101278` - do prince harry and william have last names +- `125898` - define prosthetic device +- `14151` - age requirements for name change +- `155234` - do bigger tires affect gas mileage +- `208145` - how bicycle tire tubes are sized +- `237373` - how is soil created from rocks +- `319564` - how much fiber is in carrots +- `320792` - how much is a cost to run disneyland +- `54544` - blood diseases that are sexually transmitted +- `91711` - child psychiatrist salary 2016 + +Regressed versus the original 50-query run: + +- `178627` - effects of detox juice cleanse +- `45924` - average temperatures las vegas by month + +Artifacts: + +- `examples/ablation/diagnostics/agent_loop_20260703_000851.md` +- `examples/ablation/diagnostics/agent_loop_deepseek_v4_flash_50_post_scale_plan.jsonl` + ## Interpretation Prompt-visible hints alone were not enough: DeepSeek often generated nearby but non-gold rewrites. Running the deterministic rewrite hints inside `deep_search` removes that planning variance for cheap, bounded patterns such as dropping noisy numeric years and rewriting "created from" process questions into answer-shaped phrases. diff --git a/examples/ablation/diagnostics/agent_loop_deepseek_v4_flash_50_post_scale_plan.jsonl b/examples/ablation/diagnostics/agent_loop_deepseek_v4_flash_50_post_scale_plan.jsonl new file mode 100644 index 0000000..492ae9c --- /dev/null +++ b/examples/ablation/diagnostics/agent_loop_deepseek_v4_flash_50_post_scale_plan.jsonl @@ -0,0 +1,50 @@ +{"completion_tokens": 357, "duplicate_tool_calls": 0, "elapsed_sec": 37.300143560394645, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7067032"], "prompt_tokens": 12030, "qid": "300674", "query": "how many years did william bradford serve as governor of plymouth colony?", "query_rewrites": 2, "reached": true, "relevant_docs": ["7067032"], "search_targets": ["william bradford governor plymouth colony years served", "william bradford governor 30 years"], "tool_calls": 3, "tool_sequence": ["deep_search", "get_document", "get_document"], "turns": 3, "unique_search_targets": 2, "unique_tools": 2} +{"completion_tokens": 1068, "duplicate_tool_calls": 0, "elapsed_sec": 54.59644513204694, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 21746, "qid": "125705", "query": "define preventive", "query_rewrites": 7, "reached": false, "relevant_docs": ["7067056"], "search_targets": ["preventive definition", "preventive care law & legal definition", "preventive maintenance definition", "preventive services definition", "definition of prophylactic preventive", "preventive care healthcare definition", "public health definition preventive"], "tool_calls": 7, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 7, "unique_tools": 2} +{"completion_tokens": 1300, "duplicate_tool_calls": 0, "elapsed_sec": 56.6194977411069, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7067181"], "prompt_tokens": 18522, "qid": "94798", "query": "color overlay photoshop", "query_rewrites": 4, "reached": true, "relevant_docs": ["7067181"], "search_targets": ["color overlay photoshop", "color overlay photoshop layer styles", "photoshop color overlay", "how to change a dress color using photoshop", "how to change the color of clothing in photoshop"], "tool_calls": 6, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 5, "unique_tools": 2} +{"completion_tokens": 349, "duplicate_tool_calls": 0, "elapsed_sec": 30.16610910696909, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7067274"], "prompt_tokens": 7166, "qid": "9083", "query": "____________________ is considered the father of modern medicine.", "query_rewrites": 1, "reached": true, "relevant_docs": ["7067274"], "search_targets": ["father of modern medicine"], "tool_calls": 1, "tool_sequence": ["deep_search"], "turns": 2, "unique_search_targets": 1, "unique_tools": 1} +{"completion_tokens": 806, "duplicate_tool_calls": 0, "elapsed_sec": 48.05103103630245, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 20, "found_relevant_docs": [], "prompt_tokens": 21649, "qid": "174249", "query": "does xpress bet charge to deposit money in your account", "query_rewrites": 4, "reached": false, "relevant_docs": ["7067348"], "search_targets": ["xpressbet deposit fee charge", "xpressbet deposit convenience fee", "xpressbet deposit fee online no fee", "xpressbet fees deposit wagering"], "tool_calls": 4, "tool_sequence": ["deep_search", "get_document", "search", "get_document"], "turns": 5, "unique_search_targets": 4, "unique_tools": 3} +{"completion_tokens": 1390, "duplicate_tool_calls": 0, "elapsed_sec": 70.66074568685144, "empty_tool_calls": 0, "first_relevant_tool_calls": 3, "first_relevant_turn": 2, "found_ids_count": 50, "found_relevant_docs": ["7067677"], "prompt_tokens": 34649, "qid": "320792", "query": "how much is a cost to run disneyland", "query_rewrites": 7, "reached": true, "relevant_docs": ["7067677"], "search_targets": ["disneyland operating cost", "disneyland annual operating cost dollars per year", "disneyland daily operating expenses cost to run", "disney theme parks operating cost 571 million", "disneyland park operating cost per year", "disneyland resort operating expenses annual", "7 billion park operating expenses 32 million per day disney"], "tool_calls": 7, "tool_sequence": ["deep_search", "search", "search", "get_document", "search", "search", "get_document"], "turns": 5, "unique_search_targets": 7, "unique_tools": 3} +{"completion_tokens": 509, "duplicate_tool_calls": 0, "elapsed_sec": 39.119288365822285, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7067796"], "prompt_tokens": 11861, "qid": "1090270", "query": "botulinum definition", "query_rewrites": 0, "reached": true, "relevant_docs": ["7067796"], "search_targets": ["botulinum definition", "7067796", "3022771"], "tool_calls": 3, "tool_sequence": ["deep_search", "get_document", "get_document"], "turns": 3, "unique_search_targets": 3, "unique_tools": 2} +{"completion_tokens": 1618, "duplicate_tool_calls": 0, "elapsed_sec": 78.42512316722423, "empty_tool_calls": 1, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 40, "found_relevant_docs": [], "prompt_tokens": 36538, "qid": "1101279", "query": "do physicians pay for insurance from their salaries?", "query_rewrites": 12, "reached": false, "relevant_docs": ["7067891"], "search_targets": ["do physicians pay for insurance from their salaries", "physicians malpractice insurance paid from salary", "doctors pay for own malpractice insurance out of pocket", "physicians pay for insurance from salary practice expenses malpractice", "who pays for physician malpractice insurance salary employed", "ob/gyn malpractice insurance premiums physicians pay", "doctors internal medicine pay malpractice insurance annually", "salary benefits invasive cardiologist paid malpractice insurance", "employed physician malpractice insurance paid by employer or deducted from salary", "physician malpractice insurance paid from salary", "malpractice insurance physician salary", "physician pay insurance salary deduction"], "tool_calls": 12, "tool_sequence": ["deep_search", "search", "search", "get_document", "get_document", "get_document", "get_document", "get_document", "search", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 12, "unique_tools": 3} +{"completion_tokens": 1306, "duplicate_tool_calls": 0, "elapsed_sec": 56.3994110301137, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 30, "found_relevant_docs": ["7068066"], "prompt_tokens": 22263, "qid": "201376", "query": "here there be dragons comic", "query_rewrites": 1, "reached": true, "relevant_docs": ["7068066"], "search_targets": ["here there be dragons comic", "\"here, there be dragons\" comic james owen", "7068066", "7068070", "7068073", "7068069"], "tool_calls": 6, "tool_sequence": ["deep_search", "search", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 6, "unique_tools": 3} +{"completion_tokens": 1179, "duplicate_tool_calls": 0, "elapsed_sec": 63.153891704045236, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 20, "found_relevant_docs": ["7068203"], "prompt_tokens": 24215, "qid": "54544", "query": "blood diseases that are sexually transmitted", "query_rewrites": 2, "reached": true, "relevant_docs": ["7068203"], "search_targets": ["blood diseases sexually transmitted", "7068203", "640830", "3734306", "hiv hepatitis b hepatitis c sexually transmitted blood borne", "7068202", "4515730", "1268039"], "tool_calls": 8, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "search", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 8, "unique_tools": 3} +{"completion_tokens": 717, "duplicate_tool_calls": 0, "elapsed_sec": 37.00856179418042, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 11784, "qid": "118457", "query": "define bona fides", "query_rewrites": 1, "reached": false, "relevant_docs": ["7068493"], "search_targets": ["define bona fides", "definition of bona fides"], "tool_calls": 2, "tool_sequence": ["deep_search", "get_document"], "turns": 3, "unique_search_targets": 2, "unique_tools": 2} +{"completion_tokens": 1905, "duplicate_tool_calls": 0, "elapsed_sec": 77.41898558102548, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 32, "found_relevant_docs": [], "prompt_tokens": 31143, "qid": "178627", "query": "effects of detox juice cleanse", "query_rewrites": 10, "reached": false, "relevant_docs": ["7068519", "7068520"], "search_targets": ["effects of detox juice cleanse health benefits risks", "juice cleanse benefits health effects three days", "cold-pressed juice health benefits green juice detox cleanse", "liver cleanse side effects rash breakout energy purging toxins", "non-juice detox benefits juicing without blood sugar spikes", "detox juice cleanse side effects risks benefits", "master cleanse detox juice side effects tiredness", "liver cleanse side effects energy rash pimples purging toxins", "detox juice cleanse side effects", "\"juice cleanse\" health benefits weight loss energy cravings"], "tool_calls": 10, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "search", "get_document", "get_document", "get_document", "search"], "turns": 5, "unique_search_targets": 10, "unique_tools": 3} +{"completion_tokens": 1158, "duplicate_tool_calls": 0, "elapsed_sec": 55.592501354869455, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7068907"], "prompt_tokens": 18587, "qid": "1101278", "query": "do prince harry and william have last names", "query_rewrites": 1, "reached": true, "relevant_docs": ["7068907"], "search_targets": ["prince harry and prince william last name surname", "2919121", "7068913", "8129467", "7068907", "2919125"], "tool_calls": 6, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 6, "unique_tools": 2} +{"completion_tokens": 1567, "duplicate_tool_calls": 0, "elapsed_sec": 54.79991369321942, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7069266"], "prompt_tokens": 19076, "qid": "68095", "query": "can hives be a sign of pregnancy", "query_rewrites": 5, "reached": true, "relevant_docs": ["7069266"], "search_targets": ["can hives be a sign of pregnancy", "hives pregnancy sign", "hives pregnancy early sign symptom", "hives during pregnancy", "pregnancy hives", "presumptive sign of pregnancy hives"], "tool_calls": 6, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 6, "unique_tools": 2} +{"completion_tokens": 1116, "duplicate_tool_calls": 0, "elapsed_sec": 65.11347865778953, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 33, "found_relevant_docs": ["7069601"], "prompt_tokens": 31925, "qid": "87892", "query": "causes of petechial hemorrhage", "query_rewrites": 7, "reached": true, "relevant_docs": ["7069601"], "search_targets": ["causes of petechial hemorrhage", "petechial rash causes", "16 causes of petechial rashes in adults", "petechial hemorrhage causes injury trauma medical conditions", "petechial rash causes trauma injury conditions", "petechial rash causes trauma coughing childbirth vomiting", "causes of petechial hemorrhages injury to the brain trauma", "petechial hemorrhage causes medical conditions infections medications"], "tool_calls": 9, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "search", "get_document", "get_document", "get_document", "search"], "turns": 5, "unique_search_targets": 8, "unique_tools": 3} +{"completion_tokens": 1110, "duplicate_tool_calls": 0, "elapsed_sec": 59.235608492046595, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 17996, "qid": "257309", "query": "how long does it take to get your bsrn if you already have a bachelors degree", "query_rewrites": 5, "reached": false, "relevant_docs": ["4959637"], "search_targets": ["how long does it take to get bsrn if you already have a bachelor's degree accelerated bsn", "accelerated bsn 12-18 months already have bachelor's degree", "accelerated bsn already have non-nursing bachelor's degree how long", "accelerated bsn 4-year degree students already earned bachelor's degree", "accelerated bsn shorten time frame already hold bachelor's degree"], "tool_calls": 5, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 5, "unique_tools": 2} +{"completion_tokens": 1697, "duplicate_tool_calls": 0, "elapsed_sec": 76.07700867531821, "empty_tool_calls": 1, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 45, "found_relevant_docs": ["7070556"], "prompt_tokens": 32194, "qid": "1090242", "query": "symptoms of ptsd in vietnam veterans", "query_rewrites": 11, "reached": true, "relevant_docs": ["7070556"], "search_targets": ["symptoms of ptsd in vietnam veterans", "ptsd symptoms vietnam veterans", "ptsd symptoms vietnam veterans insomnia anxiety irritability", "ptsd symptoms vietnam veterans decades after war", "clinically serious stress reaction symptoms vietnam veterans", "vietnam syndrome physical and psychological symptoms veterans", "ptsd symptoms vietnam veterans specific symptoms re-experiencing avoidance hyperarousal", "vietnam veterans ptsd symptoms flashbacks nightmares depression", "ptsd symptom types re-experiencing avoidance emotional numbing hyperarousal vietnam veterans", "ptsd four types of symptoms reliving trauma avoiding negative thoughts hyperarousal", "ptsd flashbacks avoidance hyperarousal veterans", "vietnam syndrome physical psychological symptoms ptsd veterans"], "tool_calls": 12, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "search", "search", "get_document", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 12, "unique_tools": 3} +{"completion_tokens": 946, "duplicate_tool_calls": 0, "elapsed_sec": 53.11477872077376, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 40, "found_relevant_docs": [], "prompt_tokens": 28222, "qid": "211691", "query": "how coffee works quote", "query_rewrites": 3, "reached": false, "relevant_docs": ["7070643"], "search_targets": ["how coffee works quote", "\"how coffee works\" quote", "528303", "\"how coffee works\" book quote", "\"how coffee works\" howstuffworks quote"], "tool_calls": 5, "tool_sequence": ["deep_search", "search", "get_document", "search", "search"], "turns": 5, "unique_search_targets": 5, "unique_tools": 3} +{"completion_tokens": 1401, "duplicate_tool_calls": 0, "elapsed_sec": 49.41984023991972, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7070877"], "prompt_tokens": 12812, "qid": "165002", "query": "does contraction of the ciliary muscles shorten the lens", "query_rewrites": 3, "reached": true, "relevant_docs": ["7070877"], "search_targets": ["contraction of ciliary muscles shortens the lens accommodation", "ciliary muscle contraction lens shape round up shorten", "ciliary muscle contraction lens shape accommodation"], "tool_calls": 3, "tool_sequence": ["deep_search", "get_document", "get_document"], "turns": 3, "unique_search_targets": 3, "unique_tools": 2} +{"completion_tokens": 598, "duplicate_tool_calls": 0, "elapsed_sec": 31.90777140390128, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7070950"], "prompt_tokens": 7158, "qid": "1101276", "query": "do spiders eat other animals", "query_rewrites": 0, "reached": true, "relevant_docs": ["7070950"], "search_targets": ["do spiders eat other animals"], "tool_calls": 1, "tool_sequence": ["deep_search"], "turns": 2, "unique_search_targets": 1, "unique_tools": 1} +{"completion_tokens": 547, "duplicate_tool_calls": 0, "elapsed_sec": 35.5769102787599, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7071066"], "prompt_tokens": 11567, "qid": "264827", "query": "how long is the flight from chicago to cairo", "query_rewrites": 2, "reached": true, "relevant_docs": ["7071066"], "search_targets": ["flight duration chicago to cairo", "chicago to cairo flight duration"], "tool_calls": 2, "tool_sequence": ["deep_search", "get_document"], "turns": 3, "unique_search_targets": 2, "unique_tools": 2} +{"completion_tokens": 994, "duplicate_tool_calls": 0, "elapsed_sec": 65.41964694811031, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 15, "found_relevant_docs": [], "prompt_tokens": 26062, "qid": "342285", "query": "how titanic facts", "query_rewrites": 1, "reached": false, "relevant_docs": ["7071436"], "search_targets": ["titanic facts", "4128811", "7909030", "6951491", "320210", "4591908", "1392251", "1612884", "3648418", "290705", "7060043"], "tool_calls": 11, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 11, "unique_tools": 2} +{"completion_tokens": 1561, "duplicate_tool_calls": 0, "elapsed_sec": 64.6158408029005, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 20699, "qid": "372586", "query": "how to play blu ray discs", "query_rewrites": 7, "reached": false, "relevant_docs": ["7071494"], "search_targets": ["how to play blu ray discs", "play blu ray disc requirements", "blu ray drive play discs", "play blu ray disc laptop", "blu ray player play discs", "blu ray player play 3d discs", "ultra hd blu ray player", "all blu ray players can play commercial blu ray discs"], "tool_calls": 9, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 8, "unique_tools": 2} +{"completion_tokens": 551, "duplicate_tool_calls": 0, "elapsed_sec": 44.75339879328385, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 20, "found_relevant_docs": ["7071501"], "prompt_tokens": 12449, "qid": "89786", "query": "central city definition", "query_rewrites": 2, "reached": true, "relevant_docs": ["7071501"], "search_targets": ["central city definition", "definition of central city", "\"central city\" definition urban"], "tool_calls": 3, "tool_sequence": ["deep_search", "get_document", "search"], "turns": 3, "unique_search_targets": 3, "unique_tools": 3} +{"completion_tokens": 974, "duplicate_tool_calls": 0, "elapsed_sec": 43.661400279961526, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 12358, "qid": "118448", "query": "define body muscular endurance", "query_rewrites": 1, "reached": false, "relevant_docs": ["7071642"], "search_targets": ["define body muscular endurance", "define muscular endurance"], "tool_calls": 3, "tool_sequence": ["deep_search", "get_document", "get_document"], "turns": 3, "unique_search_targets": 2, "unique_tools": 2} +{"completion_tokens": 1252, "duplicate_tool_calls": 0, "elapsed_sec": 48.97357752919197, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7072003"], "prompt_tokens": 13040, "qid": "92542", "query": "circulation money definition", "query_rewrites": 3, "reached": true, "relevant_docs": ["7072003"], "search_targets": ["circulation money definition", "circulation of money definition", "money in circulation definition", "currency in circulation definition"], "tool_calls": 4, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document"], "turns": 3, "unique_search_targets": 4, "unique_tools": 2} +{"completion_tokens": 1346, "duplicate_tool_calls": 0, "elapsed_sec": 64.04464200418442, "empty_tool_calls": 2, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 50, "found_relevant_docs": ["7072155", "7072160"], "prompt_tokens": 36510, "qid": "206117", "query": "hotels in thornton co", "query_rewrites": 8, "reached": true, "relevant_docs": ["7072155", "7072156", "7072160"], "search_targets": ["hotels in thornton colorado", "thornton hotels", "cheap hotels in thornton", "best hotels thornton colorado list hampton inn candlewood suites econo lodge", "thornton colorado hotels list", "thornton co hotels motels lodging accommodation", "thornton hotels candlewood suites denver north thornton", "popular hotels in thornton"], "tool_calls": 11, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "search", "deep_search", "get_document", "get_document"], "turns": 5, "unique_search_targets": 8, "unique_tools": 3} +{"completion_tokens": 962, "duplicate_tool_calls": 0, "elapsed_sec": 48.30916942702606, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7072290"], "prompt_tokens": 16929, "qid": "141472", "query": "derriere definition", "query_rewrites": 0, "reached": true, "relevant_docs": ["7072290"], "search_targets": ["derriere definition", "7072290", "6292406", "7402452"], "tool_calls": 4, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 4, "unique_tools": 2} +{"completion_tokens": 954, "duplicate_tool_calls": 0, "elapsed_sec": 57.515873465221375, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 50, "found_relevant_docs": [], "prompt_tokens": 29058, "qid": "293992", "query": "how many product lines does coca cola have", "query_rewrites": 5, "reached": false, "relevant_docs": ["2790193"], "search_targets": ["coca-cola product lines number of product lines", "coca-cola product lines count how many", "coca-cola product lines", "\"product lines\" coca-cola width mix breadth", "coca-cola product mix width product lines"], "tool_calls": 6, "tool_sequence": ["deep_search", "search", "get_document", "get_document", "search", "get_document"], "turns": 5, "unique_search_targets": 5, "unique_tools": 3} +{"completion_tokens": 1737, "duplicate_tool_calls": 0, "elapsed_sec": 75.51519213989377, "empty_tool_calls": 1, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 40, "found_relevant_docs": [], "prompt_tokens": 31357, "qid": "196232", "query": "government does do", "query_rewrites": 8, "reached": false, "relevant_docs": ["7072326"], "search_targets": ["government", "purpose of government functions and roles", "what does the government do functions", "4 roles of the government in a mixed market economy functions", "government structures basic functions need to do to govern", "proper functions of government constitution framework", "government functions provide stabilize redistribute protect", "roles of government what does government do for citizens"], "tool_calls": 8, "tool_sequence": ["deep_search", "deep_search", "search", "get_document", "get_document", "get_document", "search", "search"], "turns": 5, "unique_search_targets": 8, "unique_tools": 3} +{"completion_tokens": 1151, "duplicate_tool_calls": 0, "elapsed_sec": 60.14228227222338, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 25, "found_relevant_docs": [], "prompt_tokens": 28652, "qid": "352818", "query": "how to cook string beans", "query_rewrites": 5, "reached": false, "relevant_docs": ["7072358"], "search_targets": ["how to cook string beans", "how to cook snap beans string beans", "how to cook string beans boiling steaming methods", "cook green beans boiling water", "boiling or steaming beans in salted water", "stir fry string beans recipe"], "tool_calls": 7, "tool_sequence": ["deep_search", "get_document", "get_document", "search", "get_document", "get_document", "search"], "turns": 5, "unique_search_targets": 6, "unique_tools": 3} +{"completion_tokens": 1395, "duplicate_tool_calls": 0, "elapsed_sec": 71.46158277383074, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 20, "found_relevant_docs": [], "prompt_tokens": 28949, "qid": "45924", "query": "average temperatures las vegas by month", "query_rewrites": 6, "reached": false, "relevant_docs": ["5167800", "7072691"], "search_targets": ["average temperatures las vegas by month", "las vegas temperatures averages by month", "las vegas average high and low temperatures each month", "average las vegas temperatures", "las vegas april temperatures", "las vegas average high low temperature january february march april may june july august september october november december", "las vegas average temperatures by month"], "tool_calls": 8, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "search", "get_document"], "turns": 5, "unique_search_targets": 7, "unique_tools": 3} +{"completion_tokens": 1674, "duplicate_tool_calls": 0, "elapsed_sec": 82.024238826707, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 43, "found_relevant_docs": ["7072843"], "prompt_tokens": 30309, "qid": "208145", "query": "how bicycle tire tubes are sized", "query_rewrites": 9, "reached": true, "relevant_docs": ["7072838", "7072843"], "search_targets": ["bicycle tire tube sizing how are they sized", "bicycle tire tube sizing etrto metric imperial", "bicycle tire sidewall numbers sizing meaning", "bicycle tire sizing fractions decimals french international sizing", "look at sidewall of bicycle tire tube size", "bicycle tire tube sizing etrto 26 inch 700c iso bead seat diameter", "bicycle tires inner tubes sizing diameters", "bicycle tire size 26 inch 700c sizing arcane complicated systems", "how to read bicycle tire size sidewall markings 700x23c 26x2.0 tube sizing"], "tool_calls": 9, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "search", "get_document", "get_document", "search"], "turns": 5, "unique_search_targets": 9, "unique_tools": 3} +{"completion_tokens": 1623, "duplicate_tool_calls": 0, "elapsed_sec": 57.815345717128366, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 19078, "qid": "79891", "query": "can you substitute chocolate chips for semi-sweet", "query_rewrites": 5, "reached": false, "relevant_docs": ["7073211"], "search_targets": ["substitute chocolate chips for semi-sweet chocolate baking", "substitute chocolate chips semi-sweet baking chocolate", "substitute chocolate chips semi-sweet", "chocolate chips substitute semi sweet baking chocolate", "unsweetened baking chocolate substitute semi-sweet"], "tool_calls": 5, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 5, "unique_tools": 2} +{"completion_tokens": 1218, "duplicate_tool_calls": 0, "elapsed_sec": 57.147985567804426, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 20, "found_relevant_docs": ["7073272"], "prompt_tokens": 19408, "qid": "208494", "query": "how big do newfypoo's get", "query_rewrites": 5, "reached": true, "relevant_docs": ["7073272"], "search_targets": ["newfypoo dog size weight height how big", "newfypoo size height weight", "newfypoo size height weight large breed poodle", "newfypoo height weight size", "newfypoo dog breed size weight height 60 130 pounds"], "tool_calls": 5, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "search"], "turns": 4, "unique_search_targets": 5, "unique_tools": 3} +{"completion_tokens": 1196, "duplicate_tool_calls": 0, "elapsed_sec": 63.83777767093852, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7073381"], "prompt_tokens": 18736, "qid": "319564", "query": "how much fiber is in carrots", "query_rewrites": 6, "reached": true, "relevant_docs": ["7073381"], "search_targets": ["fiber content in carrots", "fiber in carrots", "fiber cooked carrots", "carrots fiber grams", "baby carrots fiber", "carrot fiber content"], "tool_calls": 6, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 6, "unique_tools": 2} +{"completion_tokens": 1277, "duplicate_tool_calls": 0, "elapsed_sec": 57.06243131496012, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["502713"], "prompt_tokens": 13122, "qid": "155234", "query": "do bigger tires affect gas mileage", "query_rewrites": 3, "reached": true, "relevant_docs": ["502713"], "search_targets": ["do bigger tires affect gas mileage", "bigger tires affect gas mileage fuel economy", "bigger tires decrease gas mileage", "tire size affect fuel economy gas mileage"], "tool_calls": 4, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document"], "turns": 3, "unique_search_targets": 4, "unique_tools": 2} +{"completion_tokens": 1142, "duplicate_tool_calls": 0, "elapsed_sec": 69.48844194272533, "empty_tool_calls": 1, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 30, "found_relevant_docs": ["7073772"], "prompt_tokens": 26971, "qid": "14151", "query": "age requirements for name change", "query_rewrites": 7, "reached": true, "relevant_docs": ["7073772"], "search_targets": ["age requirements for name change", "age requirements name change texas 18", "california name change minor age requirements", "name change iowa age requirements", "age requirement change name minor adult 18 21 legal", "texas minor name change requirements age", "18 years old name change requirements", "name change minor child parental consent age 10 12 14 requirement"], "tool_calls": 8, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "deep_search", "get_document", "get_document", "deep_search"], "turns": 5, "unique_search_targets": 8, "unique_tools": 2} +{"completion_tokens": 1099, "duplicate_tool_calls": 0, "elapsed_sec": 54.403413402847946, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 15, "found_relevant_docs": ["7074071"], "prompt_tokens": 13238, "qid": "67802", "query": "can green tea cause stomach problems", "query_rewrites": 5, "reached": true, "relevant_docs": ["7074071"], "search_targets": ["can green tea cause stomach problems", "green tea stomach upset constipation", "green tea tannins stomach acid upset", "green tea side effects upset stomach nervousness", "green tea extract empty stomach adverse effects", "green tea caffeine upset stomach"], "tool_calls": 6, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 3, "unique_search_targets": 6, "unique_tools": 2} +{"completion_tokens": 600, "duplicate_tool_calls": 0, "elapsed_sec": 42.890536203980446, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7074235"], "prompt_tokens": 20144, "qid": "1090184", "query": "synonym of subordinate", "query_rewrites": 3, "reached": true, "relevant_docs": ["7074235"], "search_targets": ["synonym of subordinate", "subordinate synonyms", "subordinate adjective synonyms", "subordinate synonyms dictionary"], "tool_calls": 4, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 4, "unique_tools": 2} +{"completion_tokens": 1336, "duplicate_tool_calls": 0, "elapsed_sec": 67.45919466810301, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 25017, "qid": "323382", "query": "how much is the stamp to send a card", "query_rewrites": 10, "reached": false, "relevant_docs": ["4778293"], "search_targets": ["stamp price to send a card postage", "stamp price postcard cost 34 cents", "standard usps first class postage stamp costs 44 cents", "price for sending a post card january 2017", "cost of a postage stamp for a post card", "stamp price usps first class mail postage rates 2014", "current us postage stamp price 46 cents", "current price of a stamp how much it costs to send letters", "postage for a 6x8 card one postage stamp", "price of a stamp current price first class stamp postcard"], "tool_calls": 10, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 10, "unique_tools": 2} +{"completion_tokens": 599, "duplicate_tool_calls": 0, "elapsed_sec": 41.39567823102698, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7074377"], "prompt_tokens": 12059, "qid": "323998", "query": "how much magnesium in kidney beans", "query_rewrites": 2, "reached": true, "relevant_docs": ["7074377"], "search_targets": ["magnesium content in kidney beans", "magnesium kidney beans"], "tool_calls": 3, "tool_sequence": ["deep_search", "get_document", "get_document"], "turns": 3, "unique_search_targets": 2, "unique_tools": 2} +{"completion_tokens": 1411, "duplicate_tool_calls": 0, "elapsed_sec": 63.85293561592698, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["1956185"], "prompt_tokens": 22395, "qid": "91711", "query": "child psychiatrist salary 2016", "query_rewrites": 3, "reached": true, "relevant_docs": ["1956185"], "search_targets": ["child psychiatrist salary 2016", "child and adolescent psychiatrist salary", "average child psychiatrist salary 2016", "child psychiatrist salary 2016 physician"], "tool_calls": 7, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 4, "unique_tools": 2} +{"completion_tokens": 942, "duplicate_tool_calls": 0, "elapsed_sec": 62.307662460021675, "empty_tool_calls": 0, "first_relevant_tool_calls": 6, "first_relevant_turn": 3, "found_ids_count": 15, "found_relevant_docs": ["7074710"], "prompt_tokens": 23275, "qid": "125898", "query": "define prosthetic device", "query_rewrites": 6, "reached": true, "relevant_docs": ["7074710"], "search_targets": ["prosthetic device definition", "prosthetic device types", "prosthetic limb components", "prosthesis definition artificial device replacement body part", "prosthesis definition", "prosthetic definition"], "tool_calls": 8, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "search", "get_document", "get_document"], "turns": 5, "unique_search_targets": 6, "unique_tools": 3} +{"completion_tokens": 620, "duplicate_tool_calls": 0, "elapsed_sec": 34.05628034891561, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 10, "found_relevant_docs": [], "prompt_tokens": 12016, "qid": "289812", "query": "how many mm is a nickel coin", "query_rewrites": 1, "reached": false, "relevant_docs": ["262205"], "search_targets": ["nickel coin thickness mm", "5117925", "5117930"], "tool_calls": 3, "tool_sequence": ["search", "get_document", "get_document"], "turns": 3, "unique_search_targets": 3, "unique_tools": 2} +{"completion_tokens": 648, "duplicate_tool_calls": 0, "elapsed_sec": 63.91820939909667, "empty_tool_calls": 0, "first_relevant_tool_calls": 4, "first_relevant_turn": 3, "found_ids_count": 35, "found_relevant_docs": ["7075218"], "prompt_tokens": 30268, "qid": "333486", "query": "how old do you have to be to get a job in idaho", "query_rewrites": 6, "reached": true, "relevant_docs": ["7075218"], "search_targets": ["minimum age to get a job in idaho", "idaho legal working age minimum employment", "idaho", "idaho child labor laws minimum age employment 14 16", "idaho child labor laws minimum age employment", "idaho minimum age agricultural work exceptions child labor"], "tool_calls": 6, "tool_sequence": ["deep_search", "search", "get_document", "search", "get_document", "search"], "turns": 5, "unique_search_targets": 6, "unique_tools": 3} +{"completion_tokens": 947, "duplicate_tool_calls": 0, "elapsed_sec": 58.580430497881025, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7075317"], "prompt_tokens": 22784, "qid": "1090171", "query": "synonyms for the word discipline", "query_rewrites": 0, "reached": true, "relevant_docs": ["7075317"], "search_targets": ["synonyms for the word discipline", "7075312", "7075317", "7075315", "7075316", "8203248", "4080702", "7119389"], "tool_calls": 8, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 8, "unique_tools": 2} +{"completion_tokens": 1588, "duplicate_tool_calls": 0, "elapsed_sec": 70.63798946421593, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 45, "found_relevant_docs": [], "prompt_tokens": 27439, "qid": "73257", "query": "can seizure meds cause low sodium?", "query_rewrites": 3, "reached": false, "relevant_docs": ["7075398"], "search_targets": ["can seizure medications cause low sodium hyponatremia", "antiepileptic drugs hyponatremia low sodium side effect", "seizure medication carbamazepine oxcarbazepine hyponatremia low sodium", "1558108", "2150740", "2150741", "7575140", "4979675"], "tool_calls": 8, "tool_sequence": ["deep_search", "search", "search", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 5, "unique_search_targets": 8, "unique_tools": 3} +{"completion_tokens": 1445, "duplicate_tool_calls": 0, "elapsed_sec": 74.30262963334098, "empty_tool_calls": 0, "first_relevant_tool_calls": 0, "first_relevant_turn": 0, "found_ids_count": 30, "found_relevant_docs": [], "prompt_tokens": 29545, "qid": "1090170", "query": "synonyms for the word, decline", "query_rewrites": 11, "reached": false, "relevant_docs": ["7075411"], "search_targets": ["synonyms for the word decline", "synonyms for decline", "what's another word for decline", "refuse decline synonyms", "synonyms decrease decline", "slump decline synonyms", "in decline synonyms", "decline synonyms absence bankruptcy", "decline thesaurus synonyms", "synonyms word meaning", "\"synonyms for decline\" thesaurus decrease refuse reject"], "tool_calls": 11, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "get_document", "search"], "turns": 5, "unique_search_targets": 11, "unique_tools": 3} +{"completion_tokens": 1201, "duplicate_tool_calls": 0, "elapsed_sec": 64.23566694185138, "empty_tool_calls": 0, "first_relevant_tool_calls": 1, "first_relevant_turn": 1, "found_ids_count": 10, "found_relevant_docs": ["7075449"], "prompt_tokens": 19397, "qid": "237373", "query": "how is soil created from rocks", "query_rewrites": 5, "reached": true, "relevant_docs": ["7075449"], "search_targets": ["how is soil created from rocks", "weathering physical breakdown chemical alteration of rocks to form soil", "ingredients for making soil tiny pieces of rock", "weathering process breaks rock into smaller pieces soil", "weathering breaking down rock into smaller pieces", "how rocks are made and how soil is formed"], "tool_calls": 6, "tool_sequence": ["deep_search", "get_document", "get_document", "get_document", "get_document", "get_document"], "turns": 4, "unique_search_targets": 6, "unique_tools": 2} diff --git a/src/synaptic/__init__.py b/src/synaptic/__init__.py index 5a0f247..0759ef6 100644 --- a/src/synaptic/__init__.py +++ b/src/synaptic/__init__.py @@ -52,6 +52,26 @@ RuleBasedRelationDetector, ) from synaptic.graph import SynapticGraph +from synaptic.indexing import ( + CandidateProvider, + CandidateScoreSource, + CandidateSearchRequest, + CandidateSearchResult, + IndexFilter, + IndexHealthBackend, + IndexLagReport, + IndexRouter, + IngestionJob, + IngestionJobStage, + IngestionJobStatus, + IngestionJobStore, + InProcessIndexRouter, + OpenSearchCandidateProvider, + QdrantCandidateProvider, + ScoredCandidate, + StorageCandidateProvider, + unique_candidates, +) from synaptic.models import ( ActivatedNode, ConsolidationLevel, @@ -104,6 +124,10 @@ "ActivityTracker", "AgentSearch", "ChainedEntityExtractor", + "CandidateProvider", + "CandidateScoreSource", + "CandidateSearchRequest", + "CandidateSearchResult", "ChunkEntityIndex", "ClassificationResult", "ConsolidationLevel", @@ -121,6 +145,17 @@ "GraphTraversal", "HybridClassifier", "HybridEntityExtractor", + "IndexFilter", + "IndexHealthBackend", + "IndexLagReport", + "IndexRouter", + "InProcessIndexRouter", + "OpenSearchCandidateProvider", + "QdrantCandidateProvider", + "IngestionJob", + "IngestionJobStage", + "IngestionJobStatus", + "IngestionJobStore", "KindClassifier", "LLMClassifier", "LLMOpenIEExtractor", @@ -154,8 +189,10 @@ "RuleBasedRelationDetector", "SearchIntent", "SearchResult", + "ScoredCandidate", "SpaCyEntityExtractor", "StorageBackend", + "StorageCandidateProvider", "SynapticGraph", "TableIngester", "TagExtractor", @@ -165,6 +202,7 @@ "purge_openie_artifacts", "memory_scope_key", "suggest_intent", + "unique_candidates", # v0.12 "DomainProfile", "ProfileGenerator", diff --git a/src/synaptic/__init__.pyi b/src/synaptic/__init__.pyi index 37868d9..b0e405f 100644 --- a/src/synaptic/__init__.pyi +++ b/src/synaptic/__init__.pyi @@ -50,6 +50,24 @@ from synaptic.extensions.relation_detector import ( from synaptic.extensions.relation_detector_llm import LLMRelationDetector as LLMRelationDetector from synaptic.extensions.table_ingester import TableIngester as TableIngester from synaptic.graph import SynapticGraph as SynapticGraph +from synaptic.indexing import CandidateProvider as CandidateProvider +from synaptic.indexing import CandidateScoreSource as CandidateScoreSource +from synaptic.indexing import CandidateSearchRequest as CandidateSearchRequest +from synaptic.indexing import CandidateSearchResult as CandidateSearchResult +from synaptic.indexing import IndexFilter as IndexFilter +from synaptic.indexing import IndexHealthBackend as IndexHealthBackend +from synaptic.indexing import IndexLagReport as IndexLagReport +from synaptic.indexing import IndexRouter as IndexRouter +from synaptic.indexing import IngestionJob as IngestionJob +from synaptic.indexing import IngestionJobStage as IngestionJobStage +from synaptic.indexing import IngestionJobStatus as IngestionJobStatus +from synaptic.indexing import IngestionJobStore as IngestionJobStore +from synaptic.indexing import InProcessIndexRouter as InProcessIndexRouter +from synaptic.indexing import OpenSearchCandidateProvider as OpenSearchCandidateProvider +from synaptic.indexing import QdrantCandidateProvider as QdrantCandidateProvider +from synaptic.indexing import ScoredCandidate as ScoredCandidate +from synaptic.indexing import StorageCandidateProvider as StorageCandidateProvider +from synaptic.indexing import unique_candidates as unique_candidates from synaptic.models import ActivatedNode as ActivatedNode from synaptic.models import ConsolidationLevel as ConsolidationLevel from synaptic.models import DigestResult as DigestResult diff --git a/src/synaptic/backends/memory.py b/src/synaptic/backends/memory.py index c7d1c72..e93424f 100644 --- a/src/synaptic/backends/memory.py +++ b/src/synaptic/backends/memory.py @@ -6,6 +6,7 @@ from collections.abc import Sequence from difflib import SequenceMatcher +from synaptic.indexing import IndexLagReport, IngestionJob, IngestionJobStage, IngestionJobStatus from synaptic.models import ( ConsolidationLevel, Edge, @@ -23,7 +24,15 @@ class MemoryBackend: """Dict-based in-memory backend. No external dependencies.""" - __slots__ = ("_edges", "_memory_events", "_memory_scores", "_nodes", "_retrieval_events") + __slots__ = ( + "_edges", + "_index_lag_reports", + "_ingestion_jobs", + "_memory_events", + "_memory_scores", + "_nodes", + "_retrieval_events", + ) def __init__(self) -> None: self._nodes: dict[str, Node] = {} @@ -31,6 +40,8 @@ def __init__(self) -> None: self._memory_events: dict[str, MemoryEvent] = {} self._retrieval_events: dict[str, RetrievalEvent] = {} self._memory_scores: dict[tuple[str, str, str], MemoryScore] = {} + self._ingestion_jobs: dict[str, IngestionJob] = {} + self._index_lag_reports: list[IndexLagReport] = [] async def connect(self) -> None: pass @@ -41,6 +52,8 @@ async def close(self) -> None: self._memory_events.clear() self._retrieval_events.clear() self._memory_scores.clear() + self._ingestion_jobs.clear() + self._index_lag_reports.clear() # --- Node CRUD --- @@ -390,6 +403,60 @@ async def list_memory_scores( break return out + async def save_ingestion_job(self, job: IngestionJob) -> None: + self._ingestion_jobs[job.id] = job + + async def get_ingestion_job(self, job_id: str) -> IngestionJob | None: + return self._ingestion_jobs.get(job_id) + + async def list_ingestion_jobs( + self, + *, + document_id: str = "", + stage: str | IngestionJobStage | None = None, + status: str | IngestionJobStatus | None = None, + limit: int = 100, + ) -> list[IngestionJob]: + out: list[IngestionJob] = [] + jobs = sorted(self._ingestion_jobs.values(), key=lambda job: job.updated_at, reverse=True) + for job in jobs: + if document_id and job.document_id != document_id: + continue + if stage is not None and str(job.stage) != str(stage): + continue + if status is not None and str(job.status) != str(status): + continue + out.append(job) + if len(out) >= limit: + break + return out + + async def save_index_lag_report(self, report: IndexLagReport) -> None: + self._index_lag_reports.append(report) + + async def list_index_lag_reports( + self, + *, + provider: str = "", + since: float | None = None, + limit: int = 100, + ) -> list[IndexLagReport]: + out: list[IndexLagReport] = [] + reports = sorted( + self._index_lag_reports, + key=lambda report: report.checked_at, + reverse=True, + ) + for report in reports: + if provider and report.provider != provider: + continue + if since is not None and report.checked_at < since: + continue + out.append(report) + if len(out) >= limit: + break + return out + # --- Search --- async def search_fts(self, query: str, *, limit: int = 20) -> list[Node]: diff --git a/src/synaptic/backends/sqlite.py b/src/synaptic/backends/sqlite.py index 54d4bd4..9b69593 100644 --- a/src/synaptic/backends/sqlite.py +++ b/src/synaptic/backends/sqlite.py @@ -396,6 +396,7 @@ def _normalize_korean(text: str, *, query_mode: bool = False) -> str: return _KO_PARTICLE.sub(r"\1", text) +from synaptic.indexing import IndexLagReport, IngestionJob, IngestionJobStage, IngestionJobStatus from synaptic.models import ( ConsolidationLevel, Edge, @@ -493,6 +494,33 @@ def _normalize_korean(text: str, *, query_mode: bool = False) -> str: PRIMARY KEY(scope_key, node_id, edge_id) ); +CREATE TABLE IF NOT EXISTS syn_ingestion_jobs ( + id TEXT PRIMARY KEY, + document_id TEXT NOT NULL DEFAULT '', + version TEXT NOT NULL DEFAULT '', + stage TEXT NOT NULL DEFAULT 'discover', + status TEXT NOT NULL DEFAULT 'pending', + attempt INTEGER NOT NULL DEFAULT 0, + error TEXT NOT NULL DEFAULT '', + properties_json TEXT NOT NULL DEFAULT '{}', + created_at REAL NOT NULL, + updated_at REAL NOT NULL +); + +CREATE TABLE IF NOT EXISTS syn_index_lag_reports ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'unknown', + index_generation TEXT NOT NULL DEFAULT '', + pending_documents INTEGER NOT NULL DEFAULT 0, + pending_chunks INTEGER NOT NULL DEFAULT 0, + failed_jobs INTEGER NOT NULL DEFAULT 0, + lag_seconds REAL NOT NULL DEFAULT 0.0, + p95_index_latency_seconds REAL NOT NULL DEFAULT 0.0, + properties_json TEXT NOT NULL DEFAULT '{}', + checked_at REAL NOT NULL +); + CREATE INDEX IF NOT EXISTS idx_syn_edges_source ON syn_edges(source_id); CREATE INDEX IF NOT EXISTS idx_syn_edges_target ON syn_edges(target_id); CREATE INDEX IF NOT EXISTS idx_syn_edges_source_kind ON syn_edges(source_id, kind); @@ -504,6 +532,12 @@ def _normalize_korean(text: str, *, query_mode: bool = False) -> str: CREATE INDEX IF NOT EXISTS idx_syn_memory_events_kind ON syn_memory_events(kind, created_at); CREATE INDEX IF NOT EXISTS idx_syn_retrieval_events_scope ON syn_retrieval_events(scope_key, created_at); CREATE INDEX IF NOT EXISTS idx_syn_memory_scores_node ON syn_memory_scores(node_id, score); +CREATE INDEX IF NOT EXISTS idx_syn_ingestion_jobs_document + ON syn_ingestion_jobs(document_id, updated_at); +CREATE INDEX IF NOT EXISTS idx_syn_ingestion_jobs_stage_status + ON syn_ingestion_jobs(stage, status, updated_at); +CREATE INDEX IF NOT EXISTS idx_syn_index_lag_reports_provider + ON syn_index_lag_reports(provider, checked_at); """ @@ -1499,6 +1533,121 @@ async def list_memory_scores( rows = await cur.fetchall() return [_row_to_memory_score(r) for r in rows] + async def save_ingestion_job(self, job: IngestionJob) -> None: + db = self._db() + await db.execute( + """INSERT INTO syn_ingestion_jobs + (id, document_id, version, stage, status, attempt, error, + properties_json, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + document_id=excluded.document_id, + version=excluded.version, + stage=excluded.stage, + status=excluded.status, + attempt=excluded.attempt, + error=excluded.error, + properties_json=excluded.properties_json, + created_at=excluded.created_at, + updated_at=excluded.updated_at""", + ( + job.id, + job.document_id, + job.version, + str(job.stage), + str(job.status), + job.attempt, + job.error, + json.dumps(job.properties), + job.created_at, + job.updated_at, + ), + ) + await db.commit() + + async def get_ingestion_job(self, job_id: str) -> IngestionJob | None: + db = self._db() + async with db.execute("SELECT * FROM syn_ingestion_jobs WHERE id = ?", (job_id,)) as cur: + row = await cur.fetchone() + return _row_to_ingestion_job(row) if row is not None else None + + async def list_ingestion_jobs( + self, + *, + document_id: str = "", + stage: str | IngestionJobStage | None = None, + status: str | IngestionJobStatus | None = None, + limit: int = 100, + ) -> list[IngestionJob]: + db = self._db() + clauses: list[str] = [] + params: list[object] = [] + if document_id: + clauses.append("document_id = ?") + params.append(document_id) + if stage is not None: + clauses.append("stage = ?") + params.append(str(stage)) + if status is not None: + clauses.append("status = ?") + params.append(str(status)) + where = f" WHERE {' AND '.join(clauses)}" if clauses else "" + params.append(int(limit)) + async with db.execute( + f"SELECT * FROM syn_ingestion_jobs{where} ORDER BY updated_at DESC LIMIT ?", + params, + ) as cur: + rows = await cur.fetchall() + return [_row_to_ingestion_job(r) for r in rows] + + async def save_index_lag_report(self, report: IndexLagReport) -> None: + db = self._db() + await db.execute( + """INSERT INTO syn_index_lag_reports + (provider, status, index_generation, pending_documents, + pending_chunks, failed_jobs, lag_seconds, + p95_index_latency_seconds, properties_json, checked_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + report.provider, + report.status, + report.index_generation, + report.pending_documents, + report.pending_chunks, + report.failed_jobs, + report.lag_seconds, + report.p95_index_latency_seconds, + json.dumps(report.properties), + report.checked_at, + ), + ) + await db.commit() + + async def list_index_lag_reports( + self, + *, + provider: str = "", + since: float | None = None, + limit: int = 100, + ) -> list[IndexLagReport]: + db = self._db() + clauses: list[str] = [] + params: list[object] = [] + if provider: + clauses.append("provider = ?") + params.append(provider) + if since is not None: + clauses.append("checked_at >= ?") + params.append(float(since)) + where = f" WHERE {' AND '.join(clauses)}" if clauses else "" + params.append(int(limit)) + async with db.execute( + f"SELECT * FROM syn_index_lag_reports{where} ORDER BY checked_at DESC LIMIT ?", + params, + ) as cur: + rows = await cur.fetchall() + return [_row_to_index_lag_report(r) for r in rows] + # --- Batch read --- async def get_nodes_batch(self, node_ids: list[str]) -> list[Node]: @@ -2336,6 +2485,16 @@ def _json_str_dict(raw: str | None) -> dict[str, str]: return {str(k): str(v) for k, v in data.items()} +def _json_object_dict(raw: str | None) -> dict[str, object]: + if not raw: + return {} + try: + data = json.loads(raw) + except json.JSONDecodeError: + return {} + return data if isinstance(data, dict) else {} + + def _prop_int(props: dict[str, str], key: str, default: int) -> int: try: return int(float(props.get(key, default))) @@ -2387,3 +2546,33 @@ def _row_to_memory_score(row: aiosqlite.Row) -> MemoryScore: score=float(row["score"]), updated_at=float(row["updated_at"]), ) + + +def _row_to_ingestion_job(row: aiosqlite.Row) -> IngestionJob: + return IngestionJob( + id=row["id"], + document_id=row["document_id"], + version=row["version"], + stage=IngestionJobStage(row["stage"]), + status=IngestionJobStatus(row["status"]), + attempt=int(row["attempt"]), + error=row["error"], + properties=_json_str_dict(row["properties_json"]), + created_at=float(row["created_at"]), + updated_at=float(row["updated_at"]), + ) + + +def _row_to_index_lag_report(row: aiosqlite.Row) -> IndexLagReport: + return IndexLagReport( + provider=row["provider"], + status=row["status"], + index_generation=row["index_generation"], + pending_documents=int(row["pending_documents"]), + pending_chunks=int(row["pending_chunks"]), + failed_jobs=int(row["failed_jobs"]), + lag_seconds=float(row["lag_seconds"]), + p95_index_latency_seconds=float(row["p95_index_latency_seconds"]), + properties=_json_object_dict(row["properties_json"]), + checked_at=float(row["checked_at"]), + ) diff --git a/src/synaptic/extensions/evidence_search.py b/src/synaptic/extensions/evidence_search.py index 0ce9217..3976fef 100644 --- a/src/synaptic/extensions/evidence_search.py +++ b/src/synaptic/extensions/evidence_search.py @@ -63,12 +63,15 @@ ScoredCandidate, ) from synaptic.extensions.query_anchor import QueryAnchorExtractor, QueryAnchors +from synaptic.indexing import CandidateSearchRequest, IndexFilter +from synaptic.models import MemoryScope, Node from synaptic.ppr import personalized_pagerank if TYPE_CHECKING: from synaptic.extensions.embedder import EmbeddingProvider from synaptic.extensions.query_anchor import PhraseExtractorProtocol from synaptic.extensions.reranker_cross import RerankerProtocol + from synaptic.indexing import IndexRouter from synaptic.protocols import QueryDecomposer, StorageBackend # RRF (Reciprocal Rank Fusion) constant — 60 is the canonical k from the @@ -170,6 +173,7 @@ class EvidenceSearch: "_expansion_budget", "_fusion_mode", "_graph_expansion", + "_index_router", "_phrase_cache", "_phrase_cache_loaded", "_query_phrase_seed_k", @@ -200,8 +204,10 @@ def __init__( adaptive: bool = False, graph_expansion: bool = True, aggregate_candidate_pool_limit: int | None = None, + index_router: IndexRouter | None = None, ) -> None: self._backend = backend + self._index_router = index_router self._embedder = embedder self._cross_reranker = reranker self._decomposer = decomposer @@ -417,6 +423,22 @@ def _tilt_weights(anchors: QueryAnchors, fts_nodes: list) -> RerankerWeights | N structural=0.10, ) + async def _hydrate_candidate_ids(self, node_ids: list[str]) -> dict[str, Node]: + """Hydrate routed candidate ids using batch reads when available.""" + if not node_ids: + return {} + unique_ids = list(dict.fromkeys(node_ids)) + get_batch = getattr(self._backend, "get_nodes_batch", None) + if callable(get_batch): + nodes = await get_batch(unique_ids) + return {node.id: node for node in nodes} + out: dict[str, object] = {} + for node_id in unique_ids: + node = await self._backend.get_node(node_id) + if node is not None: + out[node_id] = node + return out + async def search( self, query: str, @@ -425,6 +447,8 @@ async def search( fts_seed_limit: int = 20, per_document_cap: int = 2, query_embedding: list[float] | None = None, + index_filters: IndexFilter | None = None, + scope: MemoryScope | None = None, ) -> EvidenceSearchResult: """Run the full 3rd-gen pipeline for ``query``. @@ -441,6 +465,9 @@ async def search( query_embedding: Optional query vector for the semantic signal. When ``None`` the reranker falls back to lexical + graph + structural only. + index_filters: Optional provider-side filters for the routed + candidate path. The default empty filter keeps legacy behavior. + scope: Optional memory scope attached to routed candidate requests. """ t0 = time() timings_ms: dict[str, float] = {} @@ -469,31 +496,71 @@ async def search( stage_t0 = time() fts_scores: dict[str, float] = {} fts_include_embedding = query_embedding is not None + routed_fts = False # Fetch real scores when L01 is on OR adaptive mode may need them. want_scores = self._backend_scored and (self._real_scores_enabled or self._adaptive) - if want_scores: - if self._backend_fts_include_embedding: - scored_fts = await self._backend.search_fts( - query, - limit=fts_seed_limit, - with_scores=True, - include_embedding=fts_include_embedding, - ) - else: - scored_fts = await self._backend.search_fts( - query, limit=fts_seed_limit, with_scores=True - ) - fts_nodes = [n for n, _ in scored_fts] - else: - scored_fts = None - if self._backend_fts_include_embedding: - fts_nodes = await self._backend.search_fts( - query, - limit=fts_seed_limit, - include_embedding=fts_include_embedding, + if self._index_router is not None: + try: + router_result = await self._index_router.search_candidates( + CandidateSearchRequest( + query=query, + limit=fts_seed_limit, + filters=index_filters or IndexFilter(), + scope=scope or MemoryScope(), + embedding=query_embedding, + ) ) + diagnostics["router_candidate_count"] = float(len(router_result.candidates)) + diagnostics["router_provider_count"] = float(len(router_result.provider_counts)) + for provider, count in router_result.provider_counts.items(): + diagnostics[f"router_provider_{provider}_count"] = float(count) + for provider, score in router_result.score_ranges.items(): + diagnostics[f"router_provider_{provider}_max_score"] = float(score) + + candidate_ids = [candidate.node_id for candidate in router_result.candidates] + hydrated = await self._hydrate_candidate_ids(candidate_ids) + fts_nodes = [] + missing = 0 + for rank, candidate in enumerate(router_result.candidates): + node = hydrated.get(candidate.node_id) + if node is None: + missing += 1 + continue + fts_nodes.append(node) + score = max(0.0, min(1.0, float(candidate.score))) + fts_scores[candidate.node_id] = score or max(0.10, 0.95 - rank * 0.03) + diagnostics["router_missing_count"] = float(missing) + scored_fts = None + routed_fts = True + except Exception as exc: + logger.warning("index router search failed; falling back to backend FTS: %s", exc) + diagnostics["router_failed"] = 1.0 + routed_fts = False + + if not routed_fts: + if want_scores: + if self._backend_fts_include_embedding: + scored_fts = await self._backend.search_fts( + query, + limit=fts_seed_limit, + with_scores=True, + include_embedding=fts_include_embedding, + ) + else: + scored_fts = await self._backend.search_fts( + query, limit=fts_seed_limit, with_scores=True + ) + fts_nodes = [n for n, _ in scored_fts] else: - fts_nodes = await self._backend.search_fts(query, limit=fts_seed_limit) + scored_fts = None + if self._backend_fts_include_embedding: + fts_nodes = await self._backend.search_fts( + query, + limit=fts_seed_limit, + include_embedding=fts_include_embedding, + ) + else: + fts_nodes = await self._backend.search_fts(query, limit=fts_seed_limit) # Adaptive coverage signal — computed once, gates L01/L02/L05 below. coverage = self._anchor_coverage(anchors, fts_nodes) if self._adaptive else None @@ -506,7 +573,9 @@ async def search( self._real_scores_enabled or (self._adaptive and coverage is not None and coverage >= _COVERAGE_HI) ) - if use_real: + if routed_fts: + pass + elif use_real: for node, rel in scored_fts: fts_scores[node.id] = 0.10 + rel * 0.85 else: diff --git a/src/synaptic/graph.py b/src/synaptic/graph.py index 07d2871..fbfac57 100644 --- a/src/synaptic/graph.py +++ b/src/synaptic/graph.py @@ -304,6 +304,7 @@ class SynapticGraph: "_embedder", "_evidence_search_cache", "_hebbian", + "_index_router", "_json_exporter", "_md_exporter", "_ontology", @@ -331,6 +332,7 @@ def __init__( query_decomposer: QueryDecomposer | None = None, reranker: object | None = None, reranker_weights: object | None = None, + index_router: object | None = None, cache_size: int = 256, vector_min_cosine: float | None = None, vector_relative_drop: float | None = None, @@ -356,6 +358,7 @@ def __init__( self._chunk_entity_index = chunk_entity_index self._query_decomposer = query_decomposer self._reranker = reranker + self._index_router = index_router # Optional RerankerWeights for the hybrid reranker. None keeps the # built-in defaults; set it (e.g. via the ``reranker_weights`` property # or a factory arg) to enable the usage/time memory axis. @@ -1494,6 +1497,22 @@ async def sync_from_database( def backend(self) -> StorageBackend: return self._backend + @property + def index_router(self) -> object | None: + """Optional candidate router used by EvidenceSearch. + + ``None`` keeps the legacy StorageBackend seed path. Setting a router + switches new searcher instances to candidate-id retrieval while existing + cached searchers are discarded. + """ + + return self._index_router + + @index_router.setter + def index_router(self, router: object | None) -> None: + self._index_router = router + self._evidence_search_cache.clear() + @property def reranker_weights(self) -> object | None: """Weights for the hybrid reranker (a :class:`RerankerWeights`). @@ -2484,6 +2503,8 @@ async def _search_via_evidence( search_kwargs["per_document_cap"] = per_document_cap if embedding is not None: search_kwargs["query_embedding"] = embedding + if scope is not None: + search_kwargs["scope"] = scope ev_result = await searcher.search(query, **search_kwargs) # Adapter: Evidence (node + score + reason) → ActivatedNode @@ -2573,6 +2594,7 @@ def _get_evidence_search(self, active_reranker: object | None) -> object: id(active_reranker), id(self._reranker_weights), id(self._query_decomposer), + id(self._index_router), ) searcher = self._evidence_search_cache.get(key) if searcher is None: @@ -2585,6 +2607,7 @@ def _get_evidence_search(self, active_reranker: object | None) -> object: reranker=active_reranker, reranker_weights=self._reranker_weights, decomposer=self._query_decomposer, + index_router=self._index_router, ) self._evidence_search_cache[key] = searcher return searcher diff --git a/src/synaptic/indexing.py b/src/synaptic/indexing.py new file mode 100644 index 0000000..46aaec4 --- /dev/null +++ b/src/synaptic/indexing.py @@ -0,0 +1,1341 @@ +"""Scale-out indexing contracts. + +These types define the optional boundary between the retrieval core and +large external indexes. They do not change the default StorageBackend path; +instead they make the 10TB route explicit: providers return scored candidate +ids, and the normal retrieval pipeline decides how to hydrate, expand, rerank, +and record them. +""" + +from __future__ import annotations + +import inspect +from collections.abc import Sequence +from dataclasses import dataclass, field +from enum import StrEnum +from time import time +from typing import Protocol +from uuid import uuid4 + +from synaptic.models import MemoryScope + +__all__ = [ + "CandidateProvider", + "CandidateScoreSource", + "CandidateSearchRequest", + "CandidateSearchResult", + "IndexFilter", + "IndexHealthBackend", + "IndexLagReport", + "IndexRouter", + "InProcessIndexRouter", + "IngestionJob", + "IngestionJobStage", + "IngestionJobStatus", + "IngestionJobStore", + "OpenSearchCandidateProvider", + "QdrantCandidateProvider", + "ScoredCandidate", + "StorageCandidateProvider", + "unique_candidates", +] + + +def _new_id() -> str: + return uuid4().hex[:16] + + +def _str_dict() -> dict[str, str]: + return {} + + +def _object_dict() -> dict[str, object]: + return {} + + +def _int_dict() -> dict[str, int]: + return {} + + +def _float_dict() -> dict[str, float]: + return {} + + +def _scope() -> MemoryScope: + return MemoryScope() + + +class CandidateScoreSource(StrEnum): + """Origin of a candidate score before router-level fusion.""" + + LEXICAL = "lexical" + VECTOR = "vector" + GRAPH = "graph" + MEMORY = "memory" + RERANK = "rerank" + ROUTER = "router" + + +class IngestionJobStage(StrEnum): + """Durable ingestion/indexing stages for large document corpora.""" + + DISCOVER = "discover" + PARSE = "parse" + OCR = "ocr" + CHUNK = "chunk" + EMBED = "embed" + LEXICAL_INDEX = "lexical_index" + VECTOR_INDEX = "vector_index" + GRAPH_INDEX = "graph_index" + SEMANTIC_EXTRACT = "semantic_extract" + LEDGER_COMMIT = "ledger_commit" + + +class IngestionJobStatus(StrEnum): + """State of a durable ingestion job.""" + + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + SKIPPED = "skipped" + + +@dataclass(slots=True) +class IndexFilter: + """Provider-side filters that must be applied before candidate materialization. + + Empty strings and empty lists mean "no filter". Providers must not interpret + an empty string as a literal workspace/user/session/domain value. + """ + + workspace_id: str = "" + user_id: str = "" + session_id: str = "" + domain: str = "" + acl_policy_ids: list[str] = field(default_factory=list) + source_ids: list[str] = field(default_factory=list) + document_ids: list[str] = field(default_factory=list) + languages: list[str] = field(default_factory=list) + mime_types: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + created_after: float | None = None + created_before: float | None = None + updated_after: float | None = None + updated_before: float | None = None + properties: dict[str, str] = field(default_factory=_str_dict) + + +@dataclass(slots=True) +class CandidateSearchRequest: + """A routed candidate search request. + + `query` is the user-visible search text. `query_variants` contains bounded + deterministic rewrites or agent-proposed follow-up targets. Providers may + search variants independently but must preserve `query_variant` in returned + candidates so diagnostics can explain which route found each hit. + """ + + query: str + query_variants: list[str] = field(default_factory=list) + limit: int = 20 + filters: IndexFilter = field(default_factory=IndexFilter) + scope: MemoryScope = field(default_factory=_scope) + embedding: list[float] | None = None + providers: list[str] = field(default_factory=list) + properties: dict[str, str] = field(default_factory=_str_dict) + + +@dataclass(slots=True) +class ScoredCandidate: + """Scored candidate id returned by an external index provider.""" + + node_id: str + document_id: str = "" + score: float = 0.0 + score_source: CandidateScoreSource = CandidateScoreSource.ROUTER + rank: int = 0 + query_variant: str = "" + provider: str = "" + index_generation: str = "" + metadata: dict[str, object] = field(default_factory=_object_dict) + + +@dataclass(slots=True) +class CandidateSearchResult: + """Provider/router candidate response before node hydration.""" + + candidates: list[ScoredCandidate] = field(default_factory=list) + provider_counts: dict[str, int] = field(default_factory=_int_dict) + score_ranges: dict[str, float] = field(default_factory=_float_dict) + diagnostics: dict[str, object] = field(default_factory=_object_dict) + index_generation: str = "" + + +@dataclass(slots=True) +class IndexLagReport: + """Operational freshness report for an index or provider.""" + + provider: str = "" + status: str = "unknown" + index_generation: str = "" + pending_documents: int = 0 + pending_chunks: int = 0 + failed_jobs: int = 0 + lag_seconds: float = 0.0 + p95_index_latency_seconds: float = 0.0 + properties: dict[str, object] = field(default_factory=_object_dict) + checked_at: float = field(default_factory=time) + + +@dataclass(slots=True) +class IngestionJob: + """Durable unit of indexing work for incremental corpus updates.""" + + id: str = field(default_factory=_new_id) + document_id: str = "" + version: str = "" + stage: IngestionJobStage = IngestionJobStage.DISCOVER + status: IngestionJobStatus = IngestionJobStatus.PENDING + attempt: int = 0 + error: str = "" + properties: dict[str, str] = field(default_factory=_str_dict) + created_at: float = field(default_factory=time) + updated_at: float = field(default_factory=time) + + +class CandidateProvider(Protocol): + """External index provider that returns candidate ids, not hydrated nodes.""" + + @property + def name(self) -> str: ... + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: ... + + async def index_health(self) -> IndexLagReport: ... + + +class IndexRouter(Protocol): + """Coordinates multiple providers and returns fused candidate ids.""" + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: ... + + async def index_health(self) -> list[IndexLagReport]: ... + + +class IngestionJobStore(Protocol): + """Durable ingestion queue/state store.""" + + async def save_ingestion_job(self, job: IngestionJob) -> None: ... + async def get_ingestion_job(self, job_id: str) -> IngestionJob | None: ... + async def list_ingestion_jobs( + self, + *, + document_id: str = "", + stage: str | IngestionJobStage | None = None, + status: str | IngestionJobStatus | None = None, + limit: int = 100, + ) -> list[IngestionJob]: ... + + +class IndexHealthBackend(Protocol): + """Aggregated freshness/lag reporting across index providers.""" + + async def save_index_lag_report(self, report: IndexLagReport) -> None: ... + async def list_index_lag_reports( + self, + *, + provider: str = "", + since: float | None = None, + limit: int = 100, + ) -> list[IndexLagReport]: ... + + +class StorageCandidateProvider: + """Compatibility provider backed by the existing StorageBackend. + + This is the Phase-2 bridge: it proves the router boundary without requiring + OpenSearch/Qdrant. External providers should apply filters natively before + scoring; this compatibility provider over-fetches and post-filters because + legacy StorageBackend search methods do not accept a structured filter. + """ + + __slots__ = ("_backend", "_max_overfetch", "_name", "_overfetch_factor") + + def __init__( + self, + backend: object, + *, + name: str = "storage_fts", + overfetch_factor: int = 5, + max_overfetch: int = 1000, + ) -> None: + self._backend = backend + self._name = name + self._overfetch_factor = max(1, int(overfetch_factor)) + self._max_overfetch = max(1, int(max_overfetch)) + + @property + def name(self) -> str: + return self._name + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + limit = max(0, int(request.limit)) + if limit == 0: + return CandidateSearchResult(index_generation="storage") + + queries = _unique_queries(request.query, request.query_variants) + if not queries: + return CandidateSearchResult(index_generation="storage") + + search_limit = limit + if _has_filters(request.filters): + search_limit = min(self._max_overfetch, max(limit, limit * self._overfetch_factor)) + + candidates: list[ScoredCandidate] = [] + for query_variant in queries: + rows = await self._search_fts( + query_variant, + limit=search_limit, + include_embedding=request.embedding is not None, + ) + for rank, row in enumerate(rows): + node, score = _node_and_score(row, rank) + if not _matches_filter(node, request.filters): + continue + candidates.append( + ScoredCandidate( + node_id=str(getattr(node, "id", "")), + document_id=_document_id(node), + score=score, + score_source=CandidateScoreSource.LEXICAL, + rank=rank + 1, + query_variant=query_variant, + provider=self.name, + index_generation="storage", + metadata=_candidate_metadata(node), + ) + ) + + return CandidateSearchResult( + candidates=candidates, + provider_counts={self.name: len(candidates)}, + score_ranges=_provider_score_max(candidates), + diagnostics={ + "query_variant_count": len(queries), + "filtered": _has_filters(request.filters), + "search_limit": search_limit, + }, + index_generation="storage", + ) + + async def index_health(self) -> IndexLagReport: + return IndexLagReport(provider=self.name, status="ok", index_generation="storage") + + async def _search_fts( + self, + query: str, + *, + limit: int, + include_embedding: bool, + ) -> list[object]: + search_fts = self._backend.search_fts + kwargs: dict[str, object] = {"limit": limit} + try: + params = inspect.signature(search_fts).parameters + except (TypeError, ValueError): + params = {} + if "with_scores" in params: + kwargs["with_scores"] = True + if "include_embedding" in params: + kwargs["include_embedding"] = include_embedding + return list(await search_fts(query, **kwargs)) + + +class OpenSearchCandidateProvider: + """OpenSearch/Elasticsearch lexical provider. + + The provider accepts an already-created async or sync client so importing + synaptic does not require OpenSearch dependencies. It applies structured + `IndexFilter` clauses in the index query before materialization and returns + only scored candidate ids plus compact metadata. + """ + + __slots__ = ( + "_client", + "_field_map", + "_index", + "_metadata_fields", + "_name", + "_query_fields", + ) + + def __init__( + self, + client: object, + *, + index: str, + name: str = "opensearch", + query_fields: Sequence[str] = ("title^3", "content", "tags", "properties.search_keywords"), + metadata_fields: Sequence[str] = ( + "node_id", + "document_id", + "doc_id", + "title", + "kind", + "source", + "page", + "chunk_id", + "category", + "domain", + "language", + ), + field_map: dict[str, str] | None = None, + ) -> None: + self._client = client + self._index = index + self._name = name + self._query_fields = list(query_fields) + self._metadata_fields = list(metadata_fields) + self._field_map = { + "node_id": "node_id", + "document_id": "document_id", + "doc_id": "doc_id", + "workspace_id": "workspace_id", + "user_id": "user_id", + "session_id": "session_id", + "domain": "domain", + "acl_policy_id": "acl_policy_id", + "source_id": "source_id", + "source": "source", + "language": "language", + "mime_type": "mime_type", + "tags": "tags", + "created_at": "created_at", + "updated_at": "updated_at", + **(field_map or {}), + } + + @property + def name(self) -> str: + return self._name + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + limit = max(0, int(request.limit)) + if limit == 0: + return CandidateSearchResult(index_generation=self._index) + queries = _unique_queries(request.query, request.query_variants) + if not queries: + return CandidateSearchResult(index_generation=self._index) + + raw_hits: list[tuple[dict[str, object], str, int]] = [] + for query_variant in queries: + body = self._query_body(query_variant, request.filters, limit=limit) + response = await _maybe_await( + self._client.search( + index=self._index, + body=body, + size=limit, + _source=self._metadata_fields, + ) + ) + for rank, hit in enumerate(_search_hits(response)): + raw_hits.append((hit, query_variant, rank + 1)) + + max_raw_score = max( + (float(hit.get("_score") or 0.0) for hit, _query_variant, _rank in raw_hits), + default=0.0, + ) + candidates: list[ScoredCandidate] = [] + for hit, query_variant, rank in raw_hits: + source = _hit_source(hit) + node_id = _source_value(source, "node_id") or str(hit.get("_id", "")) + if not node_id: + continue + raw_score = float(hit.get("_score") or 0.0) + score = (raw_score / max_raw_score) if max_raw_score > 0 else _rank_score(rank - 1) + candidates.append( + ScoredCandidate( + node_id=node_id, + document_id=_source_value(source, "document_id", "doc_id"), + score=_clamp_score(score), + score_source=CandidateScoreSource.LEXICAL, + rank=rank, + query_variant=query_variant, + provider=self.name, + index_generation=_source_value(source, "index_generation") or self._index, + metadata=_source_metadata(source, raw_score=raw_score), + ) + ) + + return CandidateSearchResult( + candidates=unique_candidates(candidates, limit=limit, prefer_highest_score=True), + provider_counts={self.name: len(candidates)}, + score_ranges=_provider_score_max(candidates), + diagnostics={ + "query_variant_count": len(queries), + "raw_candidate_count": len(candidates), + "filtered": _has_filters(request.filters), + }, + index_generation=self._index, + ) + + async def index_health(self) -> IndexLagReport: + cluster = getattr(self._client, "cluster", None) + health = getattr(cluster, "health", None) if cluster is not None else None + if callable(health): + try: + response = await _maybe_await(health(index=self._index)) + payload = _response_dict(response) + return IndexLagReport( + provider=self.name, + status=str(payload.get("status") or "unknown"), + index_generation=self._index, + properties={"index": self._index}, + ) + except Exception as exc: + return IndexLagReport( + provider=self.name, + status="error", + index_generation=self._index, + properties={"error": type(exc).__name__, "index": self._index}, + ) + return IndexLagReport( + provider=self.name, + status="unknown", + index_generation=self._index, + properties={"index": self._index}, + ) + + def _query_body(self, query: str, filters: IndexFilter, *, limit: int) -> dict[str, object]: + filter_clauses = _opensearch_filter_clauses(filters, self._field_map) + return { + "size": limit, + "query": { + "bool": { + "must": [ + { + "multi_match": { + "query": query, + "fields": self._query_fields, + "type": "best_fields", + } + } + ], + "filter": filter_clauses, + } + }, + } + + +class QdrantCandidateProvider: + """Qdrant vector provider. + + The provider accepts an injected async or sync Qdrant-compatible client and + converts `IndexFilter` into payload filters. It returns scored candidate ids + only; graph/metadata hydration stays in the normal retrieval pipeline. + """ + + __slots__ = ( + "_client", + "_collection", + "_field_map", + "_name", + "_payload_fields", + "_using", + ) + + def __init__( + self, + client: object, + *, + collection: str, + name: str = "qdrant", + payload_fields: Sequence[str] = ( + "node_id", + "document_id", + "doc_id", + "title", + "kind", + "source", + "page", + "chunk_id", + "category", + "domain", + "language", + ), + field_map: dict[str, str] | None = None, + using: str | None = None, + ) -> None: + self._client = client + self._collection = collection + self._name = name + self._payload_fields = list(payload_fields) + self._using = using + self._field_map = { + "node_id": "node_id", + "document_id": "document_id", + "doc_id": "doc_id", + "workspace_id": "workspace_id", + "user_id": "user_id", + "session_id": "session_id", + "domain": "domain", + "acl_policy_id": "acl_policy_id", + "source_id": "source_id", + "source": "source", + "language": "language", + "mime_type": "mime_type", + "tags": "tags", + "created_at": "created_at", + "updated_at": "updated_at", + **(field_map or {}), + } + + @property + def name(self) -> str: + return self._name + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + limit = max(0, int(request.limit)) + if limit == 0: + return CandidateSearchResult(index_generation=self._collection) + if not request.embedding: + return CandidateSearchResult( + diagnostics={"missing_embedding": True}, + index_generation=self._collection, + ) + + payload_filter = _qdrant_filter(request.filters, self._field_map) + response = await self._search(request.embedding, limit=limit, payload_filter=payload_filter) + raw_points = _qdrant_points(response) + max_raw_score = max((_qdrant_score(point) for point in raw_points), default=0.0) + candidates: list[ScoredCandidate] = [] + for rank, point in enumerate(raw_points, start=1): + payload = _qdrant_payload(point) + node_id = _payload_value(payload, "node_id") or _qdrant_id(point) + if not node_id: + continue + raw_score = _qdrant_score(point) + candidates.append( + ScoredCandidate( + node_id=node_id, + document_id=_payload_value(payload, "document_id", "doc_id"), + score=_normalize_vector_score(raw_score, max_raw_score, rank), + score_source=CandidateScoreSource.VECTOR, + rank=rank, + query_variant=request.query, + provider=self.name, + index_generation=_payload_value(payload, "index_generation") + or self._collection, + metadata=_payload_metadata(payload, raw_score=raw_score), + ) + ) + + return CandidateSearchResult( + candidates=unique_candidates(candidates, limit=limit, prefer_highest_score=True), + provider_counts={self.name: len(candidates)}, + score_ranges=_provider_score_max(candidates), + diagnostics={ + "raw_candidate_count": len(candidates), + "filtered": _has_filters(request.filters), + "payload_filter": payload_filter is not None, + }, + index_generation=self._collection, + ) + + async def index_health(self) -> IndexLagReport: + getter = getattr(self._client, "get_collection", None) + if callable(getter): + try: + try: + response = await _maybe_await(getter(collection_name=self._collection)) + except TypeError: + response = await _maybe_await(getter(self._collection)) + payload = _response_dict(response) + points_count = _object_int(response, payload, "points_count") + return IndexLagReport( + provider=self.name, + status=str(_object_str(response, payload, "status") or "ok"), + index_generation=self._collection, + properties={ + "collection": self._collection, + "points_count": points_count, + }, + ) + except Exception as exc: + return IndexLagReport( + provider=self.name, + status="error", + index_generation=self._collection, + properties={"collection": self._collection, "error": type(exc).__name__}, + ) + return IndexLagReport( + provider=self.name, + status="unknown", + index_generation=self._collection, + properties={"collection": self._collection}, + ) + + async def _search( + self, + embedding: list[float], + *, + limit: int, + payload_filter: dict[str, object] | None, + ) -> object: + query_points = getattr(self._client, "query_points", None) + if callable(query_points): + kwargs: dict[str, object] = { + "collection_name": self._collection, + "query": embedding, + "limit": limit, + "with_payload": self._payload_fields, + } + if payload_filter is not None: + kwargs["query_filter"] = payload_filter + if self._using: + kwargs["using"] = self._using + return await _maybe_await(query_points(**kwargs)) + + search = getattr(self._client, "search", None) + if callable(search): + kwargs = { + "collection_name": self._collection, + "query_vector": embedding, + "limit": limit, + "with_payload": self._payload_fields, + } + if payload_filter is not None: + kwargs["query_filter"] = payload_filter + if self._using: + kwargs["vector_name"] = self._using + return await _maybe_await(search(**kwargs)) + + msg = "Qdrant client must expose query_points() or search()." + raise TypeError(msg) + + +class InProcessIndexRouter: + """Simple deterministic router for local providers. + + Production routers can add parallel fan-out, timeouts, and circuit breakers. + This implementation stays small so tests can lock the retrieval contract + before external index services are wired. + """ + + __slots__ = ("_providers",) + + def __init__(self, providers: Sequence[CandidateProvider]) -> None: + self._providers = list(providers) + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + requested = set(request.providers) + all_candidates: list[ScoredCandidate] = [] + provider_counts: dict[str, int] = {} + score_ranges: dict[str, float] = {} + diagnostics: dict[str, object] = { + "provider_count": 0, + "raw_candidate_count": 0, + "deduped_candidate_count": 0, + "failed_providers": [], + } + + for provider in self._providers: + if requested and provider.name not in requested: + continue + try: + result = await provider.search_candidates(request) + except Exception as exc: + failed = diagnostics["failed_providers"] + if isinstance(failed, list): + failed.append({"provider": provider.name, "error": type(exc).__name__}) + continue + diagnostics["provider_count"] = int(diagnostics["provider_count"]) + 1 + all_candidates.extend(result.candidates) + for name, count in result.provider_counts.items(): + provider_counts[name] = provider_counts.get(name, 0) + count + for name, score in result.score_ranges.items(): + score_ranges[name] = max(score_ranges.get(name, 0.0), score) + + diagnostics["raw_candidate_count"] = len(all_candidates) + deduped = unique_candidates( + all_candidates, + limit=max(0, int(request.limit)), + prefer_highest_score=True, + ) + diagnostics["deduped_candidate_count"] = len(deduped) + return CandidateSearchResult( + candidates=deduped, + provider_counts=provider_counts, + score_ranges=score_ranges, + diagnostics=diagnostics, + index_generation="in_process", + ) + + async def index_health(self) -> list[IndexLagReport]: + reports: list[IndexLagReport] = [] + for provider in self._providers: + try: + reports.append(await provider.index_health()) + except Exception as exc: + reports.append( + IndexLagReport( + provider=provider.name, + status="error", + properties={"error": type(exc).__name__}, + ) + ) + return reports + + +def unique_candidates( + candidates: Sequence[ScoredCandidate], + *, + limit: int = 20, + prefer_highest_score: bool = False, +) -> list[ScoredCandidate]: + """Dedupe candidates by node id. + + By default this preserves first-seen order, which lets an upstream router + pass in an already-fused candidate list. Set `prefer_highest_score=True` + when merging unordered provider outputs and the best duplicate score should + win. + """ + + if prefer_highest_score: + best_by_id: dict[str, ScoredCandidate] = {} + first_seen_rank: dict[str, int] = {} + for rank, candidate in enumerate(candidates): + if candidate.node_id not in first_seen_rank: + first_seen_rank[candidate.node_id] = rank + current = best_by_id.get(candidate.node_id) + if current is None or candidate.score > current.score: + best_by_id[candidate.node_id] = candidate + return sorted( + best_by_id.values(), + key=lambda candidate: (-candidate.score, first_seen_rank[candidate.node_id]), + )[:limit] + + seen: set[str] = set() + out: list[ScoredCandidate] = [] + for candidate in candidates: + key = candidate.node_id + if key in seen: + continue + seen.add(key) + out.append(candidate) + if len(out) >= limit: + break + return out + + +def _unique_queries(query: str, variants: Sequence[str]) -> list[str]: + out: list[str] = [] + for item in [query, *variants]: + value = str(item).strip() + if value and value not in out: + out.append(value) + return out + + +def _rank_score(rank: int) -> float: + return max(0.10, 0.95 - max(0, rank) * 0.03) + + +def _node_and_score(row: object, rank: int) -> tuple[object, float]: + if isinstance(row, tuple) and len(row) >= 2: + node = row[0] + try: + score = float(row[1]) + except (TypeError, ValueError): + score = _rank_score(rank) + return node, _clamp_score(score) or _rank_score(rank) + return row, _rank_score(rank) + + +def _clamp_score(score: float) -> float: + return max(0.0, min(1.0, score)) + + +def _has_filters(filters: IndexFilter) -> bool: + return any( + ( + filters.workspace_id, + filters.user_id, + filters.session_id, + filters.domain, + filters.acl_policy_ids, + filters.source_ids, + filters.document_ids, + filters.languages, + filters.mime_types, + filters.tags, + filters.created_after is not None, + filters.created_before is not None, + filters.updated_after is not None, + filters.updated_before is not None, + filters.properties, + ) + ) + + +def _matches_filter(node: object, filters: IndexFilter) -> bool: + if not _has_filters(filters): + return True + props = getattr(node, "properties", {}) or {} + tags = list(getattr(node, "tags", []) or []) + if filters.workspace_id and _prop(props, "workspace_id") != filters.workspace_id: + return False + if filters.user_id and _prop(props, "user_id") != filters.user_id: + return False + if filters.session_id and _prop(props, "session_id") != filters.session_id: + return False + if filters.domain and _prop(props, "domain") != filters.domain: + return False + if filters.acl_policy_ids and not _matches_any( + _prop(props, "acl_policy_id", "acl_policy_ids"), + filters.acl_policy_ids, + ): + return False + if filters.source_ids and not _matches_any( + _prop(props, "source_id", "source", default=str(getattr(node, "source", ""))), + filters.source_ids, + ): + return False + if filters.document_ids and not _matches_any( + _document_id(node) or str(getattr(node, "id", "")), + filters.document_ids, + ): + return False + if filters.languages and not _matches_any(_prop(props, "language", "lang"), filters.languages): + return False + if filters.mime_types and not _matches_any(_prop(props, "mime_type"), filters.mime_types): + return False + if filters.tags and not set(filters.tags).issubset(set(tags)): + return False + created_at = float(getattr(node, "created_at", 0.0) or 0.0) + updated_at = float(getattr(node, "updated_at", 0.0) or 0.0) + if filters.created_after is not None and created_at < filters.created_after: + return False + if filters.created_before is not None and created_at > filters.created_before: + return False + if filters.updated_after is not None and updated_at < filters.updated_after: + return False + if filters.updated_before is not None and updated_at > filters.updated_before: + return False + for key, value in filters.properties.items(): + if _prop(props, key) != value: + return False + return True + + +def _prop(props: dict[str, object], *keys: str, default: str = "") -> str: + for key in keys: + value = props.get(key) + if value is not None: + return str(value) + return default + + +def _matches_any(value: object, allowed: Sequence[str]) -> bool: + allowed_set = {str(item) for item in allowed} + if not allowed_set: + return True + if isinstance(value, (list, tuple, set)): + return any(str(item) in allowed_set for item in value) + raw = str(value) + if raw in allowed_set: + return True + return any(part.strip() in allowed_set for part in raw.split(",")) + + +def _document_id(node: object) -> str: + props = getattr(node, "properties", {}) or {} + return _prop(props, "document_id", "doc_id") + + +def _candidate_metadata(node: object) -> dict[str, object]: + props = getattr(node, "properties", {}) or {} + metadata: dict[str, object] = { + "title": str(getattr(node, "title", "")), + "kind": str(getattr(node, "kind", "")), + } + for key in ("source", "page", "chunk_id", "category", "domain", "language"): + value = props.get(key) + if value is not None: + metadata[key] = value + source = str(getattr(node, "source", "")) + if source and "source" not in metadata: + metadata["source"] = source + return metadata + + +def _provider_score_max(candidates: Sequence[ScoredCandidate]) -> dict[str, float]: + out: dict[str, float] = {} + for candidate in candidates: + provider = candidate.provider or "unknown" + out[provider] = max(out.get(provider, 0.0), candidate.score) + return out + + +async def _maybe_await(value: object) -> object: + if inspect.isawaitable(value): + return await value + return value + + +def _opensearch_filter_clauses( + filters: IndexFilter, + field_map: dict[str, str], +) -> list[dict[str, object]]: + clauses: list[dict[str, object]] = [] + _term_filter(clauses, field_map, "workspace_id", filters.workspace_id) + _term_filter(clauses, field_map, "user_id", filters.user_id) + _term_filter(clauses, field_map, "session_id", filters.session_id) + _term_filter(clauses, field_map, "domain", filters.domain) + _terms_filter(clauses, field_map, "acl_policy_id", filters.acl_policy_ids) + if filters.source_ids: + clauses.append( + { + "bool": { + "should": [ + {"terms": {field_map.get("source_id", "source_id"): filters.source_ids}}, + {"terms": {field_map.get("source", "source"): filters.source_ids}}, + ], + "minimum_should_match": 1, + } + } + ) + if filters.document_ids: + clauses.append( + { + "bool": { + "should": [ + { + "terms": { + field_map.get("document_id", "document_id"): filters.document_ids + } + }, + {"terms": {field_map.get("doc_id", "doc_id"): filters.document_ids}}, + ], + "minimum_should_match": 1, + } + } + ) + _terms_filter(clauses, field_map, "language", filters.languages) + _terms_filter(clauses, field_map, "mime_type", filters.mime_types) + _terms_filter(clauses, field_map, "tags", filters.tags) + _range_filter( + clauses, + field_map.get("created_at", "created_at"), + gte=filters.created_after, + lte=filters.created_before, + ) + _range_filter( + clauses, + field_map.get("updated_at", "updated_at"), + gte=filters.updated_after, + lte=filters.updated_before, + ) + for key, value in filters.properties.items(): + clauses.append({"term": {field_map.get(key, f"properties.{key}"): value}}) + return clauses + + +def _term_filter( + clauses: list[dict[str, object]], + field_map: dict[str, str], + key: str, + value: str, +) -> None: + if value: + clauses.append({"term": {field_map.get(key, key): value}}) + + +def _terms_filter( + clauses: list[dict[str, object]], + field_map: dict[str, str], + key: str, + values: Sequence[str], +) -> None: + if values: + clauses.append({"terms": {field_map.get(key, key): list(values)}}) + + +def _range_filter( + clauses: list[dict[str, object]], + field: str, + *, + gte: float | None, + lte: float | None, +) -> None: + bounds: dict[str, float] = {} + if gte is not None: + bounds["gte"] = gte + if lte is not None: + bounds["lte"] = lte + if bounds: + clauses.append({"range": {field: bounds}}) + + +def _response_dict(response: object) -> dict[str, object]: + if isinstance(response, dict): + return response + body = getattr(response, "body", None) + if isinstance(body, dict): + return body + return {} + + +def _search_hits(response: object) -> list[dict[str, object]]: + payload = _response_dict(response) + hits_obj = payload.get("hits", {}) + if not isinstance(hits_obj, dict): + return [] + raw_hits = hits_obj.get("hits", []) + if not isinstance(raw_hits, list): + return [] + return [hit for hit in raw_hits if isinstance(hit, dict)] + + +def _hit_source(hit: dict[str, object]) -> dict[str, object]: + source = hit.get("_source", {}) + return source if isinstance(source, dict) else {} + + +def _source_value(source: dict[str, object], *keys: str) -> str: + for key in keys: + value = source.get(key) + if value is not None: + return str(value) + return "" + + +def _source_metadata(source: dict[str, object], *, raw_score: float) -> dict[str, object]: + metadata: dict[str, object] = {"raw_score": raw_score} + for key in ( + "title", + "kind", + "source", + "page", + "chunk_id", + "category", + "domain", + "language", + ): + value = source.get(key) + if value is not None: + metadata[key] = value + return metadata + + +def _qdrant_filter( + filters: IndexFilter, + field_map: dict[str, str], +) -> dict[str, object] | None: + must: list[dict[str, object]] = [] + _qdrant_match(must, field_map, "workspace_id", filters.workspace_id) + _qdrant_match(must, field_map, "user_id", filters.user_id) + _qdrant_match(must, field_map, "session_id", filters.session_id) + _qdrant_match(must, field_map, "domain", filters.domain) + _qdrant_match_any(must, field_map, "acl_policy_id", filters.acl_policy_ids) + if filters.source_ids: + must.append( + { + "should": [ + _qdrant_match_condition( + field_map.get("source_id", "source_id"), + filters.source_ids, + ), + _qdrant_match_condition( + field_map.get("source", "source"), + filters.source_ids, + ), + ] + } + ) + if filters.document_ids: + must.append( + { + "should": [ + _qdrant_match_condition( + field_map.get("document_id", "document_id"), + filters.document_ids, + ), + _qdrant_match_condition( + field_map.get("doc_id", "doc_id"), + filters.document_ids, + ), + ] + } + ) + _qdrant_match_any(must, field_map, "language", filters.languages) + _qdrant_match_any(must, field_map, "mime_type", filters.mime_types) + _qdrant_match_any(must, field_map, "tags", filters.tags) + _qdrant_range( + must, + field_map.get("created_at", "created_at"), + gte=filters.created_after, + lte=filters.created_before, + ) + _qdrant_range( + must, + field_map.get("updated_at", "updated_at"), + gte=filters.updated_after, + lte=filters.updated_before, + ) + for key, value in filters.properties.items(): + must.append({"key": field_map.get(key, f"properties.{key}"), "match": {"value": value}}) + return {"must": must} if must else None + + +def _qdrant_match( + must: list[dict[str, object]], + field_map: dict[str, str], + key: str, + value: str, +) -> None: + if value: + must.append({"key": field_map.get(key, key), "match": {"value": value}}) + + +def _qdrant_match_any( + must: list[dict[str, object]], + field_map: dict[str, str], + key: str, + values: Sequence[str], +) -> None: + if values: + must.append(_qdrant_match_condition(field_map.get(key, key), values)) + + +def _qdrant_match_condition(field: str, values: Sequence[str]) -> dict[str, object]: + return {"key": field, "match": {"any": list(values)}} + + +def _qdrant_range( + must: list[dict[str, object]], + field: str, + *, + gte: float | None, + lte: float | None, +) -> None: + bounds: dict[str, float] = {} + if gte is not None: + bounds["gte"] = gte + if lte is not None: + bounds["lte"] = lte + if bounds: + must.append({"key": field, "range": bounds}) + + +def _qdrant_points(response: object) -> list[object]: + if isinstance(response, list): + return response + points = getattr(response, "points", None) + if isinstance(points, list): + return points + payload = _response_dict(response) + result = payload.get("result") + if isinstance(result, dict): + raw_points = result.get("points", []) + return list(raw_points) if isinstance(raw_points, list) else [] + if isinstance(result, list): + return result + return [] + + +def _qdrant_payload(point: object) -> dict[str, object]: + if isinstance(point, dict): + payload = point.get("payload", {}) + else: + payload = getattr(point, "payload", {}) + return payload if isinstance(payload, dict) else {} + + +def _qdrant_id(point: object) -> str: + if isinstance(point, dict): + value = point.get("id", "") + else: + value = getattr(point, "id", "") + return "" if value is None else str(value) + + +def _qdrant_score(point: object) -> float: + if isinstance(point, dict): + value = point.get("score", 0.0) + else: + value = getattr(point, "score", 0.0) + try: + return float(value or 0.0) + except (TypeError, ValueError): + return 0.0 + + +def _payload_value(payload: dict[str, object], *keys: str) -> str: + for key in keys: + value = payload.get(key) + if value is not None: + return str(value) + return "" + + +def _payload_metadata(payload: dict[str, object], *, raw_score: float) -> dict[str, object]: + metadata: dict[str, object] = {"raw_score": raw_score} + for key in ( + "title", + "kind", + "source", + "page", + "chunk_id", + "category", + "domain", + "language", + ): + value = payload.get(key) + if value is not None: + metadata[key] = value + return metadata + + +def _normalize_vector_score(raw_score: float, max_raw_score: float, rank: int) -> float: + if raw_score <= 0: + return _rank_score(rank - 1) + if max_raw_score > 1.0: + return _clamp_score(raw_score / max_raw_score) + return _clamp_score(raw_score) + + +def _object_str(response: object, payload: dict[str, object], key: str) -> str: + value = payload.get(key) + if value is None: + value = getattr(response, key, "") + return "" if value is None else str(value) + + +def _object_int(response: object, payload: dict[str, object], key: str) -> int: + value = payload.get(key) + if value is None: + value = getattr(response, key, 0) + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 diff --git a/tests/test_evidence_search.py b/tests/test_evidence_search.py index 200a859..4a21bda 100644 --- a/tests/test_evidence_search.py +++ b/tests/test_evidence_search.py @@ -18,6 +18,15 @@ _bounded_ppr_seed_scores, ) from synaptic.extensions.graph_expander import ExpansionBudget +from synaptic.indexing import ( + CandidateScoreSource, + CandidateSearchRequest, + CandidateSearchResult, + IndexLagReport, +) +from synaptic.indexing import ( + ScoredCandidate as RoutedCandidate, +) from synaptic.models import ( ConsolidationLevel, Edge, @@ -371,6 +380,64 @@ async def search_fts( await searcher.search("alpha", k=1, query_embedding=[0.1, 0.2]) assert backend.include_embedding_calls[0] is True + async def test_index_router_seeds_bypass_backend_fts(self): + class CountingBackend(MemoryBackend): + def __init__(self) -> None: + super().__init__() + self.fts_calls = 0 + + async def search_fts(self, query: str, *, limit: int = 20) -> list[Node]: + self.fts_calls += 1 + return await super().search_fts(query, limit=limit) + + class FakeRouter: + def __init__(self) -> None: + self.requests: list[CandidateSearchRequest] = [] + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + self.requests.append(request) + return CandidateSearchResult( + candidates=[ + RoutedCandidate( + node_id="alpha", + score=0.77, + score_source=CandidateScoreSource.LEXICAL, + rank=1, + query_variant=request.query, + provider="fake", + ) + ], + provider_counts={"fake": 1}, + score_ranges={"fake": 0.77}, + ) + + async def index_health(self) -> list[IndexLagReport]: + return [IndexLagReport(provider="fake", status="ok")] + + backend = CountingBackend() + await backend.connect() + await backend.save_node( + Node( + id="alpha", + kind=NodeKind.CHUNK, + title="alpha", + content="alpha routed content", + ) + ) + router = FakeRouter() + + searcher = EvidenceSearch(backend=backend, index_router=router, graph_expansion=False) + result = await searcher.search("alpha", k=1) + + assert backend.fts_calls == 0 + assert router.requests[0].query == "alpha" + assert result.evidence[0].node.id == "alpha" + assert result.diagnostics["router_candidate_count"] == 1.0 + assert result.diagnostics["router_provider_fake_count"] == 1.0 + async def test_ppr_seed_scores_are_bounded_by_relevance(self): scores = {f"n{i}": float(i) for i in range(80)} diff --git a/tests/test_graph.py b/tests/test_graph.py index 7fb77e6..ae9c060 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -8,6 +8,13 @@ from synaptic.backends.memory import MemoryBackend from synaptic.extensions.entity_extractor_openie import ChainedEntityExtractor, LLMOpenIEExtractor from synaptic.graph import SynapticGraph, _default_fts_seed_limit +from synaptic.indexing import ( + CandidateScoreSource, + CandidateSearchRequest, + CandidateSearchResult, + IndexLagReport, + ScoredCandidate, +) from synaptic.models import DigestResult, EdgeKind, MaintenanceResult, Node, NodeKind from synaptic.ontology import ( build_agent_ontology, @@ -47,6 +54,35 @@ async def save_nodes_batch(self, nodes): await super().save_nodes_batch(nodes) +class _FakeIndexRouter: + def __init__(self, node_id: str = "router-hit") -> None: + self.node_id = node_id + self.requests: list[CandidateSearchRequest] = [] + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + self.requests.append(request) + return CandidateSearchResult( + candidates=[ + ScoredCandidate( + node_id=self.node_id, + score=0.9, + score_source=CandidateScoreSource.LEXICAL, + rank=1, + query_variant=request.query, + provider="fake_router", + ) + ], + provider_counts={"fake_router": 1}, + score_ranges={"fake_router": 0.9}, + ) + + async def index_health(self) -> list[IndexLagReport]: + return [IndexLagReport(provider="fake_router", status="ok")] + + class TestGraphFullOpenIE: def test_full_keeps_openie_off_by_default_even_with_llm(self) -> None: graph = SynapticGraph.full(MemoryBackend(), llm=_FakeLLM()) @@ -618,6 +654,30 @@ async def test_reuses_evidence_search_for_same_runtime_options(self) -> None: await g.search("retrieval topic", limit=3) assert len(g._evidence_search_cache) == 2 + async def test_index_router_is_wired_into_search_and_cache_key(self) -> None: + backend = MemoryBackend() + await backend.connect() + await backend.save_node( + Node( + id="router-hit", + kind=NodeKind.CHUNK, + title="Router Hit", + content="retrieval topic from router", + ) + ) + router = _FakeIndexRouter() + g = SynapticGraph(backend, index_router=router) + + result = await g.search("retrieval topic", limit=1, rerank=False) + + assert result.nodes[0].node.id == "router-hit" + assert result.diagnostics["router_candidate_count"] == 1.0 + assert router.requests[0].query == "retrieval topic" + assert g._evidence_search_cache + + g.index_router = _FakeIndexRouter("missing") + assert g._evidence_search_cache == {} + async def test_reranker_weight_change_clears_evidence_search_cache(self) -> None: g = await self._graph(_SpyReranker()) await g.search("retrieval topic", limit=3, rerank=False) diff --git a/tests/test_indexing_contracts.py b/tests/test_indexing_contracts.py new file mode 100644 index 0000000..579ce67 --- /dev/null +++ b/tests/test_indexing_contracts.py @@ -0,0 +1,645 @@ +import pytest + +from synaptic import ( + CandidateScoreSource, + CandidateSearchRequest, + CandidateSearchResult, + IndexFilter, + IndexLagReport, + IngestionJob, + IngestionJobStage, + IngestionJobStatus, + InProcessIndexRouter, + Node, + NodeKind, + OpenSearchCandidateProvider, + QdrantCandidateProvider, + ScoredCandidate, + StorageCandidateProvider, + unique_candidates, +) +from synaptic.backends.memory import MemoryBackend + + +def test_index_filter_defaults_are_isolated(): + first = IndexFilter() + second = IndexFilter() + + first.acl_policy_ids.append("acl-a") + first.properties["source"] = "policy" + + assert second.acl_policy_ids == [] + assert second.properties == {} + + +def test_candidate_search_request_keeps_query_variants_and_scope(): + request = CandidateSearchRequest( + query="how much fiber is in carrots", + query_variants=["one cup cooked carrots grams fiber"], + limit=5, + filters=IndexFilter(workspace_id="workspace-a", acl_policy_ids=["acl-a"]), + ) + + assert request.query == "how much fiber is in carrots" + assert request.query_variants == ["one cup cooked carrots grams fiber"] + assert request.limit == 5 + assert request.filters.workspace_id == "workspace-a" + assert request.filters.acl_policy_ids == ["acl-a"] + assert request.scope.workspace_id == "" + + +def test_candidate_search_request_scope_defaults_are_isolated(): + first = CandidateSearchRequest(query="first") + second = CandidateSearchRequest(query="second") + + first.scope.workspace_id = "workspace-a" + + assert second.scope.workspace_id == "" + + +async def _assert_operational_store_roundtrip(backend) -> None: + first = IngestionJob( + id="job-1", + document_id="doc-a", + version="v1", + stage=IngestionJobStage.CHUNK, + status=IngestionJobStatus.RUNNING, + attempt=2, + properties={"workspace_id": "workspace-a"}, + created_at=10.0, + updated_at=11.0, + ) + second = IngestionJob( + id="job-2", + document_id="doc-b", + version="v3", + stage=IngestionJobStage.VECTOR_INDEX, + status=IngestionJobStatus.FAILED, + error="timeout", + created_at=12.0, + updated_at=13.0, + ) + + await backend.save_ingestion_job(first) + await backend.save_ingestion_job(second) + + got = await backend.get_ingestion_job("job-1") + assert got is not None + assert got.document_id == "doc-a" + assert got.stage == IngestionJobStage.CHUNK + assert got.status == IngestionJobStatus.RUNNING + assert got.attempt == 2 + assert got.properties == {"workspace_id": "workspace-a"} + assert [job.id for job in await backend.list_ingestion_jobs(limit=10)] == [ + "job-2", + "job-1", + ] + assert [ + job.id + for job in await backend.list_ingestion_jobs( + stage=IngestionJobStage.VECTOR_INDEX, + status=IngestionJobStatus.FAILED, + limit=10, + ) + ] == ["job-2"] + assert [job.id for job in await backend.list_ingestion_jobs(document_id="doc-a")] == ["job-1"] + + await backend.save_index_lag_report( + IndexLagReport( + provider="qdrant", + status="ok", + index_generation="vec-v1", + pending_documents=2, + pending_chunks=5, + failed_jobs=1, + lag_seconds=3.5, + p95_index_latency_seconds=1.2, + properties={"points_count": 42, "collection": "synaptic-vectors"}, + checked_at=20.0, + ) + ) + await backend.save_index_lag_report( + IndexLagReport( + provider="opensearch", + status="yellow", + index_generation="lex-v2", + checked_at=30.0, + ) + ) + + reports = await backend.list_index_lag_reports(limit=10) + assert [report.provider for report in reports] == ["opensearch", "qdrant"] + qdrant_reports = await backend.list_index_lag_reports(provider="qdrant", since=15.0) + assert len(qdrant_reports) == 1 + assert qdrant_reports[0].pending_chunks == 5 + assert qdrant_reports[0].properties["points_count"] == 42 + + +@pytest.mark.asyncio +async def test_memory_backend_operational_store_roundtrip(): + await _assert_operational_store_roundtrip(MemoryBackend()) + + +@pytest.mark.asyncio +async def test_sqlite_backend_operational_store_roundtrip(tmp_path): + from synaptic.backends.sqlite import SQLiteBackend + + backend = SQLiteBackend(tmp_path / "ops.db") + await backend.connect() + try: + await _assert_operational_store_roundtrip(backend) + finally: + await backend.close() + + +def test_unique_candidates_preserves_first_seen_order(): + candidates = [ + ScoredCandidate(node_id="n1", score=0.9, provider="lexical"), + ScoredCandidate(node_id="n2", score=0.8, provider="vector"), + ScoredCandidate(node_id="n1", score=0.7, provider="graph"), + ] + + deduped = unique_candidates(candidates, limit=10) + + assert [candidate.node_id for candidate in deduped] == ["n1", "n2"] + assert deduped[0].provider == "lexical" + + +def test_unique_candidates_can_prefer_highest_score(): + candidates = [ + ScoredCandidate(node_id="n1", score=0.2, provider="lexical"), + ScoredCandidate(node_id="n2", score=0.8, provider="vector"), + ScoredCandidate(node_id="n1", score=0.9, provider="graph"), + ] + + deduped = unique_candidates(candidates, limit=10, prefer_highest_score=True) + + assert [(candidate.node_id, candidate.provider) for candidate in deduped] == [ + ("n1", "graph"), + ("n2", "vector"), + ] + + +@pytest.mark.asyncio +async def test_candidate_provider_shape_with_fake_provider(): + class FakeProvider: + @property + def name(self) -> str: + return "fake" + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + return CandidateSearchResult( + candidates=[ + ScoredCandidate( + node_id="chunk-a", + document_id="doc-a", + score=0.75, + score_source=CandidateScoreSource.LEXICAL, + rank=1, + query_variant=request.query, + provider=self.name, + index_generation="gen-1", + ) + ], + provider_counts={self.name: 1}, + score_ranges={self.name: 0.75}, + index_generation="gen-1", + ) + + async def index_health(self) -> IndexLagReport: + return IndexLagReport(provider=self.name, status="ok", index_generation="gen-1") + + provider = FakeProvider() + result = await provider.search_candidates(CandidateSearchRequest(query="policy")) + health = await provider.index_health() + + assert result.candidates[0].node_id == "chunk-a" + assert result.candidates[0].score_source == CandidateScoreSource.LEXICAL + assert result.provider_counts == {"fake": 1} + assert health.status == "ok" + + +def test_ingestion_job_defaults_and_status_values(): + job = IngestionJob(document_id="doc-a", version="v1") + + assert job.id + assert job.document_id == "doc-a" + assert job.version == "v1" + assert job.stage == IngestionJobStage.DISCOVER + assert job.status == IngestionJobStatus.PENDING + assert job.created_at <= job.updated_at + + +@pytest.mark.asyncio +async def test_storage_candidate_provider_returns_ids_variants_and_filters(): + backend = MemoryBackend() + await backend.connect() + await backend.save_node( + Node( + id="alpha-w1", + kind=NodeKind.CHUNK, + title="Alpha policy", + content="alpha storage candidate", + tags=["policy"], + properties={"workspace_id": "w1", "doc_id": "doc-alpha"}, + ) + ) + await backend.save_node( + Node( + id="alpha-w2", + kind=NodeKind.CHUNK, + title="Alpha other workspace", + content="alpha storage candidate", + tags=["policy"], + properties={"workspace_id": "w2", "doc_id": "doc-other"}, + ) + ) + await backend.save_node( + Node( + id="beta-w1", + kind=NodeKind.CHUNK, + title="Beta policy", + content="beta storage candidate", + tags=["policy"], + properties={"workspace_id": "w1", "doc_id": "doc-beta"}, + ) + ) + + provider = StorageCandidateProvider(backend) + result = await provider.search_candidates( + CandidateSearchRequest( + query="alpha", + query_variants=["beta"], + limit=5, + filters=IndexFilter(workspace_id="w1", tags=["policy"]), + ) + ) + + assert [candidate.node_id for candidate in result.candidates] == ["alpha-w1", "beta-w1"] + assert [candidate.query_variant for candidate in result.candidates] == ["alpha", "beta"] + assert result.candidates[0].document_id == "doc-alpha" + assert result.provider_counts == {"storage_fts": 2} + assert result.diagnostics["filtered"] is True + + +@pytest.mark.asyncio +async def test_in_process_index_router_dedupes_and_reports_diagnostics(): + class FirstProvider: + @property + def name(self) -> str: + return "first" + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + return CandidateSearchResult( + candidates=[ + ScoredCandidate( + node_id="n1", + score=0.2, + provider=self.name, + score_source=CandidateScoreSource.LEXICAL, + ) + ], + provider_counts={self.name: 1}, + score_ranges={self.name: 0.2}, + ) + + async def index_health(self) -> IndexLagReport: + return IndexLagReport(provider=self.name, status="ok") + + class SecondProvider: + @property + def name(self) -> str: + return "second" + + async def search_candidates( + self, + request: CandidateSearchRequest, + ) -> CandidateSearchResult: + return CandidateSearchResult( + candidates=[ + ScoredCandidate( + node_id="n1", + score=0.9, + provider=self.name, + score_source=CandidateScoreSource.VECTOR, + ), + ScoredCandidate( + node_id="n2", + score=0.8, + provider=self.name, + score_source=CandidateScoreSource.VECTOR, + ), + ], + provider_counts={self.name: 2}, + score_ranges={self.name: 0.9}, + ) + + async def index_health(self) -> IndexLagReport: + return IndexLagReport(provider=self.name, status="ok") + + router = InProcessIndexRouter([FirstProvider(), SecondProvider()]) + result = await router.search_candidates(CandidateSearchRequest(query="q", limit=10)) + + assert [(candidate.node_id, candidate.provider) for candidate in result.candidates] == [ + ("n1", "second"), + ("n2", "second"), + ] + assert result.provider_counts == {"first": 1, "second": 2} + assert result.diagnostics["raw_candidate_count"] == 3 + assert result.diagnostics["deduped_candidate_count"] == 2 + + +@pytest.mark.asyncio +async def test_opensearch_candidate_provider_builds_filtered_queries_and_candidates(): + class FakeOpenSearchClient: + def __init__(self) -> None: + self.calls: list[dict[str, object]] = [] + + async def search(self, **kwargs): + self.calls.append(kwargs) + body = kwargs["body"] + query = body["query"]["bool"]["must"][0]["multi_match"]["query"] + if query == "alpha": + return { + "hits": { + "hits": [ + { + "_id": "fallback-id", + "_score": 4.0, + "_source": { + "node_id": "chunk-alpha", + "document_id": "doc-alpha", + "title": "Alpha", + "kind": "chunk", + "source": "policy", + }, + }, + { + "_id": "chunk-shared", + "_score": 2.0, + "_source": { + "node_id": "chunk-shared", + "doc_id": "doc-shared", + "title": "Shared", + "kind": "chunk", + }, + }, + ] + } + } + return { + "hits": { + "hits": [ + { + "_id": "chunk-shared", + "_score": 8.0, + "_source": { + "node_id": "chunk-shared", + "document_id": "doc-shared", + "title": "Shared Better", + "kind": "chunk", + }, + } + ] + } + } + + client = FakeOpenSearchClient() + provider = OpenSearchCandidateProvider(client, index="synaptic-chunks") + + result = await provider.search_candidates( + CandidateSearchRequest( + query="alpha", + query_variants=["beta"], + limit=5, + filters=IndexFilter( + workspace_id="workspace-a", + acl_policy_ids=["acl-a", "acl-b"], + source_ids=["policy"], + document_ids=["doc-alpha"], + created_after=10.0, + properties={"department": "risk"}, + ), + ) + ) + + assert [candidate.node_id for candidate in result.candidates] == [ + "chunk-shared", + "chunk-alpha", + ] + assert result.candidates[0].score == 1.0 + assert result.candidates[0].query_variant == "beta" + assert result.candidates[1].score == 0.5 + assert result.candidates[1].document_id == "doc-alpha" + assert result.candidates[1].metadata["source"] == "policy" + assert result.provider_counts == {"opensearch": 3} + assert result.diagnostics["query_variant_count"] == 2 + + first_body = client.calls[0]["body"] + filters = first_body["query"]["bool"]["filter"] + assert {"term": {"workspace_id": "workspace-a"}} in filters + assert {"terms": {"acl_policy_id": ["acl-a", "acl-b"]}} in filters + assert { + "bool": { + "should": [ + {"terms": {"source_id": ["policy"]}}, + {"terms": {"source": ["policy"]}}, + ], + "minimum_should_match": 1, + } + } in filters + assert {"range": {"created_at": {"gte": 10.0}}} in filters + assert {"term": {"properties.department": "risk"}} in filters + assert { + "bool": { + "should": [ + {"terms": {"document_id": ["doc-alpha"]}}, + {"terms": {"doc_id": ["doc-alpha"]}}, + ], + "minimum_should_match": 1, + } + } in filters + + +@pytest.mark.asyncio +async def test_opensearch_candidate_provider_health_uses_cluster_status(): + class _Cluster: + async def health(self, *, index: str): + return {"status": "green", "index": index} + + class FakeOpenSearchClient: + cluster = _Cluster() + + provider = OpenSearchCandidateProvider(FakeOpenSearchClient(), index="synaptic-chunks") + + report = await provider.index_health() + + assert report.provider == "opensearch" + assert report.status == "green" + assert report.index_generation == "synaptic-chunks" + + +@pytest.mark.asyncio +async def test_qdrant_candidate_provider_builds_payload_filters_and_candidates(): + class _Point: + def __init__( + self, + *, + point_id: str, + score: float, + payload: dict[str, object], + ) -> None: + self.id = point_id + self.score = score + self.payload = payload + + class _Response: + def __init__(self, points: list[_Point]) -> None: + self.points = points + + class FakeQdrantClient: + def __init__(self) -> None: + self.calls: list[dict[str, object]] = [] + + async def query_points(self, **kwargs): + self.calls.append(kwargs) + return _Response( + [ + _Point( + point_id="fallback-a", + score=0.42, + payload={ + "node_id": "chunk-alpha", + "document_id": "doc-alpha", + "title": "Alpha", + "kind": "chunk", + "source": "policy", + }, + ), + _Point( + point_id="fallback-b", + score=0.88, + payload={ + "node_id": "chunk-beta", + "doc_id": "doc-beta", + "title": "Beta", + "kind": "chunk", + }, + ), + _Point( + point_id="fallback-a", + score=0.96, + payload={ + "node_id": "chunk-alpha", + "document_id": "doc-alpha", + "title": "Alpha Better", + "kind": "chunk", + }, + ), + ] + ) + + client = FakeQdrantClient() + provider = QdrantCandidateProvider(client, collection="synaptic-vectors") + + result = await provider.search_candidates( + CandidateSearchRequest( + query="alpha", + embedding=[0.1, 0.2, 0.3], + limit=5, + filters=IndexFilter( + workspace_id="workspace-a", + acl_policy_ids=["acl-a"], + source_ids=["policy"], + document_ids=["doc-alpha"], + updated_before=20.0, + properties={"department": "risk"}, + ), + ) + ) + + assert [candidate.node_id for candidate in result.candidates] == [ + "chunk-alpha", + "chunk-beta", + ] + assert result.candidates[0].score == 0.96 + assert result.candidates[0].score_source == CandidateScoreSource.VECTOR + assert result.candidates[0].document_id == "doc-alpha" + assert result.candidates[1].document_id == "doc-beta" + assert result.provider_counts == {"qdrant": 3} + assert result.diagnostics["payload_filter"] is True + + call = client.calls[0] + assert call["collection_name"] == "synaptic-vectors" + assert call["query"] == [0.1, 0.2, 0.3] + assert call["with_payload"] == [ + "node_id", + "document_id", + "doc_id", + "title", + "kind", + "source", + "page", + "chunk_id", + "category", + "domain", + "language", + ] + payload_filter = call["query_filter"] + assert {"key": "workspace_id", "match": {"value": "workspace-a"}} in payload_filter["must"] + assert {"key": "acl_policy_id", "match": {"any": ["acl-a"]}} in payload_filter["must"] + assert {"key": "updated_at", "range": {"lte": 20.0}} in payload_filter["must"] + assert {"key": "properties.department", "match": {"value": "risk"}} in payload_filter["must"] + assert { + "should": [ + {"key": "source_id", "match": {"any": ["policy"]}}, + {"key": "source", "match": {"any": ["policy"]}}, + ] + } in payload_filter["must"] + assert { + "should": [ + {"key": "document_id", "match": {"any": ["doc-alpha"]}}, + {"key": "doc_id", "match": {"any": ["doc-alpha"]}}, + ] + } in payload_filter["must"] + + +@pytest.mark.asyncio +async def test_qdrant_candidate_provider_requires_embedding(): + class FakeQdrantClient: + async def query_points(self, **kwargs): # pragma: no cover - should not be called + raise AssertionError(kwargs) + + provider = QdrantCandidateProvider(FakeQdrantClient(), collection="synaptic-vectors") + + result = await provider.search_candidates(CandidateSearchRequest(query="alpha")) + + assert result.candidates == [] + assert result.diagnostics["missing_embedding"] is True + + +@pytest.mark.asyncio +async def test_qdrant_candidate_provider_health_uses_collection_status(): + class _Collection: + status = "green" + points_count = 42 + + class FakeQdrantClient: + async def get_collection(self, *, collection_name: str): + assert collection_name == "synaptic-vectors" + return _Collection() + + provider = QdrantCandidateProvider(FakeQdrantClient(), collection="synaptic-vectors") + + report = await provider.index_health() + + assert report.provider == "qdrant" + assert report.status == "green" + assert report.index_generation == "synaptic-vectors" + assert report.properties["points_count"] == 42