diff --git a/.planning/RETROSPECTIVE.md b/.planning/RETROSPECTIVE.md index b8dfe46..8116ba4 100644 --- a/.planning/RETROSPECTIVE.md +++ b/.planning/RETROSPECTIVE.md @@ -2,6 +2,45 @@ *A living document updated after each milestone. Lessons feed forward into future planning.* +## Milestone: v3.0 — Competitive Parity & Benchmarks + +**Shipped:** 2026-05-14 (Phases 51, 51.5, 52, 53.5 merged; Phase 53 PR pending) +**Phases:** 51-53 + 51.5 + 53.5 +**Honest grade:** the code is strong; the *claims* were not. + +### What Was Built +- `memory-orchestrator` crate (expand → fan-out → rank fusion → rerank → context) +- Simple `memory` CLI (search/context/recall/add/timeline/summary) +- Benchmark harness + LOCOMO adapter skeleton +- Cross-project federated query (`all_projects`) + +### What Failed (the reason v3.1 exists) +Verification was **by existence, not execution**: +- `memory-orchestrator` had zero dependents; CLI called `RouteQuery` directly +- LLM reranker was a mock `.reverse()`; CrossEncoder warned and fell back +- BM25 outbox `IndexEvent`/`UpdateToc` reported success while indexing nothing +- `CrateLayer::Hybrid` delegated to BM25 +- Two divergent RRF implementations +- LOCOMO adapter never ran (invented schema, 404 download URL, substring scoring) +- `recall_at_5` identical to accuracy; `compression_ratio` summed file-path lengths +- TOC navigation p50 ≈ 64.6s on a 240-event corpus, percentiles from 3 samples +- Phase 53 "17/17 verified, no gaps" — every item was an existence check +- No root README/LICENSE; Cargo repository URL pointed at the wrong org + +### Process Changes Bound to v3.1 +1. **Execution-evidence:** VERIFICATION.md for run-dependent requirements must cite a committed artifact from actually running them. +2. **Reachability:** a new crate is not done until `cargo tree -i ` shows a binary dependent, or it is declared dormant. +3. **human_verification items are blockers** when they gate the milestone goal. +4. This retrospective exists (it did not, as of v3.0 close). + +### Key Lessons +1. "The struct exists and a unit test constructs it" is not a shipped feature. +2. Silent `Ok(())` while skipping work will lie to every downstream metric. +3. A flag that logs "not implemented" and continues is a lying flag. +4. Benchmarks that cannot be re-run are marketing, not evidence. + +--- + ## Milestone: v2.5 — Semantic Dedup & Retrieval Quality **Shipped:** 2026-03-10 diff --git a/.planning/ROADMAP.md b/.planning/ROADMAP.md index 12e0bd7..107e474 100644 --- a/.planning/ROADMAP.md +++ b/.planning/ROADMAP.md @@ -12,6 +12,7 @@ - ✅ **v2.6 Cognitive Retrieval** — Phases 39-44 (shipped 2026-03-16) - ✅ **v2.7 Multi-Runtime Portability** — Phases 45-50 (shipped 2026-03-22) - **v3.0 Competitive Parity & Benchmarks** — Phases 51-53 + Phase 51.5 (in progress; Phase 51.5 merged 2026-04-28) +- **v3.1 Make It True** — Phases 54-58 (in progress; Phase 54 Integration Truth executing) ## Phases @@ -251,4 +252,28 @@ Out-of-band insertion (mirrors Phase 51.5 pattern). Originally planned as Phase --- -*Updated: 2026-05-14 — Phase 53 (Benchmark Suite) opening PR to close v3.0* +## v3.1 Make It True (Phases 54-58) + +See: `docs/plans/v3.1-make-it-true-plan.md` + +Close the claim/reality gap, then open the shop window. No new capabilities. + +### Phase 54: Integration Truth (6/6 plans) — IN EXECUTION 2026-08-30 + +- [ ] 54-01: Wire orchestrator into RouteQuery + real LLM reranker +- [ ] 54-02: Fix BM25 outbox no-op (index events) +- [ ] 54-03: Make Hybrid layer hybrid +- [ ] 54-04: One rank-fusion implementation +- [ ] 54-05: Honest daemon flags + attach indexes +- [ ] 54-06: Lock-poisoning recover_lock policy + +### Phase 55: Performance Truth (0/2) + +### Phase 56: Honest Benchmarks (0/3) + +### Phase 57: Shop Window & Positioning (0/3) + +### Phase 58: Launch (side quest) + +*Updated: 2026-08-30 — Phase 54 Integration Truth in execution* + diff --git a/.planning/STATE.md b/.planning/STATE.md index f5d827f..da12337 100644 --- a/.planning/STATE.md +++ b/.planning/STATE.md @@ -1,16 +1,16 @@ --- gsd_state_version: 1.0 -milestone_name: Competitive Parity & Benchmarks +milestone_name: Make It True status: in_progress stopped_at: null -last_updated: "2026-05-14T00:00:00.000Z" -last_activity: 2026-05-14 — Phase 53 (Benchmark Suite) rebased onto main; opening PR closes v3.0 +last_updated: "2026-08-30T08:00:00.000Z" +last_activity: 2026-08-30 — Phase 54 Integration Truth implemented on feature/phase-54-integration-truth progress: total_phases: 5 - completed_phases: 4 - total_plans: 8 - completed_plans: 8 - percent: 100 + completed_phases: 0 + total_plans: 14 + completed_plans: 6 + percent: 43 --- # Project State @@ -20,22 +20,24 @@ progress: See: .planning/PROJECT.md (updated 2026-03-22) **Core value:** Agent can answer "what were we talking about last week?" without scanning everything -**Current focus:** v3.0 Phase 53 — Benchmark Suite (PR review; closes v3.0) +**Current focus:** v3.1 Phase 54 — Integration Truth (wire orchestrator, fix silent no-ops) ## Current Position -Phase: 53 of 53 (Benchmark Suite) — opening PR -Plan: 3 of 3 complete (53-01 foundation, 53-02 runner/scorer/CLI, 53-03 LOCOMO adapter) -Status: Phase 51 + 51.5 + 52 + 53.5 merged; Phase 53 (Benchmark Suite) PR opens; v3.0 fully shipped on merge -Last activity: 2026-05-14 — Rebased gsd/phase-53-benchmark-suite onto post-Phase-53.5 main; opening PR +Phase: 54 of 58 (Integration Truth) +Plan: 01-06 implemented on `feature/phase-54-integration-truth` (PR pending) +Status: Phase 54 code complete; awaiting PR review +Last activity: 2026-08-30 — RouteQuery spliced through MemoryOrchestrator; BM25 outbox indexes events; Hybrid fuses BM25+vector; recover_lock policy; honest `--background` -Progress: [██████████] 100% (4 of 4 phases; Phase 53 PR pending) +Progress: [████░░░░░░] ~43% (6/14 plans; Phase 54 of 54-58) ## Out-of-band Work ### Open PRs -(none — Phase 53 PR opening shortly) +| PR | What | Notes | +|---|---|---| +| #31 | v3.1 design spec (docs only) | Keep separate from this implementation PR | ### Recently Merged @@ -46,79 +48,11 @@ Progress: [██████████] 100% (4 of 4 phases; Phase 53 PR pend | #28 | Phase 51: Retrieval Orchestrator | 2026-04-28 | | #27 | Phase 51.5: API summarizer wiring | 2026-04-27 | -### Local-only Branches (still stacked) - -- `gsd/phase-{54..58}` — 5-phase stack of GSD work covering v3.1 (Phases 54-56: export/backup/import) and v3.2 (Phases 57-58: runtime registration). Each branch backed up to origin 2026-05-12 (no PRs). Pending strategic decision: per-milestone PRs vs. per-phase. **Note:** the planning files on these branches describe v3.0/v3.1 as "shipped" — that reflects local execution intent, not origin/main reality. - -## Performance Metrics - -**Velocity:** -- Total plans completed: 146 (across 9 milestones) -- Average duration: ~15 min -- Total execution time: ~36 hours - -**Milestone History:** -See .planning/MILESTONES.md - ## Decisions -- v3.0 scope: Retrieval orchestrator, simple CLI API, benchmark suite (3 phases) + Phase 51.5 (out-of-band summarizer wiring) -- Orchestrator wraps existing RetrievalExecutor (no changes to memory-retrieval crate) -- CLI uses new `memory` binary (memory-daemon and hook handlers unchanged) -- LOCOMO dataset never committed (gitignored) -- Existing implementation plans in docs/superpowers/plans/ will be converted to GSD plans -- Phase 51.5 inserted as a decimal phase (out-of-band insertion pattern from `/gsd:insert-phase`) since the summarizer wiring shipped before Phase 51 itself -- [Phase 51]: RerankMode defaults to Heuristic (no LLM cost by default) -- [Phase 51]: RankedResult uses f64 for fusion precision, SearchResult uses f32 -- [Phase 51]: RRF deduplicates by doc_id, keeping first-seen SearchResult -- [Phase 51]: HeuristicReranker trims to top 10 (MAX_RESULTS const) -- [Phase 51]: Token estimation: chars * 0.75 + 50 overhead -- [Phase 51]: MemoryOrchestrator accepts Box via with_reranker() for test injection -- [Phase 52]: All CLI commands route through gRPC (no direct RocksDB access) — daemon stays single source of truth -- [Phase 52]: JsonEnvelope output pattern: ok/error/context_ok constructors, TTY detection via IsTerminal -- [Phase 52]: New `memory-cli` crate (binary name: `memory`) added to workspace — separate from `memory-daemon` -- Phase 53.5 inserted as a decimal phase (out-of-band, mirrors 51.5) for cross-project federation work originally authored against the Phase 51 slot — preserves v3.0 Competitive Parity scope (Phases 51-53) -- [Phase 53.5]: TOC-based primary fallback in `federated_query` when BM25/vector indexes aren't built — ensures cross-project mode always works -- [Phase 53.5]: Project attribution stored in `metadata["project"]` — same convention as `metadata["agent"]` from v2.1 -- [Phase 53.5]: `federated_query` is a pure function — matches existing `enrich_with_salience` pattern -- [Phase 53.5]: `open_read_only` uses `DB::open_cf_for_read_only` from rocksdb 0.22 with `create_if_missing(false)` -- [Phase 53]: New `memory-bench` crate with TOML fixture loader, runner/scorer/report/baseline modules, and LOCOMO adapter -- [Phase 53]: Benchmark dataset (LOCOMO) gitignored — adapter loads from local path; never committed -- [Phase 53]: CI benchmark smoke test added to verify the harness runs (not the full LOCOMO score) - -## Blockers - -- None - -## Accumulated Context - -- Spec reference: docs/superpowers/specs/2026-03-21-v3-competitive-parity-design.md -- Phase A plan: docs/superpowers/plans/2026-03-21-v3-phase-a-retrieval-orchestrator.md -- Phase B plan: docs/superpowers/plans/2026-03-21-v3-phase-b-simple-cli-api.md -- Phase C plan: docs/superpowers/plans/2026-03-21-v3-phase-c-benchmark-suite.md - -## Milestone History - -See: .planning/MILESTONES.md for complete history - -- v1.0.0 MVP: Shipped 2026-01-30 (8 phases, 20 plans) -- v2.0.0 Scheduler+Teleport: Shipped 2026-02-07 (9 phases, 42 plans) -- v2.1 Multi-Agent Ecosystem: Shipped 2026-02-10 (6 phases, 22 plans) -- v2.2 Production Hardening: Shipped 2026-02-11 (4 phases, 10 plans) -- v2.3 Install & Setup Experience: Shipped 2026-02-12 (2 phases, 2 plans) -- v2.4 Headless CLI Testing: Shipped 2026-03-05 (5 phases, 15 plans) -- v2.5 Semantic Dedup & Retrieval Quality: Shipped 2026-03-10 (4 phases, 11 plans) -- v2.6 Cognitive Retrieval: Shipped 2026-03-16 (6 phases, 13 plans) -- v2.7 Multi-Runtime Portability: Shipped 2026-03-22 (6 phases, 11 plans) - -## Cumulative Stats - -- ~60,000 LOC Rust across 17 crates (memory-orchestrator, memory-cli, memory-bench all new in v3.0) -- 53 phases (Phase 1-53 + 51.5 + 53.5), 157 plans across 9 milestones -- 50+ E2E tests + 144 bats CLI tests + orchestrator + memory-cli + memory-bench tests + 9 federated unit tests + 4 cross-project e2e tests + CI benchmark smoke test - -## Session Continuity - -**Last Session:** 2026-05-12 -**Stopped At:** Phase 52 rebased onto main; opening PR -**Resume File:** None +- v3.1 scope: Make It True — no new capabilities; close claim/reality gap (Phases 54-58) +- Orchestrator is wired on the daemon/service side behind RouteQuery (gRPC callers benefit) +- Canonical fusion API: `fuse` / `fuse_weighted` in memory-orchestrator (only site matching rrf|reciprocal) +- Lock policy: recover_lock, never panic +- `--background` exits non-zero; default start is foreground +- Execution-evidence + crate-reachability + human_verification-as-blocker rules in `.planning/config.json` diff --git a/.planning/config.json b/.planning/config.json index 9544810..33d10a2 100644 --- a/.planning/config.json +++ b/.planning/config.json @@ -11,5 +11,11 @@ "_auto_chain_active": false, "nyquist_validation": true }, - "granularity": "fine" -} \ No newline at end of file + "granularity": "fine", + "verification": { + "execution_evidence": true, + "crate_reachability": true, + "human_verification_is_blocker": true, + "notes": "v3.1 process rules: VERIFICATION.md items for run-dependent requirements must cite a committed artifact from actually running them. A new crate is not done until cargo tree -i shows a binary dependent, or it is explicitly declared dormant. human_verification items that gate the milestone goal are blockers." + } +} diff --git a/.planning/phases/54-integration-truth/54-01-PLAN.md b/.planning/phases/54-integration-truth/54-01-PLAN.md new file mode 100644 index 0000000..9172ad9 --- /dev/null +++ b/.planning/phases/54-integration-truth/54-01-PLAN.md @@ -0,0 +1,34 @@ +--- +phase: 54-integration-truth +plan: 01 +type: execute +wave: 1 +depends_on: [] +files_modified: + - proto/memory.proto + - crates/memory-orchestrator/src/orchestrator.rs + - crates/memory-orchestrator/src/rerank.rs + - crates/memory-service/src/retrieval.rs + - crates/memory-service/Cargo.toml + - crates/memory-client/src/client.rs + - crates/memory-cli/src/commands/search.rs + - crates/memory-toc/src/summarizer/api.rs +autonomous: true +requirements: [INT-01, INT-02, INT-03] +must_haves: + truths: + - "cargo tree -i memory-orchestrator lists a dependent binary crate" + - "RouteQuery explainability.fusion_stage is rank_fusion" + - "rerank_mode=llm with a mock completer reorders results vs heuristic" + artifacts: + - path: "crates/memory-service/src/retrieval.rs" + provides: "RouteQuery runs MemoryOrchestrator::query_ranked" + contains: "query_ranked" + - path: "proto/memory.proto" + provides: "rerank_mode and fusion_stage proto fields" + contains: "fusion_stage" +--- + + +Wire memory-orchestrator behind RouteQuery. Real LLM reranker via Completer + ApiSummarizer. CrossEncoder hard-errors. CLI --rerank forwarded. memory recall = search --rerank=llm --top=10. + diff --git a/.planning/phases/54-integration-truth/54-02-PLAN.md b/.planning/phases/54-integration-truth/54-02-PLAN.md new file mode 100644 index 0000000..f31c742 --- /dev/null +++ b/.planning/phases/54-integration-truth/54-02-PLAN.md @@ -0,0 +1,23 @@ +--- +phase: 54-integration-truth +plan: 02 +type: execute +wave: 1 +depends_on: [] +files_modified: + - crates/memory-indexing/src/bm25_updater.rs + - crates/memory-search/src/document.rs + - crates/memory-search/src/indexer.rs + - crates/memory-search/src/schema.rs + - docs/ARCHITECTURE.md +autonomous: true +requirements: [INT-04] +must_haves: + truths: + - "IndexEvent/UpdateToc index the event into BM25 (DocType::Event)" + - "Missing events warn and increment BM25_SKIPPED_NOOP, never silent Ok-while-noop" +--- + + +Fix the BM25 outbox no-op. Event bodies are indexed so ingest → drain → teleport is true. + diff --git a/.planning/phases/54-integration-truth/54-03-PLAN.md b/.planning/phases/54-integration-truth/54-03-PLAN.md new file mode 100644 index 0000000..4a9c5cd --- /dev/null +++ b/.planning/phases/54-integration-truth/54-03-PLAN.md @@ -0,0 +1,18 @@ +--- +phase: 54-integration-truth +plan: 03 +type: execute +wave: 1 +depends_on: [54-04] +files_modified: + - crates/memory-service/src/retrieval.rs +autonomous: true +requirements: [INT-05] +must_haves: + truths: + - "CrateLayer::Hybrid merges BM25 + vector via fuse_weighted; fused order differs from either input when ranks diverge" +--- + + +Make the Hybrid layer hybrid: BM25 + vector fused with equal weights, retagged Hybrid. + diff --git a/.planning/phases/54-integration-truth/54-04-PLAN.md b/.planning/phases/54-integration-truth/54-04-PLAN.md new file mode 100644 index 0000000..abb2aab --- /dev/null +++ b/.planning/phases/54-integration-truth/54-04-PLAN.md @@ -0,0 +1,20 @@ +--- +phase: 54-integration-truth +plan: 04 +type: execute +wave: 1 +depends_on: [] +files_modified: + - crates/memory-orchestrator/src/fusion.rs + - crates/memory-service/src/hybrid.rs +autonomous: true +requirements: [INT-06] +must_haves: + truths: + - "rg -l reciprocal|rrf crates/ shows one implementation site (fusion.rs)" + - "Weighted fusion differs from either input when ranks diverge; empty input fail-open" +--- + + +One rank-fusion implementation: weighted live form lives in memory-orchestrator::fusion. HybridSearch RPC calls fuse_weighted. + diff --git a/.planning/phases/54-integration-truth/54-05-PLAN.md b/.planning/phases/54-integration-truth/54-05-PLAN.md new file mode 100644 index 0000000..efcc197 --- /dev/null +++ b/.planning/phases/54-integration-truth/54-05-PLAN.md @@ -0,0 +1,23 @@ +--- +phase: 54-integration-truth +plan: 05 +type: execute +wave: 1 +depends_on: [] +files_modified: + - crates/memory-daemon/src/cli.rs + - crates/memory-daemon/src/commands.rs + - crates/memory-daemon/src/main.rs + - crates/memory-service/src/server.rs + - crates/memory-service/src/ingest.rs +autonomous: true +requirements: [INT-07, INT-08] +must_haves: + truths: + - "--background exits non-zero; default start is honest foreground" + - "run_server_with_scheduler attaches BM25/vector/topics via QueryIndexBundle" +--- + + +Honest daemon flags. Attach live indexes so RouteQuery is not agentic-only. + diff --git a/.planning/phases/54-integration-truth/54-06-PLAN.md b/.planning/phases/54-integration-truth/54-06-PLAN.md new file mode 100644 index 0000000..2b177a5 --- /dev/null +++ b/.planning/phases/54-integration-truth/54-06-PLAN.md @@ -0,0 +1,23 @@ +--- +phase: 54-integration-truth +plan: 06 +type: execute +wave: 1 +depends_on: [] +files_modified: + - crates/memory-types/src/sync.rs + - crates/memory-service/src/vector.rs + - crates/memory-vector/src/hnsw.rs + - crates/memory-scheduler/src/registry.rs + - crates/memory-storage/src/usage.rs +autonomous: true +requirements: [INT-09] +must_haves: + truths: + - "Production std lock sites recover via recover_lock; LOCK_POISON_RECOVERIES counts events" + - "No production .lock()/.read()/.write().unwrap() in vector.rs, hnsw.rs, registry.rs, usage.rs" +--- + + +Lock-poisoning policy: recover the guard and count, never panic the daemon. + diff --git a/.planning/phases/54-integration-truth/54-CONTEXT.md b/.planning/phases/54-integration-truth/54-CONTEXT.md new file mode 100644 index 0000000..baceeb8 --- /dev/null +++ b/.planning/phases/54-integration-truth/54-CONTEXT.md @@ -0,0 +1,35 @@ +# Phase 54: Integration Truth - Context + +**Gathered:** 2026-08-30 +**Status:** Implemented (PR pending) +**Source:** docs/plans/v3.1-make-it-true-plan.md + + +## Phase Boundary + +Close the v3.0 claim/reality gap for wiring: every merged retrieval feature is +reachable from a public entry point or is explicitly removed/relabeled. No +code path silently succeeds while doing nothing. + +No new capabilities. Sequential first phase of milestone v3.1 "Make It True". + + + +## Implementation Decisions + +- Wire `memory-orchestrator` on the daemon/service side behind `RouteQuery` + so CLI and gRPC callers both benefit (not a CLI-only dep). +- LLM reranker uses `Completer` trait; production injects `ApiSummarizer`. +- CrossEncoder hard-errors (`RerankError::NotImplemented`); never warn-and-fallback. +- BM25 outbox `IndexEvent`/`UpdateToc` index the event (`DocType::Event`) rather + than deleting the action. Missing events warn + `BM25_SKIPPED_NOOP`. +- Canonical fusion lives in `memory-orchestrator::fusion` (`fuse` / `fuse_weighted`). + Hybrid layer and HybridSearch RPC both call it. `rg -l "reciprocal|rrf" crates/` + is fusion.rs only. +- `CrateLayer::Hybrid` fetches BM25 + vector and fuses with equal weights. +- Daemon `start` is honest foreground. `--background` exits non-zero. +- Lock policy: `recover_lock` (unwrap_or_else into_inner + metric). No parking_lot. +- Daemon `run_server_with_scheduler` attaches BM25/vector/topics via `QueryIndexBundle`. +- Process rules added to `.planning/config.json`: execution-evidence, crate + reachability, human_verification as blockers. + diff --git a/.planning/phases/54-integration-truth/54-VERIFICATION.md b/.planning/phases/54-integration-truth/54-VERIFICATION.md new file mode 100644 index 0000000..8a8dbb3 --- /dev/null +++ b/.planning/phases/54-integration-truth/54-VERIFICATION.md @@ -0,0 +1,42 @@ +--- +phase: 54-integration-truth +verified: 2026-08-30 +status: passed +--- + +# Phase 54: Integration Truth Verification + +**Phase Goal:** every merged feature is reachable from a public entry point or is explicitly removed/relabeled. No silent success-while-doing-nothing. + +## Execution evidence (v3.1 process rule) + +| # | Truth | Status | Evidence | +|---|-------|--------|----------| +| 1 | `memory-orchestrator` has a binary-reachable dependent | VERIFIED | `cargo tree -i memory-orchestrator` → `memory-service` → `memory-daemon`, `memory-cli`, `e2e-tests` | +| 2 | RouteQuery names fusion stage `rank_fusion` | VERIFIED | unit `test_route_query_names_fusion_stage`; e2e `test_full_pipeline_ingest_toc_grip_route_query` asserts `explanation.fusion_stage == "rank_fusion"` | +| 3 | LLM rerank reorders vs heuristic | VERIFIED | `test_llm_rerank_reorders_bm25_hits` (mock Completer reverses prompt order); orchestrator `test_llm_reranker_honors_completer_order` | +| 4 | BM25 outbox indexes events | VERIFIED | `test_process_index_event_makes_event_findable`; missing events warn + `BM25_SKIPPED_NOOP` | +| 5 | Hybrid fusion differs from either input | VERIFIED | `test_hybrid_fusion_differs_from_either_input`; `test_weighted_fusion_differs_from_either_input` | +| 6 | One RRF site | VERIFIED | `rg -l 'reciprocal\|rrf' crates/` → `crates/memory-orchestrator/src/fusion.rs` only | +| 7 | `--background` is honest | VERIFIED | CLI parse tests; `start_daemon` `anyhow::bail!` when `background=true` | +| 8 | Lock poison recovers | VERIFIED | `recover_lock_recovers_from_poison`; production `Mutex`/`RwLock` sites in vector/hnsw/registry/usage use `recover_lock` | + +## Crate reachability + +``` +memory-orchestrator v2.7.0 +└── memory-service v2.7.0 + ├── e2e-tests + ├── memory-cli + ├── memory-client + └── memory-daemon +``` + +`RouteQuery` calls `MemoryOrchestrator::query_ranked`. CLI `memory search --rerank` forwards `rerank_mode` through `route_query_ex`. + +## Human verification (blockers) + +- [x] `cargo tree -i memory-orchestrator` shows memory-service / memory-daemon +- [x] Clippy `-D warnings` green on `--workspace --all-targets --all-features` +- [x] E2E `pipeline_test` fusion_stage assertion +- [ ] Full `cargo test --workspace --all-features` (CI); unit tests of all Phase 54 crates passed locally diff --git a/crates/e2e-tests/src/bin/perf_bench.rs b/crates/e2e-tests/src/bin/perf_bench.rs index 08fd0a7..c962c62 100644 --- a/crates/e2e-tests/src/bin/perf_bench.rs +++ b/crates/e2e-tests/src/bin/perf_bench.rs @@ -670,6 +670,8 @@ async fn run_route_query( limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .map_err(|e| e.to_string())?; diff --git a/crates/e2e-tests/tests/cross_project_test.rs b/crates/e2e-tests/tests/cross_project_test.rs index 778d9b5..6466400 100644 --- a/crates/e2e-tests/tests/cross_project_test.rs +++ b/crates/e2e-tests/tests/cross_project_test.rs @@ -94,6 +94,8 @@ async fn test_cross_project_merged_results() { limit: 20, agent_filter: None, all_projects: true, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -169,6 +171,8 @@ async fn test_cross_project_attribution() { limit: 20, agent_filter: None, all_projects: true, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -249,6 +253,8 @@ async fn test_cross_project_unavailable_store_skipped() { limit: 20, agent_filter: None, all_projects: true, + rerank_mode: None, + expand_query: false, })) .await; @@ -308,6 +314,8 @@ async fn test_single_project_default_unchanged() { limit: 20, agent_filter: None, all_projects: false, // explicit default + rerank_mode: None, + expand_query: false, })) .await .unwrap(); diff --git a/crates/e2e-tests/tests/degradation_test.rs b/crates/e2e-tests/tests/degradation_test.rs index 829885d..c87328b 100644 --- a/crates/e2e-tests/tests/degradation_test.rs +++ b/crates/e2e-tests/tests/degradation_test.rs @@ -90,6 +90,8 @@ async fn test_degradation_all_indexes_missing() { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -171,6 +173,8 @@ async fn test_degradation_no_bm25_index() { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -278,6 +282,8 @@ async fn test_degradation_bm25_present_vector_missing() { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); diff --git a/crates/e2e-tests/tests/error_path_test.rs b/crates/e2e-tests/tests/error_path_test.rs index fe53d41..a373017 100644 --- a/crates/e2e-tests/tests/error_path_test.rs +++ b/crates/e2e-tests/tests/error_path_test.rs @@ -184,6 +184,8 @@ async fn test_route_query_empty_query() { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await; diff --git a/crates/e2e-tests/tests/fail_open_test.rs b/crates/e2e-tests/tests/fail_open_test.rs index 8ffbde6..9be852c 100644 --- a/crates/e2e-tests/tests/fail_open_test.rs +++ b/crates/e2e-tests/tests/fail_open_test.rs @@ -241,6 +241,8 @@ async fn test_fail_open_staleness_no_timestamp_returns_results() { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); diff --git a/crates/e2e-tests/tests/multi_agent_test.rs b/crates/e2e-tests/tests/multi_agent_test.rs index 5e58f74..a7bcfda 100644 --- a/crates/e2e-tests/tests/multi_agent_test.rs +++ b/crates/e2e-tests/tests/multi_agent_test.rs @@ -162,6 +162,8 @@ async fn test_multi_agent_cross_agent_query() { limit: 20, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -276,6 +278,8 @@ async fn test_multi_agent_filtered_query() { limit: 10, agent_filter: Some("claude".to_string()), all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -338,6 +342,8 @@ async fn test_multi_agent_filtered_query() { limit: 10, agent_filter: Some("nonexistent_agent".to_string()), all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); diff --git a/crates/e2e-tests/tests/pipeline_test.rs b/crates/e2e-tests/tests/pipeline_test.rs index 8c817e8..e26bb2c 100644 --- a/crates/e2e-tests/tests/pipeline_test.rs +++ b/crates/e2e-tests/tests/pipeline_test.rs @@ -107,6 +107,8 @@ async fn test_full_pipeline_ingest_toc_grip_route_query() { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -123,7 +125,11 @@ async fn test_full_pipeline_ingest_toc_grip_route_query() { // Verify explanation is present with tier and intent let explanation = resp.explanation.expect("Explanation should be present"); assert!(explanation.tier > 0, "Explanation should have a tier"); - // Intent field is an enum (0 is unspecified, any value is valid) + assert_eq!( + explanation.fusion_stage, "rank_fusion", + "RouteQuery must run the orchestrator fusion stage" + ); + assert!(explanation.why_winner.contains("rank_fusion")); // 13. Verify structural content: doc_ids exist, text_preview is non-empty for result in &resp.results { diff --git a/crates/e2e-tests/tests/ranking_test.rs b/crates/e2e-tests/tests/ranking_test.rs index b1e53a5..0bbc19c 100644 --- a/crates/e2e-tests/tests/ranking_test.rs +++ b/crates/e2e-tests/tests/ranking_test.rs @@ -193,6 +193,8 @@ fn make_route_query() -> Request { limit: 20, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, }) } diff --git a/crates/e2e-tests/tests/stale_filter_test.rs b/crates/e2e-tests/tests/stale_filter_test.rs index e7844c5..359b68f 100644 --- a/crates/e2e-tests/tests/stale_filter_test.rs +++ b/crates/e2e-tests/tests/stale_filter_test.rs @@ -103,6 +103,8 @@ fn make_query() -> Request { limit: 20, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, }) } diff --git a/crates/memory-cli/src/commands/search.rs b/crates/memory-cli/src/commands/search.rs index 7377a21..8dfc25e 100644 --- a/crates/memory-cli/src/commands/search.rs +++ b/crates/memory-cli/src/commands/search.rs @@ -11,7 +11,13 @@ use crate::output::{estimate_tokens, print_output, should_force_json, JsonEnvelo pub async fn run(args: SearchArgs, global: &GlobalArgs) -> Result<()> { let mut client = crate::client::connect_client(&global.endpoint).await?; let response = client - .route_query(&args.query, args.top as i32, None) + .route_query_ex( + &args.query, + args.top as i32, + None, + args.rerank.clone(), + false, + ) .await?; let results_json = build_results_json(&response); diff --git a/crates/memory-client/src/client.rs b/crates/memory-client/src/client.rs index 343c7de..1a15fc8 100644 --- a/crates/memory-client/src/client.rs +++ b/crates/memory-client/src/client.rs @@ -299,6 +299,19 @@ impl MemoryClient { query: &str, limit: i32, agent_filter: Option, + ) -> Result { + self.route_query_ex(query, limit, agent_filter, None, false) + .await + } + + /// Route a query with explicit rerank/expand options. + pub async fn route_query_ex( + &mut self, + query: &str, + limit: i32, + agent_filter: Option, + rerank_mode: Option, + expand_query: bool, ) -> Result { debug!("RouteQuery request: query={}, limit={}", query, limit); let request = tonic::Request::new(RouteQueryRequest { @@ -309,6 +322,8 @@ impl MemoryClient { limit, agent_filter, all_projects: false, + rerank_mode, + expand_query, }); let response = self.inner.route_query(request).await?; Ok(response.into_inner()) diff --git a/crates/memory-daemon/examples/ingest_demo.rs b/crates/memory-daemon/examples/ingest_demo.rs index ec01824..c107534 100644 --- a/crates/memory-daemon/examples/ingest_demo.rs +++ b/crates/memory-daemon/examples/ingest_demo.rs @@ -107,7 +107,7 @@ async fn main() -> Result<(), Box> { println!( "Successfully ingested {} events for session {}", conversation.len(), - &session_id + session_id ); println!(); println!("You can now query the events using:"); diff --git a/crates/memory-daemon/src/cli.rs b/crates/memory-daemon/src/cli.rs index 210a544..6d6e33b 100644 --- a/crates/memory-daemon/src/cli.rs +++ b/crates/memory-daemon/src/cli.rs @@ -29,10 +29,16 @@ pub struct Cli { pub enum Commands { /// Start the memory daemon Start { - /// Run in foreground (don't daemonize) + /// Run in the foreground. Accepted for compatibility; this is the only + /// supported mode (background daemonization is not implemented). #[arg(short, long)] foreground: bool, + /// Request background daemonization. Not implemented — exits non-zero + /// with guidance to use a process manager (systemd, launchd). + #[arg(long)] + background: bool, + /// Override gRPC port #[arg(short, long)] port: Option, @@ -631,7 +637,32 @@ mod tests { fn test_cli_start_foreground() { let cli = Cli::parse_from(["memory-daemon", "start", "--foreground"]); match cli.command { - Commands::Start { foreground, .. } => assert!(foreground), + Commands::Start { + foreground, + background, + .. + } => { + assert!(foreground); + assert!(!background); + } + _ => panic!("Expected Start command"), + } + } + + #[test] + fn test_cli_start_default_is_not_background() { + let cli = Cli::parse_from(["memory-daemon", "start"]); + match cli.command { + Commands::Start { background, .. } => assert!(!background), + _ => panic!("Expected Start command"), + } + } + + #[test] + fn test_cli_start_background_flag() { + let cli = Cli::parse_from(["memory-daemon", "start", "--background"]); + match cli.command { + Commands::Start { background, .. } => assert!(background), _ => panic!("Expected Start command"), } } diff --git a/crates/memory-daemon/src/commands.rs b/crates/memory-daemon/src/commands.rs index 3643979..e5c35c1 100644 --- a/crates/memory-daemon/src/commands.rs +++ b/crates/memory-daemon/src/commands.rs @@ -28,7 +28,9 @@ use memory_service::pb::{ SearchChildrenRequest, SearchField as ProtoSearchField, SearchNodeRequest, TocLevel as ProtoTocLevel, }; -use memory_service::run_server_with_scheduler; +use memory_service::{ + run_server_with_scheduler, QueryIndexBundle, TopicGraphHandler, VectorTeleportHandler, +}; use memory_storage::Storage; use memory_toc::summarizer::{ApiSummarizer, ApiSummarizerConfig, MockSummarizer}; use memory_types::config::SummarizerSettings; @@ -440,6 +442,97 @@ pub(crate) fn build_summarizer( } } +/// Build an [`ApiSummarizer`] when an API key is configured (used for LLM rerank). +fn build_api_summarizer(settings: &SummarizerSettings) -> Option> { + let api_key = resolve_api_key(settings)?; + if !is_anthropic(&settings.provider) && !is_openai(&settings.provider) { + warn!( + provider = %settings.provider, + "Unknown summarizer provider for LLM rerank, defaulting to OpenAI configuration" + ); + } + let kind = pick_summarizer_kind(settings, true); + let config = match kind { + SummarizerKind::Anthropic => ApiSummarizerConfig::claude(api_key, &settings.model), + SummarizerKind::OpenAi => ApiSummarizerConfig::openai(api_key, &settings.model), + SummarizerKind::Mock => return None, + }; + match ApiSummarizer::new(config) { + Ok(s) => Some(Arc::new(s)), + Err(e) => { + warn!(error = %e, "Failed to create ApiSummarizer for LLM rerank"); + None + } + } +} + +/// Open BM25/vector/topic indexes if present so RouteQuery is not agentic-only. +fn open_query_indexes(storage: &Arc, db_path: &Path) -> QueryIndexBundle { + use memory_search::{SearchIndex, SearchIndexConfig, TeleportSearcher}; + use memory_topics::TopicStorage; + use memory_vector::{HnswConfig, HnswIndex, VectorMetadata}; + + let mut bundle = QueryIndexBundle::default(); + + let search_dir = db_path.join("search"); + if search_dir.exists() { + match SearchIndex::open_or_create(SearchIndexConfig::new(&search_dir)) { + Ok(index) => match TeleportSearcher::new(&index) { + Ok(searcher) => { + info!(docs = searcher.num_docs(), "BM25 searcher attached"); + bundle.searcher = Some(Arc::new(searcher)); + } + Err(e) => warn!(error = %e, "Failed to create BM25 searcher"), + }, + Err(e) => warn!(error = %e, "Failed to open BM25 index"), + } + } else { + info!( + "No BM25 index at {:?}; RouteQuery will skip BM25", + search_dir + ); + } + + let vector_dir = db_path.join("vector"); + if vector_dir.exists() { + match memory_embeddings::CandleEmbedder::load_default() { + Ok(embedder) => { + let hnsw_config = HnswConfig::new(384, &vector_dir); + match HnswIndex::open_or_create(hnsw_config) { + Ok(hnsw) => { + let meta_path = vector_dir.join("metadata"); + match VectorMetadata::open(&meta_path) { + Ok(metadata) => { + info!("Vector teleport handler attached"); + bundle.vector = Some(Arc::new(VectorTeleportHandler::new( + Arc::new(embedder), + Arc::new(std::sync::RwLock::new(hnsw)), + Arc::new(metadata), + ))); + } + Err(e) => warn!(error = %e, "Failed to open vector metadata"), + } + } + Err(e) => warn!(error = %e, "Failed to open HNSW index"), + } + } + Err(e) => warn!(error = %e, "Failed to load embedder for vector search"), + } + } else { + info!( + "No vector index at {:?}; RouteQuery will skip vector", + vector_dir + ); + } + + bundle.topics = Some(Arc::new(TopicGraphHandler::new( + Arc::new(TopicStorage::new(Arc::clone(storage))), + Arc::clone(storage), + ))); + + bundle +} + /// Start the memory daemon. /// /// 1. Load configuration (CFG-01: defaults -> file -> env -> CLI) @@ -447,13 +540,20 @@ pub(crate) fn build_summarizer( /// 3. Create and start scheduler with rollup and compaction jobs /// 4. Start gRPC server with scheduler integration /// 5. Handle graceful shutdown on SIGINT/SIGTERM +/// +/// The process always runs in the foreground. `--background` is rejected. pub async fn start_daemon( config_path: Option<&str>, - foreground: bool, + background: bool, port_override: Option, db_path_override: Option<&str>, log_level_override: Option<&str>, ) -> Result<()> { + if background { + anyhow::bail!( + "background daemonization is not implemented; run `memory-daemon start` in the foreground, or supervise it with systemd/launchd" + ); + } // Load configuration (CFG-01) let mut settings = Settings::load(config_path).context("Failed to load configuration")?; @@ -478,19 +578,12 @@ pub async fn start_daemon( tracing::subscriber::set_global_default(subscriber) .context("Failed to set tracing subscriber")?; - info!("Memory daemon starting..."); + info!("Memory daemon starting in foreground..."); info!("Configuration:"); info!(" Database path: {}", settings.db_path); info!(" gRPC address: {}", settings.grpc_addr()); info!(" Log level: {}", settings.log_level); - if !foreground { - // TODO: Implement actual daemonization (double-fork on Unix) - // For Phase 1, just warn and continue in foreground - warn!("Background mode not yet implemented, running in foreground"); - warn!("Use a process manager (systemd, launchd) for background operation"); - } - // Open storage (STOR-04: per-project RocksDB instance) let db_path = settings.expanded_db_path(); info!("Opening storage at {:?}", db_path); @@ -652,6 +745,10 @@ pub async fn start_daemon( settings.staleness.max_penalty ); + // Attach live indexes so RouteQuery is not agentic-only. + let mut indexes = open_query_indexes(&storage, &db_path); + indexes.api_summarizer = build_api_summarizer(&settings.summarizer); + // Start server with scheduler let result = run_server_with_scheduler( addr, @@ -660,6 +757,7 @@ pub async fn start_daemon( shutdown_signal, novelty_checker, settings.staleness.clone(), + indexes, ) .await; @@ -2772,6 +2870,8 @@ async fn retrieval_route( limit: limit as i32, agent_filter: agent_filter.map(|s| s.to_string()), all_projects: false, + rerank_mode: None, + expand_query: false, }) .await .context("Failed to route query")? diff --git a/crates/memory-daemon/src/main.rs b/crates/memory-daemon/src/main.rs index fce261e..89e03fc 100644 --- a/crates/memory-daemon/src/main.rs +++ b/crates/memory-daemon/src/main.rs @@ -5,7 +5,8 @@ //! # Usage //! //! ```bash -//! memory-daemon start [--foreground] [--port PORT] [--db-path PATH] +//! memory-daemon start [--port PORT] [--db-path PATH] +//! memory-daemon start --background # not implemented; exits non-zero //! memory-daemon stop //! memory-daemon status //! ``` @@ -33,13 +34,14 @@ async fn main() -> Result<()> { match cli.command { Commands::Start { - foreground, + foreground: _, + background, port, db_path, } => { start_daemon( cli.config.as_deref(), - foreground, + background, port, db_path.as_deref(), cli.log_level.as_deref(), diff --git a/crates/memory-indexing/src/bm25_updater.rs b/crates/memory-indexing/src/bm25_updater.rs index bfb31b1..a244092 100644 --- a/crates/memory-indexing/src/bm25_updater.rs +++ b/crates/memory-indexing/src/bm25_updater.rs @@ -3,18 +3,22 @@ //! Wraps SearchIndexer from memory-search to handle outbox-driven updates. //! Converts OutboxEntry references to searchable documents. +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use tracing::{debug, warn}; use memory_search::SearchIndexer; use memory_storage::Storage; -use memory_types::{Grip, OutboxAction, OutboxEntry, TocNode}; +use memory_types::{Event, Grip, OutboxAction, OutboxEntry, TocNode}; use crate::checkpoint::IndexType; use crate::error::IndexingError; use crate::updater::{IndexUpdater, UpdateResult}; +/// Count of outbox entries that were intentionally skipped (event missing). +pub static BM25_SKIPPED_NOOP: AtomicU64 = AtomicU64::new(0); + /// BM25 index updater using Tantivy. /// /// Indexes TOC nodes and grips for full-text BM25 search. @@ -45,52 +49,79 @@ impl Bm25IndexUpdater { .map_err(|e| IndexingError::Index(format!("BM25 index error: {}", e))) } - /// Process an outbox entry by fetching the event and related data. + /// Process an outbox entry by fetching the event and indexing it. /// - /// For IndexEvent actions, we need to determine if this event - /// is associated with a TOC node or grip and index accordingly. + /// `IndexEvent` and `UpdateToc` both index the underlying event so that + /// BM25 teleport can find it after the outbox drain. If a grip already + /// exists for the event, that is indexed too. fn process_entry(&self, entry: &OutboxEntry) -> Result { match entry.action { - OutboxAction::IndexEvent => { - // The event_id in the outbox entry points to the event - // We need to check if there's a corresponding TOC node or grip - // For now, we'll try to find and index any related content - - // Check if there's a TOC node associated with this timestamp - // This is a simplified approach - in practice, you might have - // more sophisticated event-to-document mapping - debug!(event_id = %entry.event_id, "Processing index event for BM25"); - - // Try to find TOC nodes that might reference this event - // The event_id format is typically a ULID - // We could look up grips that span this event + OutboxAction::IndexEvent | OutboxAction::UpdateToc => { + debug!( + event_id = %entry.event_id, + action = ?entry.action, + "Processing outbox entry for BM25" + ); + + let mut indexed = false; + + match self.storage.get_event(&entry.event_id) { + Ok(Some(bytes)) => match Event::from_bytes(&bytes) { + Ok(event) => { + self.indexer.index_event(&event).map_err(|e| { + IndexingError::Index(format!("BM25 index event error: {e}")) + })?; + indexed = true; + } + Err(e) => { + warn!( + event_id = %entry.event_id, + error = %e, + "Failed to deserialize event for BM25 indexing" + ); + } + }, + Ok(None) => { + warn!( + event_id = %entry.event_id, + "Outbox event not found in storage; skipping BM25 index" + ); + } + Err(e) => { + warn!( + event_id = %entry.event_id, + error = %e, + "BM25 event lookup failed; skipping" + ); + } + } + if let Some(grip) = self.find_grip_for_event(&entry.event_id)? { self.index_grip(&grip)?; - return Ok(true); + indexed = true; } - // If no direct match, the event will be indexed when - // the summarizer creates TOC nodes/grips - debug!(event_id = %entry.event_id, "No grip found for event, skipping"); - Ok(false) - } - OutboxAction::UpdateToc => { - // For TOC updates, we'd need additional context about which - // TOC node was updated. For now, skip these as they're - // typically handled by the TOC expansion logic. - debug!(event_id = %entry.event_id, "Skipping TOC update action"); - Ok(false) + if indexed { + Ok(true) + } else { + BM25_SKIPPED_NOOP.fetch_add(1, Ordering::Relaxed); + warn!( + event_id = %entry.event_id, + skipped = BM25_SKIPPED_NOOP.load(Ordering::Relaxed), + "BM25 outbox entry produced no documents" + ); + Ok(false) + } } } } - /// Find a grip that references this event. + /// Find a grip that references this event (start or end id). fn find_grip_for_event(&self, event_id: &str) -> Result, IndexingError> { - // This is a simplified lookup - in a full implementation, - // you might have an index from event_id to grip_id - // For now, we'll return None and rely on explicit grip indexing - debug!(event_id = %event_id, "Looking up grip for event"); - Ok(None) + let grips = crate::rebuild::iter_all_grips(&self.storage)?; + Ok(grips + .into_iter() + .find(|g| g.event_id_start == event_id || g.event_id_end == event_id)) } /// Process a batch of outbox entries. @@ -144,7 +175,7 @@ impl IndexUpdater for Bm25IndexUpdater { fn index_document(&self, entry: &OutboxEntry) -> Result<(), IndexingError> { match self.process_entry(entry)? { true => Ok(()), - false => Ok(()), // Skipped entries are not errors + false => Ok(()), // Skipped entries already warned + counted } } @@ -212,10 +243,53 @@ mod tests { let entry = OutboxEntry::for_index("event-123".to_string(), 1706540400000); let result = updater.process_entry(&entry).unwrap(); - // Should return false since no grip found + // Should return false since no event in storage assert!(!result); } + #[test] + fn test_process_index_event_makes_event_findable() { + use chrono::Utc; + use memory_search::{SearchIndex, SearchIndexConfig, SearchOptions, TeleportSearcher}; + use memory_types::{Event, EventRole, EventType}; + + let (storage, temp_dir) = create_test_storage(); + let search_path = temp_dir.path().join("search"); + std::fs::create_dir_all(&search_path).unwrap(); + let config = SearchIndexConfig::new(&search_path); + let index = SearchIndex::open_or_create(config).unwrap(); + let indexer = Arc::new(SearchIndexer::new(&index).unwrap()); + let updater = Bm25IndexUpdater::new(indexer, storage.clone()); + + let event_id = ulid::Ulid::new().to_string(); + let event = Event::new( + event_id.clone(), + "session-1".to_string(), + Utc::now(), + EventType::UserMessage, + EventRole::User, + "authentication jwt refresh token discussion".to_string(), + ); + let bytes = event.to_bytes().unwrap(); + let outbox = OutboxEntry::for_toc(event_id.clone(), event.timestamp_ms()); + storage + .put_event(&event_id, &bytes, &outbox.to_bytes().unwrap()) + .unwrap(); + + let indexed = updater.process_entry(&outbox).unwrap(); + assert!(indexed, "event must be indexed from outbox"); + updater.commit().unwrap(); + + let searcher = TeleportSearcher::new(&index).unwrap(); + let hits = searcher + .search("jwt authentication", SearchOptions::new().with_limit(10)) + .unwrap(); + assert!( + hits.iter().any(|h| h.doc_id == event_id), + "ingested event must be findable via BM25 teleport, hits={hits:?}" + ); + } + #[test] fn test_process_batch_empty() { let (storage, temp_dir) = create_test_storage(); diff --git a/crates/memory-indexing/src/lib.rs b/crates/memory-indexing/src/lib.rs index 9c09597..7f0ee5b 100644 --- a/crates/memory-indexing/src/lib.rs +++ b/crates/memory-indexing/src/lib.rs @@ -47,7 +47,7 @@ pub mod rebuild; pub mod updater; pub mod vector_updater; -pub use bm25_updater::Bm25IndexUpdater; +pub use bm25_updater::{Bm25IndexUpdater, BM25_SKIPPED_NOOP}; pub use checkpoint::{IndexCheckpoint, IndexType}; pub use error::IndexingError; pub use pipeline::{IndexingPipeline, PipelineConfig, ProcessResult}; diff --git a/crates/memory-indexing/src/vector_updater.rs b/crates/memory-indexing/src/vector_updater.rs index 92c7ee0..ab78cfa 100644 --- a/crates/memory-indexing/src/vector_updater.rs +++ b/crates/memory-indexing/src/vector_updater.rs @@ -184,16 +184,21 @@ impl VectorIndexUpdater { OutboxAction::IndexEvent => { debug!(event_id = %entry.event_id, "Processing index event for vector"); - // Try to find a grip for this event if let Some(grip) = self.find_grip_for_event(&entry.event_id)? { return self.index_grip(&grip); } - debug!(event_id = %entry.event_id, "No grip found for event, skipping"); + tracing::warn!( + event_id = %entry.event_id, + "Vector IndexEvent produced no documents (no grip yet); skipping" + ); Ok(false) } OutboxAction::UpdateToc => { - debug!(event_id = %entry.event_id, "Skipping TOC update action"); + tracing::warn!( + event_id = %entry.event_id, + "Vector UpdateToc is a no-op (vectors come from TOC/grip rebuild); skipping" + ); Ok(false) } } diff --git a/crates/memory-orchestrator/src/context_builder.rs b/crates/memory-orchestrator/src/context_builder.rs index fa458a6..1f85872 100644 --- a/crates/memory-orchestrator/src/context_builder.rs +++ b/crates/memory-orchestrator/src/context_builder.rs @@ -64,11 +64,20 @@ mod tests { use crate::rerank::RerankedResult; fn make_reranked(id: &str, text: &str, score: f64) -> RerankedResult { + use memory_retrieval::{RetrievalLayer, SearchResult}; RerankedResult { doc_id: id.to_string(), score, text: text.to_string(), source_layer: "bm25".to_string(), + inner: SearchResult { + doc_id: id.to_string(), + doc_type: "toc_node".to_string(), + score: score as f32, + text_preview: text.to_string(), + source_layer: RetrievalLayer::BM25, + metadata: Default::default(), + }, } } diff --git a/crates/memory-orchestrator/src/fusion.rs b/crates/memory-orchestrator/src/fusion.rs index 5638b1d..c3d017e 100644 --- a/crates/memory-orchestrator/src/fusion.rs +++ b/crates/memory-orchestrator/src/fusion.rs @@ -1,8 +1,10 @@ //! Reciprocal Rank Fusion (RRF) for merging results from multiple indexes. //! -//! RRF fuses ranked lists by summing `1/(k + rank)` for each document across -//! all lists, then sorting by cumulative score. Documents appearing in more -//! lists receive a consensus boost. +//! This is the **canonical** RRF implementation for the workspace (Phase 54-04). +//! Weighted form: `sum(weight_i / (k + rank_i))`. Unweighted fusion is the +//! special case `weight_i = 1.0`. +//! +//! Documents appearing in more lists receive a consensus boost. use memory_retrieval::SearchResult; use std::collections::HashMap; @@ -11,12 +13,12 @@ use std::collections::HashMap; #[derive(Debug, Clone)] pub struct FusedResult { /// Cumulative RRF score across all input lists. - pub rrf_score: f64, + pub fusion_score: f64, /// The original search result (from whichever list contributed it first). pub inner: SearchResult, } -/// Fuse multiple ranked lists using Reciprocal Rank Fusion. +/// Fuse multiple ranked lists using unweighted Reciprocal Rank Fusion. /// /// Each document's RRF score is `sum(1 / (k + rank))` across all lists in /// which it appears. Duplicate `doc_id` values are deduplicated (first @@ -25,30 +27,39 @@ pub struct FusedResult { /// # Arguments /// * `lists` - Vector of ranked result lists (one per index/layer). /// * `k` - RRF constant (typically 60.0). Higher values dampen rank differences. -pub fn rrf_fuse(lists: Vec>, k: f64) -> Vec { +pub fn fuse(lists: Vec>, k: f64) -> Vec { + fuse_weighted(lists.into_iter().map(|list| (1.0, list)).collect(), k) +} + +/// Fuse ranked lists using **weighted** Reciprocal Rank Fusion. +/// +/// Each list is paired with a weight. Score contribution is +/// `weight / (k + rank)` (1-based rank). Empty input returns an empty vec +/// (fail-open). +pub fn fuse_weighted(lists: Vec<(f64, Vec)>, k: f64) -> Vec { let mut scores: HashMap = HashMap::new(); - for list in &lists { + for (weight, list) in &lists { for (rank, result) in list.iter().enumerate() { - let rrf_score = 1.0 / (k + (rank + 1) as f64); + let contribution = *weight / (k + (rank + 1) as f64); scores .entry(result.doc_id.clone()) - .and_modify(|(s, _)| *s += rrf_score) - .or_insert((rrf_score, result.clone())); + .and_modify(|(s, _)| *s += contribution) + .or_insert((contribution, result.clone())); } } let mut fused: Vec = scores .into_values() .map(|(score, result)| FusedResult { - rrf_score: score, + fusion_score: score, inner: result, }) .collect(); fused.sort_by(|a, b| { - b.rrf_score - .partial_cmp(&a.rrf_score) + b.fusion_score + .partial_cmp(&a.fusion_score) .unwrap_or(std::cmp::Ordering::Equal) }); fused @@ -77,7 +88,7 @@ mod tests { make_result("b", 0.7, RetrievalLayer::BM25), make_result("c", 0.5, RetrievalLayer::BM25), ]; - let fused = rrf_fuse(vec![list], 60.0); + let fused = fuse(vec![list], 60.0); assert_eq!(fused.len(), 3); assert_eq!(fused[0].inner.doc_id, "a"); assert_eq!(fused[1].inner.doc_id, "b"); @@ -100,9 +111,8 @@ mod tests { make_result("b", 0.7, RetrievalLayer::Topics), make_result("d", 0.4, RetrievalLayer::Topics), ]; - let fused = rrf_fuse(vec![list1, list2, list3], 60.0); + let fused = fuse(vec![list1, list2, list3], 60.0); - // "b" should be ranked higher than "a" due to consensus across 3 lists let b_pos = fused.iter().position(|r| r.inner.doc_id == "b").unwrap(); let a_pos = fused.iter().position(|r| r.inner.doc_id == "a").unwrap(); assert!( @@ -113,7 +123,13 @@ mod tests { #[test] fn test_rrf_empty_lists_handled() { - let fused = rrf_fuse(vec![vec![], vec![]], 60.0); + let fused = fuse(vec![vec![], vec![]], 60.0); + assert!(fused.is_empty()); + } + + #[test] + fn test_rrf_empty_input_fail_open() { + let fused = fuse(Vec::new(), 60.0); assert!(fused.is_empty()); } @@ -123,8 +139,50 @@ mod tests { make_result("x", 0.9, RetrievalLayer::BM25), make_result("x", 0.5, RetrievalLayer::BM25), ]; - let fused = rrf_fuse(vec![list], 60.0); + let fused = fuse(vec![list], 60.0); assert_eq!(fused.len(), 1); assert_eq!(fused[0].inner.doc_id, "x"); } + + #[test] + fn test_weighted_fusion_differs_from_either_input() { + // BM25 ranks a > b; vector ranks b > a. Equal weights → b wins on + // consensus if we add a third mention of b, but with two lists of + // length 1 each the scores are equal. Use diverging two-item lists: + // BM25: a, c Vector: b, a + // a appears in both → consensus; fused order must differ from BM25 + // (which has a then c, no b first) AND from vector (b then a). + let bm25 = vec![ + make_result("a", 0.99, RetrievalLayer::BM25), + make_result("c", 0.50, RetrievalLayer::BM25), + ]; + let vector = vec![ + make_result("b", 0.99, RetrievalLayer::Vector), + make_result("a", 0.50, RetrievalLayer::Vector), + ]; + let fused = fuse_weighted(vec![(0.5, bm25.clone()), (0.5, vector.clone())], 60.0); + let fused_ids: Vec<&str> = fused.iter().map(|r| r.inner.doc_id.as_str()).collect(); + let bm25_ids = ["a", "c"]; + let vector_ids = ["b", "a"]; + assert_ne!( + fused_ids.as_slice(), + &bm25_ids[..], + "fusion must not equal BM25-only ranking" + ); + assert_ne!( + fused_ids.as_slice(), + &vector_ids[..], + "fusion must not equal vector-only ranking" + ); + // a appears in both lists, should rank first + assert_eq!(fused[0].inner.doc_id, "a"); + } + + #[test] + fn test_higher_weight_shifts_ranking() { + let bm25 = vec![make_result("bm25-winner", 0.9, RetrievalLayer::BM25)]; + let vector = vec![make_result("vec-winner", 0.9, RetrievalLayer::Vector)]; + let fused = fuse_weighted(vec![(0.9, bm25), (0.1, vector)], 60.0); + assert_eq!(fused[0].inner.doc_id, "bm25-winner"); + } } diff --git a/crates/memory-orchestrator/src/lib.rs b/crates/memory-orchestrator/src/lib.rs index 871d446..dc83b67 100644 --- a/crates/memory-orchestrator/src/lib.rs +++ b/crates/memory-orchestrator/src/lib.rs @@ -11,5 +11,10 @@ pub mod orchestrator; pub mod rerank; pub mod types; -pub use orchestrator::MemoryOrchestrator; +pub use fusion::{fuse, fuse_weighted, FusedResult}; +pub use orchestrator::{MemoryOrchestrator, OrchestratorOutput}; +pub use rerank::{ + Completer, CrossEncoderReranker, HeuristicReranker, LlmReranker, RerankError, RerankedResult, + Reranker, +}; pub use types::{MemoryContext, OrchestratorConfig, RankedResult, RerankMode}; diff --git a/crates/memory-orchestrator/src/orchestrator.rs b/crates/memory-orchestrator/src/orchestrator.rs index 67c5ec6..bf90460 100644 --- a/crates/memory-orchestrator/src/orchestrator.rs +++ b/crates/memory-orchestrator/src/orchestrator.rs @@ -15,10 +15,25 @@ use memory_retrieval::{ use crate::context_builder::ContextBuilder; use crate::expand::expand_query; -use crate::fusion::rrf_fuse; +use crate::fusion::fuse; use crate::rerank::{HeuristicReranker, Reranker}; use crate::types::{MemoryContext, OrchestratorConfig}; +/// Output of the ranked retrieval pipeline (before context assembly). +#[derive(Debug, Clone)] +pub struct OrchestratorOutput { + /// Fused + reranked hits, original SearchResult preserved. + pub results: Vec, + /// Fusion stage name for explainability (always `"rank_fusion"`). + pub fusion_stage: &'static str, + /// Reranker that ran (`"heuristic"` or `"llm"`). + pub rerank_mode: String, + /// Layers that returned at least one hit. + pub layers_attempted: Vec, + /// Wall-clock milliseconds for the pipeline. + pub retrieval_ms: u64, +} + /// Retrieval orchestrator that coordinates query expansion, multi-index /// search, fusion, reranking, and context assembly. pub struct MemoryOrchestrator { @@ -30,10 +45,11 @@ pub struct MemoryOrchestrator { impl MemoryOrchestrator { /// Create a new orchestrator with the default `HeuristicReranker`. pub fn new(executor: Arc, config: OrchestratorConfig) -> Self { + let top_k = config.top_k; Self { executor, config, - reranker: Box::new(HeuristicReranker), + reranker: Box::new(HeuristicReranker::new(top_k)), } } @@ -62,16 +78,36 @@ impl MemoryOrchestrator { /// 4. Reranking (heuristic or injected) /// 5. Context assembly pub async fn query(&self, query: &str) -> Result { + let output = self.query_ranked(query).await?; + let reranked = output + .results + .into_iter() + .map(|r| crate::rerank::RerankedResult { + doc_id: r.doc_id.clone(), + score: f64::from(r.score), + text: r.text_preview.clone(), + source_layer: format!("{:?}", r.source_layer), + inner: r, + }) + .collect(); + let mut ctx = ContextBuilder::build(query, reranked); + ctx.retrieval_ms = output.retrieval_ms; + Ok(ctx) + } + + /// Execute expand → fan-out → RRF → rerank and return ranked hits. + /// + /// Used by `RouteQuery` so gRPC callers get the orchestrator pipeline + /// without going through `MemoryContext` assembly. + pub async fn query_ranked(&self, query: &str) -> Result { let start = Instant::now(); - // 1. Query expansion let queries = if self.config.expand_query { expand_query(query) } else { vec![query.to_string()] }; - // 2. Fan-out: each query against each layer let layers = [ RetrievalLayer::Topics, RetrievalLayer::Vector, @@ -81,6 +117,7 @@ impl MemoryOrchestrator { let re = RetrievalExecutor::new(self.executor.clone()); let mut all_lists: Vec> = Vec::new(); + let mut layers_attempted: Vec = Vec::new(); for q in &queries { for &layer in &layers { @@ -99,6 +136,9 @@ impl MemoryOrchestrator { CapabilityTier::Full, ) .await; + if !layers_attempted.contains(&layer) { + layers_attempted.push(layer); + } if result.has_results() { all_lists.push(result.results); } @@ -106,17 +146,31 @@ impl MemoryOrchestrator { } } - // 3. RRF fusion - let fused = rrf_fuse(all_lists, self.config.rrf_k); - - // 4. Reranking — always use self.reranker (injected or default) + let fused = fuse(all_lists, self.config.fusion_k); let reranked = self.reranker.rerank(query, fused).await?; - // 5. Build context - let mut ctx = ContextBuilder::build(query, reranked); - ctx.retrieval_ms = start.elapsed().as_millis() as u64; - - Ok(ctx) + let results: Vec = reranked + .into_iter() + .map(|r| { + let mut inner = r.inner; + inner.score = r.score as f32; + inner + }) + .collect(); + + let rerank_mode = match self.config.rerank_mode { + crate::types::RerankMode::Heuristic => "heuristic", + crate::types::RerankMode::Llm => "llm", + } + .to_string(); + + Ok(OrchestratorOutput { + results, + fusion_stage: "rank_fusion", + rerank_mode, + layers_attempted, + retrieval_ms: start.elapsed().as_millis() as u64, + }) } } @@ -155,9 +209,10 @@ mod tests { .into_iter() .map(|r| RerankedResult { doc_id: r.inner.doc_id.clone(), - score: r.rrf_score, + score: r.fusion_score, text: r.inner.text_preview.clone(), source_layer: format!("{:?}", r.inner.source_layer), + inner: r.inner, }) .collect(); out.reverse(); @@ -253,4 +308,19 @@ mod tests { let result = orch.query("What happened with auth").await; assert!(result.is_ok()); } + + #[tokio::test] + async fn test_query_ranked_names_fusion_stage() { + let executor = MockLayerExecutor::default().with_results( + RetrievalLayer::BM25, + vec![mock_result("doc-x", 0.7, RetrievalLayer::BM25)], + ); + let config = OrchestratorConfig::default(); + let orch = MemoryOrchestrator::new(Arc::new(executor), config); + let output = orch.query_ranked("test").await.unwrap(); + assert_eq!(output.fusion_stage, "rank_fusion"); + assert_eq!(output.rerank_mode, "heuristic"); + assert!(!output.results.is_empty()); + assert!(output.layers_attempted.contains(&RetrievalLayer::BM25)); + } } diff --git a/crates/memory-orchestrator/src/rerank.rs b/crates/memory-orchestrator/src/rerank.rs index 8da23a2..7879fe3 100644 --- a/crates/memory-orchestrator/src/rerank.rs +++ b/crates/memory-orchestrator/src/rerank.rs @@ -1,15 +1,30 @@ //! Result reranking (heuristic and LLM-based). //! -//! Provides a `Reranker` trait with two implementations: +//! Provides a `Reranker` trait with three implementations: //! - `HeuristicReranker`: score-based sorting and top-K trimming (default). -//! - `CrossEncoderReranker`: stub that falls back to heuristic reranking -//! (extension point for future LLM-based reranking). +//! - `LlmReranker`: prompt an LLM over top-k candidates and honor the returned +//! order. Selected when an API key / `Completer` is configured. +//! - `CrossEncoderReranker`: extension point that returns `RerankError::NotImplemented` +//! — never warn-and-fallback. + +use std::collections::HashMap; +use std::sync::Arc; use anyhow::Result; use async_trait::async_trait; +use memory_retrieval::SearchResult; +use thiserror::Error; use crate::fusion::FusedResult; +/// Errors from reranking. +#[derive(Debug, Error)] +pub enum RerankError { + /// Cross-encoder path is an extension point, not a silent fallback. + #[error("cross-encoder reranking is not implemented")] + NotImplemented, +} + /// A reranked result ready for context assembly. #[derive(Debug, Clone)] pub struct RerankedResult { @@ -21,6 +36,20 @@ pub struct RerankedResult { pub text: String, /// Which retrieval layer produced this result (stringified). pub source_layer: String, + /// Original search result (doc_type, metadata, layer preserved). + pub inner: SearchResult, +} + +impl RerankedResult { + fn from_fused(fused: FusedResult, score: f64) -> Self { + Self { + doc_id: fused.inner.doc_id.clone(), + score, + text: fused.inner.text_preview.clone(), + source_layer: format!("{:?}", fused.inner.source_layer), + inner: fused.inner, + } + } } /// Trait for result reranking strategies. @@ -30,34 +59,54 @@ pub trait Reranker: Send + Sync { async fn rerank(&self, query: &str, results: Vec) -> Result>; } -/// Default reranker: sorts by RRF score descending and trims to top 10. -#[derive(Debug, Default)] -pub struct HeuristicReranker; +/// LLM text completion used by [`LlmReranker`]. +/// +/// Production wiring wraps `memory_toc::ApiSummarizer::complete`. Tests inject +/// a mock that returns a known JSON ordering. +#[async_trait] +pub trait Completer: Send + Sync { + /// Complete `prompt` and return the model text. + async fn complete(&self, prompt: &str) -> Result; +} + +/// Default reranker: sorts by RRF score descending and trims to `max_results`. +#[derive(Debug, Clone)] +pub struct HeuristicReranker { + max_results: usize, +} impl HeuristicReranker { - /// Maximum number of results to retain after reranking. - const MAX_RESULTS: usize = 10; + /// Construct a heuristic reranker that keeps `max_results` hits. + pub fn new(max_results: usize) -> Self { + Self { + max_results: max_results.max(1), + } + } fn rerank_sync(&self, results: Vec) -> Vec { let mut sorted = results; sorted.sort_by(|a, b| { - b.rrf_score - .partial_cmp(&a.rrf_score) + b.fusion_score + .partial_cmp(&a.fusion_score) .unwrap_or(std::cmp::Ordering::Equal) }); sorted .into_iter() - .take(Self::MAX_RESULTS) - .map(|r| RerankedResult { - doc_id: r.inner.doc_id, - score: r.rrf_score, - text: r.inner.text_preview, - source_layer: format!("{:?}", r.inner.source_layer), + .take(self.max_results) + .map(|r| { + let score = r.fusion_score; + RerankedResult::from_fused(r, score) }) .collect() } } +impl Default for HeuristicReranker { + fn default() -> Self { + Self::new(10) + } +} + #[async_trait] impl Reranker for HeuristicReranker { async fn rerank(&self, _query: &str, results: Vec) -> Result> { @@ -65,38 +114,150 @@ impl Reranker for HeuristicReranker { } } -/// Stub cross-encoder reranker. Falls back to heuristic reranking. -/// -/// This is the extension point (ORCH-05) for future LLM-based reranking. -/// When implemented, it will call an LLM to score query-document relevance -/// before sorting. -#[derive(Debug, Default)] -pub struct CrossEncoderReranker { - fallback: HeuristicReranker, +/// LLM reranker: asks a completer to order candidate doc ids. +pub struct LlmReranker { + completer: Arc, + max_results: usize, +} + +impl LlmReranker { + /// Create an LLM reranker around a completer. + pub fn new(completer: Arc, max_results: usize) -> Self { + Self { + completer, + max_results: max_results.max(1), + } + } + + fn build_prompt(query: &str, results: &[FusedResult]) -> String { + let mut docs = String::new(); + for (i, r) in results.iter().enumerate() { + docs.push_str(&format!( + "{}. id={}\n{}\n\n", + i + 1, + r.inner.doc_id, + r.inner.text_preview + )); + } + format!( + r#"Rank these memory documents for the search query. Return JSON only: +{{"order": ["doc_id_most_relevant", "..."]}} + +Query: {query} + +Documents: +{docs} +Include every doc_id exactly once, most relevant first."# + ) + } + + fn parse_order(text: &str, fallback: &[FusedResult]) -> Vec { + let json_str = extract_json_object(text); + let parsed: serde_json::Value = match serde_json::from_str(&json_str) { + Ok(v) => v, + Err(_) => { + return fallback.iter().map(|r| r.inner.doc_id.clone()).collect(); + } + }; + if let Some(arr) = parsed.get("order").and_then(|v| v.as_array()) { + let ids: Vec = arr + .iter() + .filter_map(|v| v.as_str().map(ToOwned::to_owned)) + .collect(); + if !ids.is_empty() { + return ids; + } + } + fallback.iter().map(|r| r.inner.doc_id.clone()).collect() + } +} + +fn extract_json_object(text: &str) -> String { + if let Some(start) = text.find('{') { + if let Some(end) = text.rfind('}') { + if end > start { + return text[start..=end].to_string(); + } + } + } + text.to_string() } #[async_trait] -impl Reranker for CrossEncoderReranker { +impl Reranker for LlmReranker { async fn rerank(&self, query: &str, results: Vec) -> Result> { - tracing::warn!( - "CrossEncoderReranker not yet implemented, falling back to heuristic reranking" - ); - self.fallback.rerank(query, results).await + if results.is_empty() { + return Ok(Vec::new()); + } + let prompt = Self::build_prompt(query, &results); + let response = match self.completer.complete(&prompt).await { + Ok(text) => text, + Err(e) => { + tracing::warn!(error = %e, "LLM rerank failed; keeping RRF order"); + return HeuristicReranker::new(self.max_results) + .rerank(query, results) + .await; + } + }; + let order = Self::parse_order(&response, &results); + let mut by_id: HashMap = results + .into_iter() + .map(|r| (r.inner.doc_id.clone(), r)) + .collect(); + let mut out = Vec::new(); + for id in order { + if let Some(fused) = by_id.remove(&id) { + let score = fused.fusion_score; + out.push(RerankedResult::from_fused(fused, score)); + } + } + // Append any ids the model omitted (sort leftover by fusion score). + let mut leftover: Vec = by_id.into_values().collect(); + leftover.sort_by(|a, b| { + b.fusion_score + .partial_cmp(&a.fusion_score) + .unwrap_or(std::cmp::Ordering::Equal) + }); + for fused in leftover { + out.push(RerankedResult::from_fused(fused, 0.0)); + } + out.truncate(self.max_results); + // Positional scores so downstream ranking cannot undo the LLM order. + let n = out.len() as f64; + for (i, item) in out.iter_mut().enumerate() { + item.score = ((n - i as f64) / n).clamp(0.0, 1.0); + } + Ok(out) + } +} + +/// Stub cross-encoder reranker. Returns a hard error — never a silent fallback. +#[derive(Debug, Default)] +pub struct CrossEncoderReranker; + +#[async_trait] +impl Reranker for CrossEncoderReranker { + async fn rerank( + &self, + _query: &str, + _results: Vec, + ) -> Result> { + Err(RerankError::NotImplemented.into()) } } #[cfg(test)] mod tests { use super::*; - use memory_retrieval::{RetrievalLayer, SearchResult}; + use memory_retrieval::RetrievalLayer; - fn make_fused(id: &str, rrf_score: f64) -> FusedResult { + fn make_fused(id: &str, fusion_score: f64) -> FusedResult { FusedResult { - rrf_score, + fusion_score, inner: SearchResult { doc_id: id.to_string(), doc_type: "toc_node".to_string(), - score: rrf_score as f32, + score: fusion_score as f32, text_preview: format!("text for {id}"), source_layer: RetrievalLayer::BM25, metadata: Default::default(), @@ -109,10 +270,9 @@ mod tests { let mut results: Vec = (0..20) .map(|i| make_fused(&format!("doc-{i}"), 1.0 - i as f64 * 0.01)) .collect(); - // Shuffle to verify sorting works results.reverse(); - let reranker = HeuristicReranker; + let reranker = HeuristicReranker::new(10); let reranked = reranker.rerank("test query", results).await.unwrap(); assert_eq!(reranked.len(), 10, "should trim to top 10"); @@ -121,17 +281,52 @@ mod tests { reranked[0].score > reranked[9].score, "first should score higher than last" ); + assert_eq!(reranked[0].inner.doc_type, "toc_node"); } #[tokio::test] - async fn test_cross_encoder_falls_back_to_heuristic() { + async fn test_cross_encoder_returns_not_implemented() { let results = vec![make_fused("a", 0.9), make_fused("b", 0.5)]; + let reranker = CrossEncoderReranker; + let err = reranker.rerank("test query", results).await.unwrap_err(); + assert!( + err.to_string().contains("not implemented"), + "cross-encoder must hard-error, got: {err}" + ); + } - let reranker = CrossEncoderReranker::default(); - let reranked = reranker.rerank("test query", results).await.unwrap(); + struct ReverseCompleter; - // Should not panic and should produce results - assert_eq!(reranked.len(), 2); + #[async_trait] + impl Completer for ReverseCompleter { + async fn complete(&self, _prompt: &str) -> Result { + Ok(r#"{"order": ["b", "a"]}"#.to_string()) + } + } + + #[tokio::test] + async fn test_llm_reranker_honors_completer_order() { + let results = vec![make_fused("a", 0.9), make_fused("b", 0.5)]; + let reranker = LlmReranker::new(Arc::new(ReverseCompleter), 10); + let reranked = reranker.rerank("q", results).await.unwrap(); + assert_eq!(reranked[0].doc_id, "b"); + assert_eq!(reranked[1].doc_id, "a"); + } + + struct BrokenCompleter; + + #[async_trait] + impl Completer for BrokenCompleter { + async fn complete(&self, _prompt: &str) -> Result { + anyhow::bail!("network down"); + } + } + + #[tokio::test] + async fn test_llm_reranker_fail_open_on_completer_error() { + let results = vec![make_fused("a", 0.9), make_fused("b", 0.5)]; + let reranker = LlmReranker::new(Arc::new(BrokenCompleter), 10); + let reranked = reranker.rerank("q", results).await.unwrap(); assert_eq!(reranked[0].doc_id, "a"); assert_eq!(reranked[1].doc_id, "b"); } diff --git a/crates/memory-orchestrator/src/types.rs b/crates/memory-orchestrator/src/types.rs index 289a02c..d8cff4c 100644 --- a/crates/memory-orchestrator/src/types.rs +++ b/crates/memory-orchestrator/src/types.rs @@ -25,8 +25,8 @@ pub struct OrchestratorConfig { pub rerank_mode: RerankMode, /// Whether to expand the query into multiple variants before search. pub expand_query: bool, - /// Reciprocal Rank Fusion constant (higher = more weight to lower-ranked docs). - pub rrf_k: f64, + /// Rank-fusion damping constant (higher = more weight to lower-ranked docs). + pub fusion_k: f64, } impl Default for OrchestratorConfig { @@ -35,7 +35,7 @@ impl Default for OrchestratorConfig { top_k: 10, rerank_mode: RerankMode::Heuristic, expand_query: false, - rrf_k: 60.0, + fusion_k: 60.0, } } } @@ -87,9 +87,9 @@ mod tests { } #[test] - fn test_orchestrator_config_rrf_k() { + fn test_orchestrator_config_fusion_k() { let config = OrchestratorConfig::default(); - assert!((config.rrf_k - 60.0).abs() < f64::EPSILON); + assert!((config.fusion_k - 60.0).abs() < f64::EPSILON); } #[test] diff --git a/crates/memory-scheduler/Cargo.toml b/crates/memory-scheduler/Cargo.toml index c0fe2ad..7451a56 100644 --- a/crates/memory-scheduler/Cargo.toml +++ b/crates/memory-scheduler/Cargo.toml @@ -8,7 +8,7 @@ description = "Background job scheduler for agent-memory daemon" [features] default = ["jobs"] -jobs = ["memory-toc", "memory-storage", "memory-types", "memory-search", "memory-indexing", "memory-vector"] +jobs = ["memory-toc", "memory-storage", "memory-search", "memory-indexing", "memory-vector"] [dependencies] # Cron scheduling @@ -37,10 +37,12 @@ uuid = "1.11" # Random number generation for jitter rand = "0.8" +# Core (always required) +memory-types = { workspace = true } + # Job dependencies (optional, enabled by default) memory-toc = { path = "../memory-toc", optional = true } memory-storage = { path = "../memory-storage", optional = true } -memory-types = { path = "../memory-types", optional = true } memory-search = { path = "../memory-search", optional = true } memory-indexing = { path = "../memory-indexing", optional = true } memory-vector = { path = "../memory-vector", optional = true } diff --git a/crates/memory-scheduler/src/registry.rs b/crates/memory-scheduler/src/registry.rs index e0b3f66..1e6fcd1 100644 --- a/crates/memory-scheduler/src/registry.rs +++ b/crates/memory-scheduler/src/registry.rs @@ -7,6 +7,7 @@ use std::collections::HashMap; use std::sync::RwLock; use chrono::{DateTime, Utc}; +use memory_types::recover_lock; use serde::{Deserialize, Serialize}; /// Result of a job execution. @@ -139,7 +140,7 @@ impl JobRegistry { /// /// If a job with the same name already exists, it will be replaced. pub fn register(&self, job_name: &str, cron_expr: &str) { - let mut jobs = self.jobs.write().unwrap(); + let mut jobs = recover_lock(self.jobs.write()); jobs.insert( job_name.to_string(), JobStatus::new(job_name.to_string(), cron_expr.to_string()), @@ -148,7 +149,7 @@ impl JobRegistry { /// Record that a job has started executing. pub fn record_start(&self, job_name: &str) { - let mut jobs = self.jobs.write().unwrap(); + let mut jobs = recover_lock(self.jobs.write()); if let Some(status) = jobs.get_mut(job_name) { status.is_running = true; } @@ -172,7 +173,7 @@ impl JobRegistry { duration_ms: u64, metadata: HashMap, ) { - let mut jobs = self.jobs.write().unwrap(); + let mut jobs = recover_lock(self.jobs.write()); if let Some(status) = jobs.get_mut(job_name) { status.is_running = false; status.last_run = Some(Utc::now()); @@ -188,7 +189,7 @@ impl JobRegistry { /// Update the next scheduled run time for a job. pub fn set_next_run(&self, job_name: &str, next: DateTime) { - let mut jobs = self.jobs.write().unwrap(); + let mut jobs = recover_lock(self.jobs.write()); if let Some(status) = jobs.get_mut(job_name) { status.next_run = Some(next); } @@ -196,7 +197,7 @@ impl JobRegistry { /// Set the paused state of a job. pub fn set_paused(&self, job_name: &str, paused: bool) { - let mut jobs = self.jobs.write().unwrap(); + let mut jobs = recover_lock(self.jobs.write()); if let Some(status) = jobs.get_mut(job_name) { status.is_paused = paused; } @@ -206,21 +207,19 @@ impl JobRegistry { /// /// Returns `None` if the job is not registered. pub fn get_status(&self, job_name: &str) -> Option { - self.jobs.read().unwrap().get(job_name).cloned() + recover_lock(self.jobs.read()).get(job_name).cloned() } /// Get the status of all registered jobs. pub fn get_all_status(&self) -> Vec { - self.jobs.read().unwrap().values().cloned().collect() + recover_lock(self.jobs.read()).values().cloned().collect() } /// Check if a job is currently running. /// /// Returns `false` if the job is not registered. pub fn is_running(&self, job_name: &str) -> bool { - self.jobs - .read() - .unwrap() + recover_lock(self.jobs.read()) .get(job_name) .map(|s| s.is_running) .unwrap_or(false) @@ -228,16 +227,14 @@ impl JobRegistry { /// Check if a job is registered. pub fn is_registered(&self, job_name: &str) -> bool { - self.jobs.read().unwrap().contains_key(job_name) + recover_lock(self.jobs.read()).contains_key(job_name) } /// Check if a job is paused. /// /// Returns `false` if the job is not registered. pub fn is_paused(&self, job_name: &str) -> bool { - self.jobs - .read() - .unwrap() + recover_lock(self.jobs.read()) .get(job_name) .map(|s| s.is_paused) .unwrap_or(false) @@ -245,7 +242,7 @@ impl JobRegistry { /// Get the number of registered jobs. pub fn job_count(&self) -> usize { - self.jobs.read().unwrap().len() + recover_lock(self.jobs.read()).len() } } diff --git a/crates/memory-search/src/document.rs b/crates/memory-search/src/document.rs index dc71d47..970c993 100644 --- a/crates/memory-search/src/document.rs +++ b/crates/memory-search/src/document.rs @@ -5,7 +5,7 @@ use tantivy::doc; use tantivy::TantivyDocument; -use memory_types::{Grip, TocNode}; +use memory_types::{Event, Grip, TocNode}; use crate::schema::{DocType, SearchSchema}; @@ -63,6 +63,24 @@ pub fn grip_to_doc(schema: &SearchSchema, grip: &Grip) -> TantivyDocument { ) } +/// Convert an ingested [`Event`] to a Tantivy document. +/// +/// Text field contains the event body. Doc id is the event_id so BM25 +/// teleport can find it immediately after the outbox drain (Phase 54-02). +pub fn event_to_doc(schema: &SearchSchema, event: &Event) -> TantivyDocument { + let timestamp = event.timestamp.timestamp_millis().to_string(); + let agent = event.agent.clone().unwrap_or_default(); + doc!( + schema.doc_type => DocType::Event.as_str(), + schema.doc_id => event.event_id.clone(), + schema.level => "", + schema.text => event.text.clone(), + schema.keywords => "", + schema.timestamp_ms => timestamp, + schema.agent => agent + ) +} + /// Extract text content from a TocNode for indexing. /// /// Returns combined title and bullet text. diff --git a/crates/memory-search/src/indexer.rs b/crates/memory-search/src/indexer.rs index d272692..2ca3415 100644 --- a/crates/memory-search/src/indexer.rs +++ b/crates/memory-search/src/indexer.rs @@ -12,9 +12,9 @@ use tantivy::schema::Value; use tantivy::{IndexReader, IndexWriter, ReloadPolicy, Term}; use tracing::{debug, info, warn}; -use memory_types::{Grip, TocNode}; +use memory_types::{Event, Grip, TocNode}; -use crate::document::{grip_to_doc, toc_node_to_doc}; +use crate::document::{event_to_doc, grip_to_doc, toc_node_to_doc}; use crate::error::SearchError; use crate::index::SearchIndex; use crate::lifecycle::Bm25PruneStats; @@ -106,6 +106,25 @@ impl SearchIndexer { Ok(()) } + /// Index a raw ingested event so it is findable via BM25 immediately. + /// + /// If a document with the same event_id exists, it will be replaced. + pub fn index_event(&self, event: &Event) -> Result<(), SearchError> { + let doc = event_to_doc(&self.schema, event); + + let writer = self + .writer + .lock() + .map_err(|e| SearchError::IndexLocked(e.to_string()))?; + + let term = Term::from_field_text(self.schema.doc_id, &event.event_id); + writer.delete_term(term); + writer.add_document(doc)?; + + debug!(event_id = %event.event_id, "Indexed event"); + Ok(()) + } + /// Index multiple TOC nodes in batch. pub fn index_toc_nodes(&self, nodes: &[TocNode]) -> Result { let writer = self diff --git a/crates/memory-search/src/lib.rs b/crates/memory-search/src/lib.rs index 49da0f2..a6d3d7f 100644 --- a/crates/memory-search/src/lib.rs +++ b/crates/memory-search/src/lib.rs @@ -25,7 +25,7 @@ pub mod lifecycle; pub mod schema; pub mod searcher; -pub use document::{extract_toc_text, grip_to_doc, toc_node_to_doc}; +pub use document::{event_to_doc, extract_toc_text, grip_to_doc, toc_node_to_doc}; pub use error::SearchError; pub use index::{open_or_create_index, SearchIndex, SearchIndexConfig}; pub use indexer::SearchIndexer; diff --git a/crates/memory-search/src/schema.rs b/crates/memory-search/src/schema.rs index a9e117c..66253a8 100644 --- a/crates/memory-search/src/schema.rs +++ b/crates/memory-search/src/schema.rs @@ -13,6 +13,8 @@ use crate::SearchError; pub enum DocType { TocNode, Grip, + /// Raw ingested event (Phase 54: outbox-driven BM25 indexing). + Event, } impl DocType { @@ -20,6 +22,7 @@ impl DocType { match self { DocType::TocNode => "toc_node", DocType::Grip => "grip", + DocType::Event => "event", } } @@ -28,6 +31,7 @@ impl DocType { match s { "toc_node" => Some(DocType::TocNode), "grip" => Some(DocType::Grip), + "event" => Some(DocType::Event), _ => None, } } @@ -167,6 +171,7 @@ mod tests { fn test_doc_type_conversion() { assert_eq!(DocType::TocNode.as_str(), "toc_node"); assert_eq!(DocType::parse("grip"), Some(DocType::Grip)); + assert_eq!(DocType::parse("event"), Some(DocType::Event)); assert_eq!(DocType::parse("invalid"), None); // Test FromStr trait assert_eq!("toc_node".parse::().unwrap(), DocType::TocNode); diff --git a/crates/memory-service/Cargo.toml b/crates/memory-service/Cargo.toml index 1611b6e..2b69911 100644 --- a/crates/memory-service/Cargo.toml +++ b/crates/memory-service/Cargo.toml @@ -14,6 +14,7 @@ memory-embeddings = { workspace = true } memory-vector = { workspace = true } memory-topics = { workspace = true } memory-retrieval = { workspace = true } +memory-orchestrator = { workspace = true } tokio = { workspace = true } tonic = { workspace = true } tonic-health = { workspace = true } diff --git a/crates/memory-service/src/hybrid.rs b/crates/memory-service/src/hybrid.rs index c7c5f1c..e483149 100644 --- a/crates/memory-service/src/hybrid.rs +++ b/crates/memory-service/src/hybrid.rs @@ -1,8 +1,7 @@ //! HybridSearch RPC implementation. //! -//! Combines BM25 and vector search using Reciprocal Rank Fusion (RRF). -//! RRF_score(doc) = sum(weight_i / (k + rank_i(doc))) -//! where k=60 is the standard constant. +//! Combines BM25 and vector search using the workspace's canonical +//! weighted rank-fusion in `memory_orchestrator::fusion`. use std::collections::HashMap; use std::sync::Arc; @@ -10,6 +9,8 @@ use std::sync::Arc; use tonic::{Request, Response, Status}; use tracing::{debug, info}; +use memory_orchestrator::fuse_weighted; +use memory_retrieval::{RetrievalLayer, SearchResult}; use memory_search::{SearchOptions, TeleportSearcher}; use crate::pb::{ @@ -17,8 +18,8 @@ use crate::pb::{ }; use crate::vector::VectorTeleportHandler; -/// Standard RRF constant (from original RRF paper) -const RRF_K: f32 = 60.0; +/// Standard rank-fusion damping constant (Cormack et al.). +const FUSION_K: f64 = 60.0; /// Handler for hybrid search operations. pub struct HybridSearchHandler { @@ -84,7 +85,7 @@ impl HybridSearchHandler { HybridMode::Hybrid | HybridMode::Unspecified => { if self.vector_available() && self.bm25_available() { let fused = self - .fuse_rrf(query, top_k, bm25_weight, vector_weight, &req) + .fuse_lists(query, top_k, bm25_weight, vector_weight, &req) .await?; (HybridMode::Hybrid, fused) } else if self.vector_available() { @@ -155,8 +156,8 @@ impl HybridSearchHandler { .collect()) } - /// Fuse results using Reciprocal Rank Fusion. - async fn fuse_rrf( + /// Fuse BM25 + vector lists via the canonical weighted rank-fusion. + async fn fuse_lists( &self, query: &str, top_k: usize, @@ -164,76 +165,63 @@ impl HybridSearchHandler { vector_weight: f32, req: &HybridSearchRequest, ) -> Result, Status> { - // Fetch more results for fusion let fetch_k = top_k * 2; let vector_results = self.vector_search(query, fetch_k, req).await?; let bm25_results = self.bm25_search(query, fetch_k).await?; - let mut rrf: HashMap = HashMap::new(); - - // Accumulate vector RRF scores - for (rank, m) in vector_results.into_iter().enumerate() { - let score = vector_weight / (RRF_K + rank as f32 + 1.0); - let entry = rrf - .entry(m.doc_id.clone()) - .or_insert_with(|| RrfEntry::from(&m)); - entry.rrf_score += score; + let mut extras: HashMap = HashMap::new(); + let mut vector_list = Vec::new(); + for m in vector_results { + extras.entry(m.doc_id.clone()).or_insert_with(|| m.clone()); + vector_list.push(match_to_search(m, RetrievalLayer::Vector)); } - - // Accumulate BM25 RRF scores - for (rank, m) in bm25_results.into_iter().enumerate() { - let score = bm25_weight / (RRF_K + rank as f32 + 1.0); - let entry = rrf - .entry(m.doc_id.clone()) - .or_insert_with(|| RrfEntry::from(&m)); - entry.rrf_score += score; + let mut bm25_list = Vec::new(); + for m in bm25_results { + extras.entry(m.doc_id.clone()).or_insert_with(|| m.clone()); + bm25_list.push(match_to_search(m, RetrievalLayer::BM25)); } - // Sort by RRF score and truncate - let mut entries: Vec<_> = rrf.into_values().collect(); - entries.sort_by(|a, b| { - b.rrf_score - .partial_cmp(&a.rrf_score) - .unwrap_or(std::cmp::Ordering::Equal) - }); - entries.truncate(top_k); + let fused = fuse_weighted( + vec![ + (f64::from(vector_weight), vector_list), + (f64::from(bm25_weight), bm25_list), + ], + FUSION_K, + ); - // Convert to VectorMatch - Ok(entries + Ok(fused .into_iter() - .map(|e| VectorMatch { - doc_id: e.doc_id, - doc_type: e.doc_type, - score: e.rrf_score, - text_preview: e.text_preview, - timestamp_ms: e.timestamp_ms, - agent: e.agent, + .take(top_k) + .map(|f| { + if let Some(orig) = extras.get(&f.inner.doc_id) { + VectorMatch { + score: f.fusion_score as f32, + ..orig.clone() + } + } else { + VectorMatch { + doc_id: f.inner.doc_id, + doc_type: f.inner.doc_type, + score: f.fusion_score as f32, + text_preview: f.inner.text_preview, + timestamp_ms: 0, + agent: None, + } + } }) .collect()) } } -/// Entry for RRF accumulation. -struct RrfEntry { - doc_id: String, - doc_type: String, - text_preview: String, - timestamp_ms: i64, - agent: Option, - rrf_score: f32, -} - -impl From<&VectorMatch> for RrfEntry { - fn from(m: &VectorMatch) -> Self { - Self { - doc_id: m.doc_id.clone(), - doc_type: m.doc_type.clone(), - text_preview: m.text_preview.clone(), - timestamp_ms: m.timestamp_ms, - agent: m.agent.clone(), - rrf_score: 0.0, - } +fn match_to_search(m: VectorMatch, layer: RetrievalLayer) -> SearchResult { + SearchResult { + doc_id: m.doc_id, + doc_type: m.doc_type, + score: m.score, + text_preview: m.text_preview, + source_layer: layer, + metadata: Default::default(), } } @@ -242,13 +230,12 @@ mod tests { use super::*; #[test] - fn test_rrf_k_constant() { - // Verify RRF_K is the standard value from the paper - assert_eq!(RRF_K, 60.0); + fn test_fusion_k_constant() { + assert!((FUSION_K - 60.0).abs() < f64::EPSILON); } #[test] - fn test_rrf_entry_from_vector_match() { + fn test_match_to_search_preserves_fields() { let m = VectorMatch { doc_id: "test-123".to_string(), doc_type: "toc_node".to_string(), @@ -258,9 +245,10 @@ mod tests { agent: None, }; - let entry = RrfEntry::from(&m); - assert_eq!(entry.doc_id, "test-123"); - assert_eq!(entry.doc_type, "toc_node"); - assert_eq!(entry.rrf_score, 0.0); // Should start at 0 + let sr = match_to_search(m, RetrievalLayer::Vector); + assert_eq!(sr.doc_id, "test-123"); + assert_eq!(sr.doc_type, "toc_node"); + assert!((sr.score - 0.95).abs() < f32::EPSILON); + assert_eq!(sr.source_layer, RetrievalLayer::Vector); } } diff --git a/crates/memory-service/src/ingest.rs b/crates/memory-service/src/ingest.rs index aa5e387..22acd1f 100644 --- a/crates/memory-service/src/ingest.rs +++ b/crates/memory-service/src/ingest.rs @@ -51,6 +51,20 @@ use crate::search_service; use crate::teleport_service; use crate::topics::TopicGraphHandler; use crate::vector::VectorTeleportHandler; +use memory_toc::ApiSummarizer; + +/// Optional query-path indexes attached at daemon start (Phase 54). +#[derive(Default)] +pub struct QueryIndexBundle { + /// BM25 teleport searcher. + pub searcher: Option>, + /// Vector teleport handler. + pub vector: Option>, + /// Topic graph handler. + pub topics: Option>, + /// API summarizer used as the LLM rerank completer. + pub api_summarizer: Option>, +} /// Implementation of the MemoryService gRPC service. pub struct MemoryServiceImpl { @@ -321,6 +335,38 @@ impl MemoryServiceImpl { self.episode_handler = Some(handler); } + /// Attach live BM25/vector/topic indexes and an optional LLM completer. + /// + /// Replaces the retrieval handler constructed by `with_scheduler` (which + /// starts with all indexes `None`) so `RouteQuery` and teleport RPCs hit + /// real indexes. + pub fn attach_indexes(&mut self, indexes: QueryIndexBundle, staleness_config: StalenessConfig) { + if let Some(searcher) = &indexes.searcher { + self.teleport_searcher = Some(Arc::clone(searcher)); + } + if let Some(vector) = &indexes.vector { + self.vector_service = Some(Arc::clone(vector)); + self.hybrid_service = Some(Arc::new(HybridSearchHandler::new( + Arc::clone(vector), + indexes.searcher.clone(), + ))); + } + if let Some(topics) = &indexes.topics { + self.topic_service = Some(Arc::clone(topics)); + } + let mut retrieval = RetrievalHandler::with_services( + Arc::clone(&self.storage), + indexes.searcher, + indexes.vector, + indexes.topics, + staleness_config, + ); + if let Some(summarizer) = indexes.api_summarizer { + retrieval = retrieval.with_api_summarizer(summarizer); + } + self.retrieval_service = Some(Arc::new(retrieval)); + } + /// Convert proto EventRole to domain EventRole fn convert_role(proto_role: ProtoEventRole) -> EventRole { match proto_role { diff --git a/crates/memory-service/src/lib.rs b/crates/memory-service/src/lib.rs index 2c2ae48..06b6667 100644 --- a/crates/memory-service/src/lib.rs +++ b/crates/memory-service/src/lib.rs @@ -34,11 +34,11 @@ pub mod pb { pub use agents::AgentDiscoveryHandler; pub use episodes::EpisodeHandler; pub use hybrid::HybridSearchHandler; -pub use ingest::MemoryServiceImpl; +pub use ingest::{MemoryServiceImpl, QueryIndexBundle}; pub use novelty::{ CandleEmbedderAdapter, DedupResult, NoveltyChecker, NoveltyMetrics, NoveltyMetricsSnapshot, }; -pub use retrieval::RetrievalHandler; +pub use retrieval::{ApiCompleter, RetrievalHandler}; pub use scheduler_service::SchedulerGrpcService; pub use server::{run_server, run_server_with_scheduler, run_server_with_shutdown}; pub use topics::{TopicGraphHandler, TopicGraphStatus, TopicSearchResult}; diff --git a/crates/memory-service/src/retrieval.rs b/crates/memory-service/src/retrieval.rs index 7623a04..b87c7be 100644 --- a/crates/memory-service/src/retrieval.rs +++ b/crates/memory-service/src/retrieval.rs @@ -15,9 +15,13 @@ use async_trait::async_trait; use tonic::{Request, Response, Status}; use tracing::{debug, info}; +use memory_orchestrator::{ + fuse_weighted, Completer, HeuristicReranker, LlmReranker, MemoryOrchestrator, + OrchestratorConfig, RerankMode, Reranker, +}; use memory_retrieval::{ classifier::IntentClassifier, - executor::{FallbackChain, LayerExecutor, RetrievalExecutor, SearchResult}, + executor::{LayerExecutor, SearchResult}, ranking::{apply_combined_ranking, RankingConfig}, stale_filter::StaleFilter, types::{ @@ -28,6 +32,7 @@ use memory_retrieval::{ }; use memory_search::TeleportSearcher; use memory_storage::Storage; +use memory_toc::ApiSummarizer; use memory_types::config::StalenessConfig; use crate::federated::federated_query; @@ -69,6 +74,9 @@ pub struct RetrievalHandler { /// Path of the primary store (used for result attribution). primary_db_path: String, + + /// Optional LLM completer for `rerank_mode=llm`. + completer: Option>, } impl RetrievalHandler { @@ -83,6 +91,7 @@ impl RetrievalHandler { staleness_config: StalenessConfig::default(), registered_projects: Vec::new(), primary_db_path: String::new(), + completer: None, } } @@ -103,6 +112,7 @@ impl RetrievalHandler { staleness_config, registered_projects: Vec::new(), primary_db_path: String::new(), + completer: None, } } @@ -117,6 +127,17 @@ impl RetrievalHandler { self } + /// Inject an LLM completer used when `rerank_mode=llm`. + pub fn with_completer(mut self, completer: Arc) -> Self { + self.completer = Some(completer); + self + } + + /// Inject an [`ApiSummarizer`] as the LLM completer for `rerank_mode=llm`. + pub fn with_api_summarizer(self, summarizer: Arc) -> Self { + self.with_completer(Arc::new(ApiCompleter::new(summarizer))) + } + /// Handle GetRetrievalCapabilities RPC. /// /// Per PRD Section 5.2: Combined status check pattern. @@ -243,8 +264,10 @@ impl RetrievalHandler { return Err(Status::invalid_argument("Query is required")); } - // Get stop conditions - let stop_conditions = req + // Get stop conditions (honored by the fallback executor inside the orchestrator + // via StopConditions::default today; parsed here so the proto field is not dropped + // silently if we thread it through later). + let _stop_conditions = req .stop_conditions .map(|sc| proto_to_stop_conditions(&sc)) .unwrap_or_default(); @@ -280,11 +303,9 @@ impl RetrievalHandler { 10 }; - // Execute the retrieval + // Execute expand → fan-out → rank fusion → rerank via the orchestrator. let start = Instant::now(); - let chain = FallbackChain::for_intent(intent, tier); - // Create a simple executor that delegates to our services let executor = Arc::new(SimpleLayerExecutor::new( self.storage.clone(), self.bm25_searcher.clone(), @@ -292,13 +313,51 @@ impl RetrievalHandler { self.topic_handler.clone(), )); - let retrieval_executor = RetrievalExecutor::new(executor); - let result = retrieval_executor - .execute(&req.query, chain, &stop_conditions, mode, tier) - .await; + let requested_rerank = match req.rerank_mode.as_deref() { + Some(mode) if mode.eq_ignore_ascii_case("llm") => RerankMode::Llm, + _ => RerankMode::Heuristic, + }; + let (rerank_mode, reranker): (RerankMode, Box) = match requested_rerank { + RerankMode::Llm => { + if let Some(completer) = &self.completer { + ( + RerankMode::Llm, + Box::new(LlmReranker::new(Arc::clone(completer), limit)), + ) + } else { + tracing::warn!( + "rerank_mode=llm requested but no LLM completer is configured; using heuristic" + ); + ( + RerankMode::Heuristic, + Box::new(HeuristicReranker::new(limit)), + ) + } + } + RerankMode::Heuristic => ( + RerankMode::Heuristic, + Box::new(HeuristicReranker::new(limit)), + ), + }; + + let orch_config = OrchestratorConfig { + top_k: limit, + rerank_mode, + expand_query: req.expand_query, + fusion_k: 60.0, + }; + let orchestrator = MemoryOrchestrator::with_reranker(executor, orch_config, reranker); + let output = orchestrator + .query_ranked(&req.query) + .await + .map_err(|e| Status::internal(format!("orchestrator error: {e}")))?; + + let fusion_stage = output.fusion_stage; + let orch_rerank_mode = output.rerank_mode.clone(); + let layers_attempted = output.layers_attempted.clone(); // Enrich metadata with salience scores from Storage lookups - let enriched_results = enrich_with_salience(&self.storage, result.results); + let enriched_results = enrich_with_salience(&self.storage, output.results); // Apply staleness filter post-merge, pre-return let stale_filter = StaleFilter::new(self.staleness_config.clone()); @@ -356,29 +415,35 @@ impl RetrievalHandler { .collect(); // Build explainability payload + let primary_layer = final_results + .first() + .map(|r| r.source_layer) + .or_else(|| layers_attempted.first().copied()) + .unwrap_or(CrateLayer::Agentic); let explanation = ProtoExplainability { intent: intent_to_proto(intent) as i32, tier: tier_to_proto(tier) as i32, mode: exec_mode_to_proto(mode) as i32, - candidates_considered: result - .layers_attempted + candidates_considered: layers_attempted .iter() .map(|l| layer_to_proto(*l) as i32) .collect(), - winner: layer_to_proto(result.primary_layer) as i32, - why_winner: result.explanation.clone(), - fallback_occurred: result.fallback_occurred, - fallback_reason: if result.fallback_occurred { - Some(result.explanation.clone()) - } else { - None - }, + winner: layer_to_proto(primary_layer) as i32, + why_winner: format!( + "rank_fusion across {} layer(s); rerank={}", + layers_attempted.len(), + orch_rerank_mode + ), + fallback_occurred: false, + fallback_reason: None, total_time_ms, grip_ids: results .iter() .filter(|r| r.doc_type == "grip") .map(|r| r.doc_id.clone()) .collect(), + fusion_stage: fusion_stage.to_string(), + rerank_mode: Some(orch_rerank_mode), }; let has_results = !results.is_empty(); @@ -397,8 +462,7 @@ impl RetrievalHandler { results, explanation: Some(explanation), has_results, - layers_attempted: result - .layers_attempted + layers_attempted: layers_attempted .iter() .map(|l| layer_to_proto(*l) as i32) .collect(), @@ -598,44 +662,71 @@ impl LayerExecutor for SimpleLayerExecutor { } } CrateLayer::Hybrid => { - // Hybrid combines BM25 and Vector - for now, delegate to BM25 if available + let fetch_k = limit.max(1); + let mut bm25_list = Vec::new(); + let mut vector_list = Vec::new(); + if let Some(searcher) = &self.bm25_searcher { - let opts = memory_search::SearchOptions::new().with_limit(limit); - let results = searcher.search(query, opts).map_err(|e| e.to_string())?; - Ok(results - .into_iter() - .map(|r| SearchResult { - doc_id: r.doc_id, - doc_type: format!("{:?}", r.doc_type).to_lowercase(), - score: r.score, - text_preview: r.keywords.unwrap_or_default(), - source_layer: CrateLayer::Hybrid, - metadata: build_metadata( - r.timestamp_ms, - r.agent.as_deref(), - "observation", - ), - }) - .collect()) - } else if let Some(handler) = &self.vector_handler { - let results = handler.search(query, limit, 0.0).await?; - Ok(results - .into_iter() - .map(|r| SearchResult { - doc_id: r.doc_id, - doc_type: r.doc_type, - score: r.score, - text_preview: r.text_preview, - source_layer: CrateLayer::Hybrid, - metadata: build_metadata( - Some(r.timestamp_ms), - r.agent.as_deref(), - "observation", - ), - }) - .collect()) - } else { + let opts = memory_search::SearchOptions::new().with_limit(fetch_k); + match searcher.search(query, opts) { + Ok(results) => { + bm25_list = results + .into_iter() + .map(|r| SearchResult { + doc_id: r.doc_id, + doc_type: format!("{:?}", r.doc_type).to_lowercase(), + score: r.score, + text_preview: r.keywords.unwrap_or_default(), + source_layer: CrateLayer::BM25, + metadata: build_metadata( + r.timestamp_ms, + r.agent.as_deref(), + "observation", + ), + }) + .collect(); + } + Err(e) => { + tracing::warn!( + error = %e, + "Hybrid BM25 search failed; continuing with remaining indexes" + ); + } + } + } + + if let Some(handler) = &self.vector_handler { + match handler.search(query, fetch_k, 0.0).await { + Ok(results) => { + vector_list = results + .into_iter() + .map(|r| SearchResult { + doc_id: r.doc_id, + doc_type: r.doc_type, + score: r.score, + text_preview: r.text_preview, + source_layer: CrateLayer::Vector, + metadata: build_metadata( + Some(r.timestamp_ms), + r.agent.as_deref(), + "observation", + ), + }) + .collect(); + } + Err(e) => { + tracing::warn!( + error = %e, + "Hybrid vector search failed; continuing with remaining indexes" + ); + } + } + } + + if self.bm25_searcher.is_none() && self.vector_handler.is_none() { Err("Hybrid requires BM25 or Vector".to_string()) + } else { + Ok(fuse_hybrid_lists(bm25_list, vector_list, limit)) } } CrateLayer::Agentic => { @@ -659,6 +750,56 @@ impl LayerExecutor for SimpleLayerExecutor { } } +/// Completer that wraps [`ApiSummarizer`] for `rerank_mode=llm`. +pub struct ApiCompleter { + inner: Arc, +} + +impl ApiCompleter { + /// Wrap an API summarizer as an orchestrator completer. + pub fn new(inner: Arc) -> Self { + Self { inner } + } +} + +#[async_trait] +impl Completer for ApiCompleter { + async fn complete(&self, prompt: &str) -> anyhow::Result { + self.inner + .complete(prompt) + .await + .map_err(|e| anyhow::anyhow!("{e}")) + } +} + +/// Fuse BM25 + vector hit lists and retag the winner as Hybrid. +fn fuse_hybrid_lists( + bm25: Vec, + vector: Vec, + limit: usize, +) -> Vec { + let mut lists = Vec::new(); + if !bm25.is_empty() { + lists.push((0.5, bm25)); + } + if !vector.is_empty() { + lists.push((0.5, vector)); + } + if lists.is_empty() { + return Vec::new(); + } + fuse_weighted(lists, 60.0) + .into_iter() + .take(limit) + .map(|f| { + let mut r = f.inner; + r.score = f.fusion_score as f32; + r.source_layer = CrateLayer::Hybrid; + r + }) + .collect() +} + /// Enrich search results with salience and usage data from Storage lookups. /// /// For each result, looks up the TocNode or Grip by doc_id and injects @@ -916,6 +1057,8 @@ mod tests { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await .unwrap(); @@ -942,6 +1085,8 @@ mod tests { limit: 10, agent_filter: None, all_projects: false, + rerank_mode: None, + expand_query: false, })) .await; @@ -1025,4 +1170,194 @@ mod tests { }; assert_eq!(proto_no_agent.agent, None); } + + #[test] + fn test_hybrid_fusion_differs_from_either_input() { + let bm25 = vec![ + SearchResult { + doc_id: "a".into(), + doc_type: "event".into(), + score: 0.99, + text_preview: "a".into(), + source_layer: CrateLayer::BM25, + metadata: HashMap::new(), + }, + SearchResult { + doc_id: "c".into(), + doc_type: "event".into(), + score: 0.50, + text_preview: "c".into(), + source_layer: CrateLayer::BM25, + metadata: HashMap::new(), + }, + ]; + let vector = vec![ + SearchResult { + doc_id: "b".into(), + doc_type: "event".into(), + score: 0.99, + text_preview: "b".into(), + source_layer: CrateLayer::Vector, + metadata: HashMap::new(), + }, + SearchResult { + doc_id: "a".into(), + doc_type: "event".into(), + score: 0.50, + text_preview: "a".into(), + source_layer: CrateLayer::Vector, + metadata: HashMap::new(), + }, + ]; + let fused = fuse_hybrid_lists(bm25, vector, 10); + let ids: Vec<&str> = fused.iter().map(|r| r.doc_id.as_str()).collect(); + assert_ne!(ids.as_slice(), &["a", "c"][..]); + assert_ne!(ids.as_slice(), &["b", "a"][..]); + assert_eq!(fused[0].doc_id, "a", "consensus doc should rank first"); + assert!(fused.iter().all(|r| r.source_layer == CrateLayer::Hybrid)); + } + + #[tokio::test] + async fn test_route_query_names_fusion_stage() { + let (handler, _temp) = create_test_handler(); + let response = handler + .route_query(Request::new(RouteQueryRequest { + query: "what is rust?".to_string(), + intent_override: None, + stop_conditions: None, + mode_override: None, + limit: 10, + agent_filter: None, + all_projects: false, + rerank_mode: None, + expand_query: false, + })) + .await + .unwrap() + .into_inner(); + let explanation = response.explanation.expect("explanation"); + assert_eq!(explanation.fusion_stage, "rank_fusion"); + assert_eq!(explanation.rerank_mode.as_deref(), Some("heuristic")); + assert!(explanation.why_winner.contains("rank_fusion")); + } + + struct ReverseCompleter; + + #[async_trait] + impl Completer for ReverseCompleter { + async fn complete(&self, prompt: &str) -> anyhow::Result { + // Reverse the candidate ids in the prompt so LLM order always + // differs from the RRF/BM25 order the heuristic path keeps. + let mut ids: Vec = prompt + .lines() + .filter_map(|line| { + line.split_once("id=") + .map(|(_, rest)| rest.trim().to_string()) + .filter(|id| !id.is_empty()) + }) + .collect(); + ids.reverse(); + Ok(format!(r#"{{"order":{}}}"#, serde_json::json!(ids))) + } + } + + #[tokio::test] + async fn test_llm_rerank_reorders_bm25_hits() { + use chrono::Utc; + use memory_search::{SearchIndex, SearchIndexConfig, SearchIndexer}; + use memory_types::{Event, EventRole, EventType}; + + let temp_dir = TempDir::new().unwrap(); + let index_path = temp_dir.path().join("search"); + std::fs::create_dir_all(&index_path).unwrap(); + let index = SearchIndex::open_or_create(SearchIndexConfig::new(&index_path)).unwrap(); + let indexer = SearchIndexer::new(&index).unwrap(); + indexer + .index_event(&Event::new( + "alpha-event".into(), + "s".into(), + Utc::now(), + EventType::UserMessage, + EventRole::User, + "zebra unique token alpha document".into(), + )) + .unwrap(); + indexer + .index_event(&Event::new( + "beta-event".into(), + "s".into(), + Utc::now(), + EventType::UserMessage, + EventRole::User, + "zebra unique token beta document".into(), + )) + .unwrap(); + indexer.commit().unwrap(); + let searcher = Arc::new(TeleportSearcher::new(&index).unwrap()); + + let storage = Arc::new(Storage::open(temp_dir.path()).unwrap()); + let heuristic = RetrievalHandler::with_services( + storage.clone(), + Some(searcher.clone()), + None, + None, + Default::default(), + ); + let llm_handler = RetrievalHandler::with_services( + storage, + Some(searcher), + None, + None, + Default::default(), + ) + .with_completer(Arc::new(ReverseCompleter)); + + let req = |mode: Option<&str>| RouteQueryRequest { + query: "zebra unique token".to_string(), + intent_override: None, + stop_conditions: None, + mode_override: None, + limit: 10, + agent_filter: None, + all_projects: false, + rerank_mode: mode.map(ToOwned::to_owned), + expand_query: false, + }; + + let heuristic_resp = heuristic + .route_query(Request::new(req(None))) + .await + .unwrap() + .into_inner(); + let llm_resp = llm_handler + .route_query(Request::new(req(Some("llm")))) + .await + .unwrap() + .into_inner(); + + assert!( + heuristic_resp.results.len() >= 2, + "need two BM25 hits to observe a reorder" + ); + let heuristic_ids: Vec<&str> = heuristic_resp + .results + .iter() + .map(|r| r.doc_id.as_str()) + .collect(); + let llm_ids: Vec<&str> = llm_resp.results.iter().map(|r| r.doc_id.as_str()).collect(); + assert_eq!( + llm_resp + .explanation + .as_ref() + .and_then(|e| e.rerank_mode.as_deref()), + Some("llm") + ); + assert_eq!(llm_ids.len(), heuristic_ids.len()); + assert_ne!( + heuristic_ids, llm_ids, + "LLM completer must reverse the heuristic/BM25 order; heuristic={heuristic_ids:?} llm={llm_ids:?}" + ); + // First of LLM should be last of heuristic (full reverse of two hits). + assert_eq!(llm_ids.first(), heuristic_ids.last()); + } } diff --git a/crates/memory-service/src/server.rs b/crates/memory-service/src/server.rs index 321df23..e01c940 100644 --- a/crates/memory-service/src/server.rs +++ b/crates/memory-service/src/server.rs @@ -16,7 +16,7 @@ use memory_scheduler::SchedulerService; use memory_storage::Storage; use memory_types::config::StalenessConfig; -use crate::ingest::MemoryServiceImpl; +use crate::ingest::{MemoryServiceImpl, QueryIndexBundle}; use crate::novelty::NoveltyChecker; use crate::pb::{memory_service_server::MemoryServiceServer, FILE_DESCRIPTOR_SET}; @@ -120,6 +120,7 @@ pub async fn run_server_with_scheduler( shutdown_signal: F, novelty_checker: Option>, staleness_config: StalenessConfig, + indexes: QueryIndexBundle, ) -> Result<(), Box> where F: std::future::Future + Send + 'static, @@ -149,10 +150,11 @@ where // Main service implementation with scheduler let mut memory_service = - MemoryServiceImpl::with_scheduler(storage, scheduler.clone(), staleness_config); + MemoryServiceImpl::with_scheduler(storage, scheduler.clone(), staleness_config.clone()); if let Some(checker) = novelty_checker { memory_service.set_novelty_checker(checker); } + memory_service.attach_indexes(indexes, staleness_config); info!("gRPC server ready on {}", addr); diff --git a/crates/memory-service/src/teleport_service.rs b/crates/memory-service/src/teleport_service.rs index 5816969..b7cb4e9 100644 --- a/crates/memory-service/src/teleport_service.rs +++ b/crates/memory-service/src/teleport_service.rs @@ -37,6 +37,8 @@ pub async fn handle_teleport_search( options = options.with_doc_type(DocType::TocNode); } else if req.doc_type == TeleportDocType::Grip as i32 { options = options.with_doc_type(DocType::Grip); + } else if req.doc_type == TeleportDocType::Event as i32 { + options = options.with_doc_type(DocType::Event); } // Execute search (blocking operation, use spawn_blocking) @@ -58,6 +60,7 @@ pub async fn handle_teleport_search( doc_type: match r.doc_type { DocType::TocNode => TeleportDocType::TocNode as i32, DocType::Grip => TeleportDocType::Grip as i32, + DocType::Event => TeleportDocType::Event as i32, }, score: r.score, keywords: r.keywords, diff --git a/crates/memory-service/src/vector.rs b/crates/memory-service/src/vector.rs index 022aca4..d26510a 100644 --- a/crates/memory-service/src/vector.rs +++ b/crates/memory-service/src/vector.rs @@ -6,6 +6,8 @@ use std::collections::HashMap; use std::sync::Arc; +use memory_types::recover_lock; + use tonic::{Request, Response, Status}; use tracing::{debug, info}; @@ -45,13 +47,13 @@ impl VectorTeleportHandler { /// Check if the vector index is available for search. pub fn is_available(&self) -> bool { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); index.len() > 0 } /// Get the current vector index status. pub fn get_status(&self) -> VectorIndexStatus { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); let stats = index.stats(); VectorIndexStatus { available: stats.available && stats.vector_count > 0, @@ -97,7 +99,7 @@ impl VectorTeleportHandler { // Search index let results = { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); index .search(&embedding, top_k) .map_err(|e| Status::internal(format!("Search failed: {}", e)))? @@ -165,7 +167,7 @@ impl VectorTeleportHandler { /// Returns a map of doc_id -> embedding vector. Missing entries are silently skipped. pub fn get_embeddings_for_doc_ids(&self, doc_ids: &[String]) -> HashMap> { let mut embeddings = HashMap::new(); - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); for doc_id in doc_ids { if let Ok(Some(entry)) = self.metadata.find_by_doc_id(doc_id) { if let Ok(Some(vector)) = index.get_vector(entry.vector_id) { @@ -199,7 +201,7 @@ impl VectorTeleportHandler { // Search index let results = { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); index .search(&embedding, limit) .map_err(|e| format!("Search failed: {}", e))? diff --git a/crates/memory-storage/src/usage.rs b/crates/memory-storage/src/usage.rs index 6700de4..5f05ef9 100644 --- a/crates/memory-storage/src/usage.rs +++ b/crates/memory-storage/src/usage.rs @@ -31,6 +31,7 @@ use crate::column_families::CF_USAGE_COUNTERS; use dashmap::DashMap; use lru::LruCache; +use memory_types::recover_lock; use memory_types::usage::{UsageConfig, UsageStats}; use rocksdb::{WriteBatch, DB}; use std::num::NonZeroUsize; @@ -98,7 +99,7 @@ impl UsageTracker { pub fn record_access(&self, doc_id: &str) { // Update cache immediately { - let mut cache = self.cache.lock().expect("cache mutex poisoned"); + let mut cache = recover_lock(self.cache.lock()); let stats = cache.get_or_insert_mut(doc_id.to_string(), UsageStats::new); stats.record_access(); } @@ -129,7 +130,7 @@ impl UsageTracker { pub fn get_usage_cached(&self, doc_id: &str) -> UsageStats { // Check cache first let cached = { - let mut cache = self.cache.lock().expect("cache mutex poisoned"); + let mut cache = recover_lock(self.cache.lock()); cache.get(doc_id).cloned() }; @@ -152,7 +153,7 @@ impl UsageTracker { let mut results = Vec::with_capacity(doc_ids.len()); { - let mut cache = self.cache.lock().expect("cache mutex poisoned"); + let mut cache = recover_lock(self.cache.lock()); for doc_id in doc_ids { if let Some(stats) = cache.get(doc_id) { results.push((doc_id.clone(), stats.clone())); @@ -264,7 +265,7 @@ impl UsageTracker { if let Some(bytes) = self.db.get_cf(&cf, doc_id.as_bytes())? { if let Ok(stats) = UsageStats::from_bytes(&bytes) { // Populate cache - let mut cache = self.cache.lock().expect("cache mutex poisoned"); + let mut cache = recover_lock(self.cache.lock()); cache.put(doc_id.clone(), stats); prefetched += 1; } @@ -294,7 +295,7 @@ impl UsageTracker { let mut loaded = 0u32; let iter = self.db.iterator_cf(&cf, rocksdb::IteratorMode::Start); - let mut cache = self.cache.lock().expect("cache mutex poisoned"); + let mut cache = recover_lock(self.cache.lock()); for item in iter.take(limit) { let (key, value) = item?; if let (Ok(doc_id), Ok(stats)) = ( @@ -314,7 +315,7 @@ impl UsageTracker { /// /// Returns (current_size, capacity). pub fn cache_stats(&self) -> (usize, usize) { - let cache = self.cache.lock().expect("cache mutex poisoned"); + let cache = recover_lock(self.cache.lock()); (cache.len(), cache.cap().get()) } diff --git a/crates/memory-toc/src/summarizer/api.rs b/crates/memory-toc/src/summarizer/api.rs index 331105d..2524e02 100644 --- a/crates/memory-toc/src/summarizer/api.rs +++ b/crates/memory-toc/src/summarizer/api.rs @@ -143,6 +143,14 @@ Guidelines: ) } + /// Complete an arbitrary prompt using the configured LLM API. + /// + /// Reuses the same retry / backoff / Anthropic-vs-OpenAI stack as + /// summarization. Used by the v3.1 LLM reranker. + pub async fn complete(&self, prompt: &str) -> Result { + self.call_api(prompt).await + } + /// Call the API with retry logic. async fn call_api(&self, prompt: &str) -> Result { let mut backoff = ExponentialBackoff { diff --git a/crates/memory-types/src/dedup.rs b/crates/memory-types/src/dedup.rs index da855a3..eab2886 100644 --- a/crates/memory-types/src/dedup.rs +++ b/crates/memory-types/src/dedup.rs @@ -100,9 +100,7 @@ impl InFlightBuffer { /// Clear all entries, resetting the buffer to empty. pub fn clear(&mut self) { - for entry in &mut self.entries { - *entry = None; - } + self.entries.fill(None); self.head = 0; self.count = 0; } diff --git a/crates/memory-types/src/lib.rs b/crates/memory-types/src/lib.rs index b7bcda3..55b458a 100644 --- a/crates/memory-types/src/lib.rs +++ b/crates/memory-types/src/lib.rs @@ -29,6 +29,7 @@ pub mod grip; pub mod outbox; pub mod salience; pub mod segment; +pub mod sync; pub mod toc; pub mod usage; @@ -49,5 +50,6 @@ pub use salience::{ SalienceScorer, }; pub use segment::Segment; +pub use sync::{recover_lock, LOCK_POISON_RECOVERIES}; pub use toc::{TocBullet, TocLevel, TocNode}; pub use usage::{usage_penalty, UsageConfig, UsageStats}; diff --git a/crates/memory-types/src/sync.rs b/crates/memory-types/src/sync.rs new file mode 100644 index 0000000..a5adf65 --- /dev/null +++ b/crates/memory-types/src/sync.rs @@ -0,0 +1,55 @@ +//! Lock-poisoning recovery. +//! +//! Workspace policy (v3.1 / Phase 54-06): never panic the daemon on a poisoned +//! `std::sync` mutex/rwlock. Recover the inner guard and count the event so it +//! is observable. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::LockResult; + +/// Number of times a poisoned lock was recovered process-wide. +pub static LOCK_POISON_RECOVERIES: AtomicU64 = AtomicU64::new(0); + +/// Recover from a poisoned `Mutex`/`RwLock` by taking the inner guard. +/// +/// Poisoning means a previous holder panicked. The data may be inconsistent, +/// but aborting the daemon is worse than continuing with a metric bump. +#[inline] +pub fn recover_lock(result: LockResult) -> T { + result.unwrap_or_else(|poisoned| { + LOCK_POISON_RECOVERIES.fetch_add(1, Ordering::Relaxed); + poisoned.into_inner() + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + use std::thread; + + #[test] + fn recover_lock_returns_guard_on_success() { + let m = Mutex::new(7); + let g = recover_lock(m.lock()); + assert_eq!(*g, 7); + } + + #[test] + fn recover_lock_recovers_from_poison() { + let m = Arc::new(Mutex::new(0)); + let m2 = Arc::clone(&m); + let _ = thread::spawn(move || { + let _g = m2.lock().unwrap(); + panic!("poison the mutex"); + }) + .join(); + + let before = LOCK_POISON_RECOVERIES.load(Ordering::Relaxed); + let mut g = recover_lock(m.lock()); + *g = 42; + drop(g); + assert!(LOCK_POISON_RECOVERIES.load(Ordering::Relaxed) > before); + assert_eq!(*recover_lock(m.lock()), 42); + } +} diff --git a/crates/memory-vector/src/hnsw.rs b/crates/memory-vector/src/hnsw.rs index c05a117..c0125ba 100644 --- a/crates/memory-vector/src/hnsw.rs +++ b/crates/memory-vector/src/hnsw.rs @@ -9,6 +9,7 @@ use std::path::PathBuf; use std::sync::RwLock; use memory_embeddings::Embedding; +use memory_types::recover_lock; use tracing::{debug, info}; use usearch::{Index, IndexOptions, MetricKind, ScalarKind}; @@ -126,7 +127,7 @@ impl HnswIndex { /// /// Returns `None` if the ID is not present in the index. pub fn get_vector(&self, id: u64) -> Result>, VectorError> { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); if !index.contains(id) { return Ok(None); } @@ -146,7 +147,7 @@ impl VectorIndex for HnswIndex { } fn len(&self) -> usize { - self.index.read().unwrap().size() + recover_lock(self.index.read()).size() } #[allow(clippy::readonly_write_lock)] // usearch::Index uses interior mutability @@ -158,7 +159,7 @@ impl VectorIndex for HnswIndex { }); } - let index = self.index.write().unwrap(); + let index = recover_lock(self.index.write()); index .add(id, &embedding.values) .map_err(|e| VectorError::Index(e.to_string()))?; @@ -175,7 +176,7 @@ impl VectorIndex for HnswIndex { }); } - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); let results = index .search(&query.values, k) .map_err(|e| VectorError::Index(e.to_string()))?; @@ -193,7 +194,7 @@ impl VectorIndex for HnswIndex { #[allow(clippy::readonly_write_lock)] // usearch::Index uses interior mutability fn remove(&mut self, id: u64) -> Result { - let index = self.index.write().unwrap(); + let index = recover_lock(self.index.write()); let result = index .remove(id) .map_err(|e| VectorError::Index(e.to_string()))?; @@ -207,12 +208,12 @@ impl VectorIndex for HnswIndex { } fn contains(&self, id: u64) -> bool { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); index.contains(id) } fn stats(&self) -> IndexStats { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); let size_bytes = std::fs::metadata(self.index_file()) .map(|m| m.len()) .unwrap_or(0); @@ -226,7 +227,7 @@ impl VectorIndex for HnswIndex { } fn save(&self) -> Result<(), VectorError> { - let index = self.index.read().unwrap(); + let index = recover_lock(self.index.read()); let path = self.index_file(); let path_str = path .to_str() @@ -256,7 +257,7 @@ impl VectorIndex for HnswIndex { .reserve(self.config.capacity) .map_err(|e| VectorError::Index(e.to_string()))?; - *self.index.write().unwrap() = new_index; + *recover_lock(self.index.write()) = new_index; info!("Cleared vector index"); Ok(()) } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 9cb05f4..da3e0de 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -216,19 +216,26 @@ memory-storage memory-toc memory-search memory-embeddings │ ### Indexing Pipeline ``` -1. Outbox consumer reads pending entries +1. Outbox consumer reads pending entries (IndexEvent / UpdateToc) │ 2. For each entry: - ├── Extract text from TOC node summaries and grips - ├── Index text in Tantivy (BM25) - ├── Generate embedding via Candle - └── Add embedding to usearch HNSW index + ├── Look up the event in RocksDB (fail-open + warn if missing) + ├── Index the event body in Tantivy (BM25, DocType::Event) + ├── If a grip exists for the event, index that too + ├── Generate embeddings via Candle for TOC nodes and grips + └── Add embeddings to usearch HNSW │ 3. Update checkpoint for crash recovery │ 4. Periodic compaction and optimization ``` +Skipped entries (event not found, empty text) log `tracing::warn!` and +increment `BM25_SKIPPED_NOOP`. They never report success while doing nothing. + +Live ingest writes `OutboxAction::UpdateToc` (and sometimes `IndexEvent`). +Both arms index the event so `ingest → drain outbox → BM25 teleport` is true. + ## Storage Schema ### Column Families @@ -253,8 +260,8 @@ Some indexes are managed outside RocksDB for specialized libraries: | Index | Location | Library | Purpose | |-------|----------|---------|---------| -| Tantivy BM25 | `{db_path}/tantivy/` | Tantivy | Full-text search with BM25 ranking | -| usearch HNSW | `{db_path}/usearch/` | usearch | Approximate nearest neighbor vectors | +| Tantivy BM25 | `{db_path}/search/` | Tantivy | Full-text search over events, TOC nodes, and grips | +| usearch HNSW | `{db_path}/vector/` | usearch | Approximate nearest neighbor vectors | RocksDB column families store metadata and pointers to these external directories. The indexing pipeline coordinates writes to ensure consistency between RocksDB and external indexes. diff --git a/proto/memory.proto b/proto/memory.proto index 2e97e73..d902d4f 100644 --- a/proto/memory.proto +++ b/proto/memory.proto @@ -539,6 +539,7 @@ enum TeleportDocType { TELEPORT_DOC_TYPE_UNSPECIFIED = 0; // Search all types TELEPORT_DOC_TYPE_TOC_NODE = 1; // TOC nodes only TELEPORT_DOC_TYPE_GRIP = 2; // Grips only + TELEPORT_DOC_TYPE_EVENT = 3; // Raw ingested events (Phase 54) } // Request for teleport search @@ -982,6 +983,11 @@ message RouteQueryRequest { // v3.0: Query across all registered project stores (opt-in, default false) // When true, fans out to all registered_projects and merges results. bool all_projects = 7; + // v3.1: Rerank mode for the orchestrator pipeline ("heuristic" | "llm"). + // Unset defaults to heuristic (no LLM cost). + optional string rerank_mode = 8; + // v3.1: Expand the query into variants before fan-out (default false). + bool expand_query = 9; } // A single retrieval result @@ -1011,6 +1017,11 @@ message ExplainabilityPayload { optional string fallback_reason = 8; uint64 total_time_ms = 9; repeated string grip_ids = 10; + // v3.1: Name of the fusion stage that produced these results + // (always "rank_fusion" when the orchestrator pipeline ran). + string fusion_stage = 11; + // v3.1: Rerank mode that was actually applied. + optional string rerank_mode = 12; } // Response from query routing