poimen-memory/tasks/M8.5-hybrid-query-worker.md

# M8.5 — Hybrid query worker: parallel retrieval + fusion

| Field | Value |
|---|---|
| Phase | M8 — Hybrid Search |
| Size | L — 2–3 days |
| Status | ⬜ |
| Flags | — |
| Spec | inlined below |
| Blocks | M8.6, M8.7 |
| Depends | M8.1 (OpenSearch running), M8.2 (dual-write), M8.3 (query optimizer), M8.4 (RRF) |

## Goal

`HybridQueryWorker` orchestrates parallel retrieval from pgvector and OpenSearch, applies RRF fusion, and returns ranked results with score breakdown and latency metrics. It replaces `QueryWorker` as the primary query engine behind `GET /memory/query`.

## Design

**Execution flow:**
1. Call `QueryOptimizer::optimize_query()` → get `QueryContext` with `SearchStrategy`.
2. Generate embedding via `EmbeddingsClient::embed()`.
3. Execute strategy:
   - **Hybrid:** `tokio::try_join!` pgvector top-50 + OpenSearch top-50. Fuse with RRF.
   - **LexicalFirst:** OpenSearch top-200 → extract IDs → pgvector `WHERE id IN (...)` top-10.
   - **SemanticOnly:** pgvector top-50 (fallback if OpenSearch unavailable).
   - **LexicalOnly:** OpenSearch top-50 (fallback if embedding model unavailable).
4. Build response with score breakdown.

**Critical: use actual VectorStore API.**

The existing `VectorStore` exposes `search_l1(project, &embedding, limit)` and `search_l2(project, &embedding)`. The worker must call these — not invented methods.

For the cascading strategy (`LexicalFirst`), a new `search_l1_by_ids(project, &embedding, limit, &[chunk_id])` method is needed on `VectorStore`. This is a filtered pgvector query:
```sql
SELECT id, 1 - (embedding <=> $1) as score, text, source
FROM chunks
WHERE project = $2 AND id = ANY($3)
ORDER BY embedding <=> $1
LIMIT $4
```

**OpenSearch query:**
```json
{
  "size": 50,
  "query": {
    "bool": {
      "must": [{
        "multi_match": {
          "query": "...",
          "fields": ["content^2", "section_title^1.5", "breadcrumb", "source"],
          "type": "best_fields",
          "fuzziness": "AUTO"
        }
      }],
      "filter": [
        {"term": {"project_id": "..."}},
        {"terms": {"level": ["L0", "L1", "L2"]}}
      ]
    }
  }
}
```

**Response struct:**
```rust
pub struct HybridQueryResponse {
    pub query: String,
    pub project: String,
    pub search_strategy: String,
    pub results: Vec<HybridQueryResult>,
    pub metrics: QueryMetrics,
}

pub struct HybridQueryResult {
    pub id: String,
    pub text: String,
    pub source: String,
    pub level: String,
    pub breadcrumb: Vec<String>,
    pub final_score: f32,
    pub semantic_rank: Option<usize>,
    pub lexical_rank: Option<usize>,
    pub fusion_method: String,
}

pub struct QueryMetrics {
    pub total_time_ms: u128,
    pub semantic_time_ms: Option<u128>,
    pub lexical_time_ms: Option<u128>,
    pub fusion_time_ms: u128,
    pub semantic_candidates: Option<usize>,
    pub lexical_candidates: Option<usize>,
    pub final_count: usize,
}
```

## Steps

1. Add `search_l1_by_ids()` to `VectorStore` (new SQL query with `id = ANY($3)`).
2. Make `OpenSearchClient::lexical_search()` public.
3. Implement `HybridQueryWorker::new()` taking `VectorStore`, `EmbeddingsClient`, `Option<OpenSearchClient>`.
4. Implement `query()` method with strategy dispatch.
5. Implement `retrieve_hybrid()` using `tokio::try_join!`.
6. Implement `retrieve_cascading()` (2-stage).
7. Implement fallback methods (`retrieve_semantic()`, `retrieve_lexical()`).
8. Wire RRF fusion into the result pipeline.
9. Build response with per-result rank tracking.
10. Write integration tests with mock VectorStore + mock OpenSearch.

## Acceptance

1. Hybrid query returns results from **both** engines — check `semantic_rank` and `lexical_rank` are both `Some` for documents appearing in both lists.
2. Cascading query's pgvector call receives only IDs from the OpenSearch narrowing step — verify with query log or mock.
3. If `OpenSearchClient` is `None`, strategy automatically falls back to `SemanticOnly`.
4. If embedding generation fails, strategy falls back to `LexicalOnly` (if OpenSearch available) or returns error.
5. `metrics.total_time_ms` is populated and < 500ms for test fixtures.
6. `metrics.semantic_time_ms` and `lexical_time_ms` are roughly equal (parallel execution, not serial).
7. Results are sorted by `final_score` descending.
8. `final_count` ≤ `RRFConfig.final_k`.

## Verify

```bash
cargo test -p mem-cli hybrid_query -- --nocapture
```

**False pass:** Worker always returns semantic-only results even when strategy is `Hybrid`. Check by asserting `lexical_rank.is_some()` on at least one result when OpenSearch is configured. Another false pass: serial execution disguised as parallel — assert `max(semantic_time_ms, lexical_time_ms) ≈ total - fusion_time_ms`, not `sum`.

## Artifacts

- `crates/mem-cli/src/hybrid_query_worker.rs` (rewrite from current stub)
- Modified `crates/mem-store/src/lib.rs` (add `search_l1_by_ids`)
- Modified `crates/mem-cli/src/opensearch_client.rs` (make `lexical_search` pub)