feat(recall): structured per-stage scores and two-level min_scores filtering

hindsight-api-slim/hindsight_api/api/http.py

@@ -154,6 +154,8 @@ def FieldWithDefault(default_factory: Callable, **kwargs) -> Any:
     VALID_RECALL_FACT_TYPES,
     DryRunExtractionResult,
     MemoryFact,
+    MinScores,
+    RecallScores,
     TokenUsage,
 )
 from hindsight_api.engine.search.tags import TagGroup, TagsMatch
@@ -312,6 +314,15 @@ class RecallRequest(BaseModel):
         description="Compound tag filter using boolean groups. Groups in the list are AND-ed. "
         "Each group is a leaf {tags, match} or compound {and: [...]}, {or: [...]}, {not: ...}.",
     )
+    min_scores: MinScores | None = Field(
+        default=None,
+        description="Optional per-stage score floors (all inclusive, AND-ed). `semantic` and `keyword` are "
+        "retrieval-level cutoffs pushed into the SQL arms (overriding the global similarity/BM25 minimums for "
+        "this request); `reranker` and `final` are post-ranking filters on the scored results. Any field left "
+        "unset imposes no floor; omitting `min_scores` entirely (the default) applies no score filtering. Use "
+        "with care — the reranker's absolute scores are not calibrated across queries (a clearly-relevant match "
+        "may score ~0.001 even though it is ranked first).",
+    )
 
     @field_validator("query")
     @classmethod
@@ -367,6 +378,7 @@ class RecallResult(BaseModel):
     source_fact_ids: list[str] | None = (
         None  # IDs of source facts (observation type only, when source_facts is enabled)
     )
+    scores: RecallScores | None = None  # Per-stage recall scores (final/reranker/semantic/text)
 
 
 class EntityObservationResponse(BaseModel):
@@ -3857,6 +3869,7 @@ async def api_recall(
                         tags=request.tags,
                         tags_match=request.tags_match,
                         tag_groups=request.tag_groups,
+                        min_scores=request.min_scores,
                     ),
                     operation="recall",
                     bank_id=bank_id,
@@ -3878,6 +3891,7 @@ def _fact_to_result(fact: "MemoryFact") -> RecallResult:
                     chunk_id=fact.chunk_id,
                     tags=fact.tags,
                     source_fact_ids=fact.source_fact_ids,
+                    scores=fact.scores,
                 )
 
             recall_results = [_fact_to_result(fact) for fact in core_result.results]

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -360,6 +360,8 @@ def validate_sql_schema(sql: str) -> None:
     EntityState,
     LLMCallTrace,
     MemoryFact,
+    MinScores,
+    RecallScores,
     ReflectResult,
     TokenUsage,
     ToolCallTrace,
@@ -3919,6 +3921,7 @@ async def recall_async(
         tag_groups: list[TagGroup] | None = None,
         created_after: datetime | None = None,
         created_before: datetime | None = None,
+        min_scores: MinScores | None = None,
         _connection_budget: int | None = None,
         _quiet: bool = False,
         reranking: RecallReranking = "cross_encoder",
@@ -4084,6 +4087,7 @@ async def recall_async(
                             tag_groups=tag_groups,
                             created_after=created_after,
                             created_before=created_before,
+                            min_scores=min_scores,
                             connection_budget=_connection_budget,
                             quiet=_quiet,
                             include_source_facts=include_source_facts,
@@ -4221,6 +4225,7 @@ async def _search_with_retries(
         tag_groups: list[TagGroup] | None = None,
         created_after: datetime | None = None,
         created_before: datetime | None = None,
+        min_scores: MinScores | None = None,
         connection_budget: int | None = None,
         quiet: bool = False,
         include_source_facts: bool = False,
@@ -4357,6 +4362,8 @@ async def _search_with_retries(
                         tag_groups=tag_groups,
                         created_after=created_after,
                         created_before=created_before,
+                        min_semantic=min_scores.semantic if min_scores else None,
+                        min_keyword=min_scores.keyword if min_scores else None,
                     )
                     parallel_duration = time.time() - parallel_start
             finally:
@@ -4723,6 +4730,30 @@ def to_tuple_format(results):
                 if strategy_boosts:
                     log_buffer.append(f"  [4.7] Strategy boosts applied: {strategy_boosts}")
 
+            # Step 4.9: post-query min_scores filters (reranker + final). The
+            # semantic/text floors are applied earlier inside the SQL arms (see
+            # retrieve_semantic_bm25_combined); here we apply the post-rank floors on
+            # the scored results, after the final sort and before truncation, so every
+            # downstream step (prefer_observations dedup, truncation, token filtering)
+            # operates on the filtered set. Inclusive (>=), AND-ed, opt-in: a None
+            # threshold is a no-op. There is deliberately no default — the
+            # cross-encoder's absolute scores are not calibrated for a fixed cutoff
+            # (a clearly-relevant match can score ~0.001 while its *ranking* is right).
+            min_reranker = min_scores.reranker if min_scores else None
+            min_final = min_scores.final if min_scores else None
+            if (min_reranker is not None or min_final is not None) and scored_results:
+                before_min_score = len(scored_results)
+                scored_results = [
+                    sr
+                    for sr in scored_results
+                    if (min_reranker is None or sr.cross_encoder_score_normalized >= min_reranker)
+                    and (min_final is None or sr.weight >= min_final)
+                ]
+                log_buffer.append(
+                    f"  [4.9] min_scores(reranker={min_reranker}, final={min_final}): "
+                    f"{before_min_score}->{len(scored_results)} results"
+                )
+
             # Add reranked results to tracer AFTER combined scoring (so normalized values are included)
             if tracer:
                 results_dict = [sr.to_dict() for sr in scored_results]
@@ -5134,6 +5165,25 @@ def _make_source_fact(sid: str, r: Any) -> MemoryFact:
                             )
 
             # Convert results to MemoryFact objects
+            # Build per-result scores (final/reranker/semantic/text) keyed by id.
+            # reranker is None when the configured reranker is a passthrough (rrf /
+            # interleave modes, or the RRFPassthroughCrossEncoder), since its
+            # cross_encoder_score_normalized is then a rank-derived placeholder, not a
+            # true relevance score.
+            ce_model = self._cross_encoder_reranker.cross_encoder
+            reranker_passthrough = (reranking != "cross_encoder") or (
+                ce_model is not None and getattr(ce_model, "provider_name", None) == "rrf"
+            )
+            scores_by_id: dict[str, RecallScores] = {
+                sr.id: RecallScores(
+                    final=sr.weight,
+                    reranker=None if reranker_passthrough else sr.cross_encoder_score_normalized,
+                    semantic=sr.candidate.arm_scores.semantic,
+                    keyword=sr.candidate.arm_scores.keyword,
+                )
+                for sr in top_scored
+            }
+
             memory_facts = []
             for result_dict in top_results_dicts:
                 result_id = str(result_dict.get("id"))
@@ -5157,6 +5207,7 @@ def _make_source_fact(sid: str, r: Any) -> MemoryFact:
                         chunk_id=result_dict.get("chunk_id"),
                         tags=result_dict.get("tags"),
                         source_fact_ids=source_fact_ids_by_obs.get(result_id) if include_source_facts else None,
+                        scores=scores_by_id.get(result_id),
                     )
                 )
 

hindsight-api-slim/hindsight_api/engine/response_models.py

@@ -172,6 +172,47 @@ class DispositionTraits(BaseModel):
     model_config = ConfigDict(json_schema_extra={"example": {"skepticism": 3, "literalism": 3, "empathy": 3}})
 
 
+class RecallScores(BaseModel):
+    """Per-result recall scores from different stages of the pipeline.
+
+    ``final`` is the value results are ranked by. The others are diagnostic and
+    can be filtered on via the recall ``min_scores`` request parameter. ``semantic``
+    and ``keyword`` are the raw per-strategy retrieval scores (``None`` when that
+    strategy did not surface this result); ``reranker`` is the cross-encoder's
+    normalized relevance.
+    """
+
+    final: float = Field(description="Final ranking score (combined reranker + recency/temporal/proof boosts)")
+    reranker: float | None = Field(
+        default=None,
+        description="Cross-encoder relevance, normalized 0-1. None when the reranker is a passthrough (rrf/interleave modes).",
+    )
+    semantic: float | None = Field(
+        default=None, description="Vector cosine similarity (0-1). None if this result was not surfaced semantically."
+    )
+    keyword: float | None = Field(
+        default=None,
+        description="Keyword/full-text (BM25) score (>= 0, unbounded). None if this result was not surfaced by keyword search.",
+    )
+
+
+class MinScores(BaseModel):
+    """Optional per-stage score floors for recall (all inclusive, AND-ed).
+
+    ``semantic`` and ``keyword`` are **retrieval-level** cutoffs pushed into the SQL
+    arms (overriding the global ``semantic_min_similarity`` / ``bm25_min_score``
+    config for this request), so they prune weak matches before fusion. ``reranker``
+    and ``final`` are **post-query** filters applied to the scored results after
+    reranking. Any field left None imposes no floor; all-None (the default) means
+    no score filtering.
+    """
+
+    semantic: float | None = Field(default=None, description="Retrieval-level: minimum vector similarity (0-1).")
+    keyword: float | None = Field(default=None, description="Retrieval-level: minimum keyword/full-text (BM25) score.")
+    reranker: float | None = Field(default=None, description="Post-query: minimum normalized reranker score (0-1).")
+    final: float | None = Field(default=None, description="Post-query: minimum final ranking score.")
+
+
 class MemoryFact(BaseModel):
     """
     A single memory fact returned by search or think operations.
@@ -231,6 +272,10 @@ def parse_metadata(cls, v: Any) -> dict[str, str] | None:
         None,
         description="IDs of source facts this observation was derived from (observation type only, when source_facts is enabled)",
     )
+    scores: RecallScores | None = Field(
+        None,
+        description="Recall scores from each pipeline stage (final/reranker/semantic/text). Not returned for source facts.",
+    )
 
 
 class ChunkInfo(BaseModel):

hindsight-api-slim/hindsight_api/engine/search/fusion.py

@@ -2,7 +2,7 @@
 Helper functions for hybrid search (semantic + BM25 + graph).
 """
 
-from .types import MergedCandidate, RetrievalResult
+from .types import ArmScores, MergedCandidate, RetrievalResult
 
 
 def cap_per_source(results: list[RetrievalResult], cap: int) -> list[RetrievalResult]:
@@ -51,6 +51,7 @@ def reciprocal_rank_fusion(result_lists: list[list[RetrievalResult]], k: int = 6
     rrf_scores = {}
     source_ranks = {}  # Track rank from each source for each doc_id
     all_retrievals = {}  # Store the actual RetrievalResult (use first occurrence)
+    arm_scores: dict[str, ArmScores] = {}  # doc_id -> raw per-strategy scores across arms
 
     source_names = ["semantic", "bm25", "graph", "temporal"]
 
@@ -79,17 +80,29 @@ def reciprocal_rank_fusion(result_lists: list[list[RetrievalResult]], k: int = 6
             if doc_id not in rrf_scores:
                 rrf_scores[doc_id] = 0.0
                 source_ranks[doc_id] = {}
+                arm_scores[doc_id] = ArmScores()
 
             rrf_scores[doc_id] += 1.0 / (k + rank)
             source_ranks[doc_id][f"{source_name}_rank"] = rank
 
+            # Capture this arm's raw score for the doc (the merged RetrievalResult
+            # below keeps only the first arm's score, so record each arm here).
+            if source_name == "semantic" and retrieval.similarity is not None:
+                arm_scores[doc_id].semantic = retrieval.similarity
+            elif source_name == "bm25" and retrieval.bm25_score is not None:
+                arm_scores[doc_id].keyword = retrieval.bm25_score
+
     # Combine into final results with metadata
     merged_results = []
     for rrf_rank, (doc_id, rrf_score) in enumerate(
         sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True), start=1
     ):
         merged_candidate = MergedCandidate(
-            retrieval=all_retrievals[doc_id], rrf_score=rrf_score, rrf_rank=rrf_rank, source_ranks=source_ranks[doc_id]
+            retrieval=all_retrievals[doc_id],
+            rrf_score=rrf_score,
+            rrf_rank=rrf_rank,
+            source_ranks=source_ranks[doc_id],
+            arm_scores=arm_scores[doc_id],
         )
         merged_results.append(merged_candidate)
 
@@ -118,6 +131,7 @@ def interleave_fusion(result_lists: list[list[RetrievalResult]]) -> list[MergedC
     source_names = ["semantic", "bm25", "graph", "temporal"]
     source_ranks: dict[str, dict[str, int]] = {}
     all_retrievals: dict[str, RetrievalResult] = {}
+    arm_scores: dict[str, ArmScores] = {}
 
     for source_idx, results in enumerate(result_lists):
         source_name = source_names[source_idx] if source_idx < len(source_names) else f"source_{source_idx}"
@@ -129,6 +143,11 @@ def interleave_fusion(result_lists: list[list[RetrievalResult]]) -> list[MergedC
             doc_id = retrieval.id
             all_retrievals.setdefault(doc_id, retrieval)
             source_ranks.setdefault(doc_id, {})[f"{source_name}_rank"] = rank
+            arm = arm_scores.setdefault(doc_id, ArmScores())
+            if source_name == "semantic" and retrieval.similarity is not None:
+                arm.semantic = retrieval.similarity
+            elif source_name == "bm25" and retrieval.bm25_score is not None:
+                arm.keyword = retrieval.bm25_score
 
     # Round-robin pick across arms in priority order: all #1s, then all #2s, ...
     ordered_ids: list[str] = []
@@ -151,6 +170,7 @@ def interleave_fusion(result_lists: list[list[RetrievalResult]]) -> list[MergedC
             rrf_score=float(n - pos),
             rrf_rank=pos + 1,
             source_ranks=source_ranks[doc_id],
+            arm_scores=arm_scores[doc_id],
         )
         for pos, doc_id in enumerate(ordered_ids)
     ]

hindsight-api-slim/hindsight_api/engine/search/reranking.py

@@ -177,6 +177,9 @@ def apply_combined_scoring(
         else:
             # Neutral baseline is precisely 0.5, ensuring neutral multiplier (1.0)
             proof_norm = 0.5
+        # Surface the proof signal so the trace can show the proof_count_boost
+        # factor (otherwise the reranked breakdown can't reconcile CE × boosts).
+        sr.proof_norm = proof_norm
 
         # RRF: kept at 0.0 for trace continuity but excluded from scoring.
         # RRF is batch-relative (min-max normalised) and redundant after reranking.

hindsight-api-slim/hindsight_api/engine/search/retrieval.py

@@ -104,6 +104,8 @@ async def retrieve_semantic_bm25_combined(
     tag_groups: list[TagGroup] | None = None,
     created_after: datetime | None = None,
     created_before: datetime | None = None,
+    min_semantic: float | None = None,
+    min_keyword: float | None = None,
 ) -> dict[str, tuple[list[RetrievalResult], list[RetrievalResult]]]:
     """
     Combined semantic + BM25 retrieval for multiple fact types in a single query.
@@ -143,6 +145,12 @@ async def retrieve_semantic_bm25_combined(
     config = get_config()
     tokens = tokenize_query(query_text)
 
+    # Per-request retrieval-level score floors (recall min_scores.semantic / .keyword)
+    # override the global config defaults for this query, pruning weak matches in
+    # the SQL arms before fusion.
+    sem_min = min_semantic if min_semantic is not None else config.semantic_min_similarity
+    bm25_min = min_keyword if min_keyword is not None else config.bm25_min_score
+
     # Over-fetch for HNSW approximation; semantic results trimmed to limit in Python.
     hnsw_fetch = max(limit * 5, 100)
 
@@ -203,7 +211,7 @@ async def retrieve_semantic_bm25_combined(
             embedding_param="$1",
             bank_id_param="$2",
             fetch_limit=hnsw_fetch,
-            min_similarity=config.semantic_min_similarity,
+            min_similarity=sem_min,
             tags_clause=tags_clause,
             groups_clause=groups_clause,
             extra_where=created_range_clause,
@@ -229,7 +237,7 @@ async def retrieve_semantic_bm25_combined(
                     arm_index=i,
                     text_search_extension=text_ext,
                     bm25_language=config.text_search_extension_native_language,
-                    bm25_min_score=config.bm25_min_score,
+                    bm25_min_score=bm25_min,
                     extra_where=created_range_clause,
                 )
             )
@@ -277,7 +285,7 @@ async def retrieve_semantic_bm25_combined(
                     embedding_param="$1",
                     bank_id_param="$2",
                     fetch_limit=hnsw_fetch,
-                    min_similarity=config.semantic_min_similarity,
+                    min_similarity=sem_min,
                     tags_clause=fb_tags_clause,
                     groups_clause=fb_groups_clause,
                     extra_where=fb_created_clause,
@@ -706,6 +714,8 @@ async def retrieve_all_fact_types_parallel(
     tag_groups: list[TagGroup] | None = None,
     created_after: datetime | None = None,
     created_before: datetime | None = None,
+    min_semantic: float | None = None,
+    min_keyword: float | None = None,
 ) -> MultiFactTypeRetrievalResult:
     """
     Optimized retrieval for multiple fact types using batched queries.
@@ -766,6 +776,8 @@ async def retrieve_all_fact_types_parallel(
             tag_groups=tag_groups,
             created_after=created_after,
             created_before=created_before,
+            min_semantic=min_semantic,
+            min_keyword=min_keyword,
         )
         semantic_bm25_time = time.time() - semantic_bm25_start
 
@@ -781,7 +793,7 @@ async def retrieve_all_fact_types_parallel(
                 tc_start,
                 tc_end,
                 budget=thinking_budget,
-                semantic_threshold=0.1,
+                semantic_threshold=min_semantic if min_semantic is not None else 0.1,
                 tags=tags,
                 tags_match=tags_match,
                 tag_groups=tag_groups,

hindsight-api-slim/hindsight_api/engine/search/tracer.py

@@ -392,7 +392,7 @@ def add_reranked(self, reranked_results: list[dict[str, Any]], rrf_merged: list)
 
             # Extract score components (only include non-None values)
             # Keys from ScoredResult.to_dict(): cross_encoder_score, cross_encoder_score_normalized,
-            # rrf_normalized, temporal, recency, combined_score, weight
+            # rrf_normalized, temporal, recency, proof_norm, combined_score, weight
             score_components = {}
             for key in [
                 "cross_encoder_score",
@@ -401,6 +401,7 @@ def add_reranked(self, reranked_results: list[dict[str, Any]], rrf_merged: list)
                 "rrf_normalized",
                 "temporal",
                 "recency",
+                "proof_norm",
                 "combined_score",
             ]:
                 if key in result and result[key] is not None: