Coverage for node / src / stigmem_node / routes / recall / ranking.py: 85%
64 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Recall scoring and packing stages."""
3from __future__ import annotations
5from ...auth import Identity
6from ...fact_visibility import caller_read_scope
7from ...models.facts import FactRecord
8from ...models.recall import RecallWeights, ScoreBreakdown, ScoredFact
9from ...plugins import get_registry
10from .common import _estimate_tokens, _recency_score
13def _filter_visible_gardens(
14 facts: dict[str, FactRecord], identity: Identity
15) -> dict[str, FactRecord]:
16 """Drop facts whose garden the caller cannot see (audit M3, defense-in-depth).
18 Applied to the candidate set of EVERY recall path (live and time-travel/as_of)
19 so the per-record garden check is a redundant backstop rather than the sole
20 gate. Uses the projected ``record.garden_id`` (consistent with the ranker) and
21 a single batched membership lookup via the shared read scope. No-op only when
22 the boundary is not enforced (fail-closed: it stays on once gardens exist).
23 """
24 scope = caller_read_scope(identity)
25 return {k: v for k, v in facts.items() if scope.garden_allows(v.garden_id)}
28def _score_candidates(
29 all_facts: dict[str, FactRecord],
30 lex_scores: dict[str, float],
31 sem_scores: dict[str, float],
32 graph_hops: dict[str, int],
33 weights: RecallWeights,
34 identity: Identity,
35 depth: int,
36) -> list[ScoredFact]:
37 """Compute composite score for each candidate fact."""
38 w = weights
39 total_weight = w.lexical + w.semantic + w.graph + w.recency
40 if total_weight <= 0:
41 total_weight = 1.0
43 results: list[ScoredFact] = []
45 # Garden ACL: resolve the caller's read scope ONCE, then filter in-memory per
46 # fact via the shared predicate (batched, audit M3 secure-path). Redundant
47 # backstop — the candidate set is pre-filtered by _filter_visible_gardens —
48 # but it keeps the ranker self-defending.
49 scope = caller_read_scope(identity)
51 for fact_id, record in all_facts.items():
52 # Skip quarantined / fully-redacted
53 if record.quarantine_status == "pending": 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true
54 continue
56 # Salience signal: contradiction-resolution status
57 contradiction_factor = 0.1 if record.contradicted else 1.0
59 # Salience signal: garden tier (quarantine garden = 0, normal = 1)
60 garden_factor = 1.0
61 if scope.enforce_gardens and record.garden_id is not None: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true
62 if not scope.garden_allows(record.garden_id):
63 continue # hidden by ACL
64 # Penalise quarantine-tagged gardens (§17)
65 garden_factor = 0.5
67 # Source-trust rank contribution is plugin-owned. Default installs keep
68 # this salience signal at 0 and can receive deltas from recall_rank.
69 st_score = 0.0
71 # Lexical signal
72 lex = lex_scores.get(fact_id, 0.0)
74 # Semantic signal
75 sem = sem_scores.get(fact_id, 0.0)
77 # Graph signal: inverse of hop distance (direct=0 → no graph bonus, 1 hop → 0.5, etc.)
78 hops = graph_hops.get(fact_id)
79 if hops is not None and hops > 0: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 graph_s = 1.0 / (1.0 + hops)
81 else:
82 graph_s = 0.0 if hops is None else 1.0
84 # Salience signal: recency
85 recency_s = _recency_score(record.timestamp)
87 # Salience signal: decay proxy (confidence itself encodes decay)
88 decay_factor = max(0.0, record.confidence)
90 # Weighted sum (normalised by sum of non-zero weights)
91 raw_total = (
92 w.lexical * lex
93 + w.semantic * sem
94 + w.graph * graph_s
95 + w.recency * recency_s
96 ) / total_weight
98 # Apply multiplicative adjustments
99 final_score = raw_total * decay_factor * contradiction_factor * garden_factor
101 breakdown = ScoreBreakdown(
102 lexical=round(lex, 4),
103 semantic=round(sem, 4),
104 graph=round(graph_s, 4),
105 source_trust=round(st_score, 4),
106 recency=round(recency_s, 4),
107 weighted_total=round(final_score, 6),
108 )
110 results.append(
111 ScoredFact(
112 fact=record,
113 score=round(final_score, 6),
114 score_breakdown=breakdown,
115 hop_distance=hops if hops is not None else 0,
116 token_estimate=_estimate_tokens(record),
117 )
118 )
120 source_deltas = get_registry().fire_score_delta(
121 "recall_rank",
122 results,
123 identity=identity,
124 weights=weights,
125 depth=depth,
126 )
127 if source_deltas:
128 adjusted: list[ScoredFact] = []
129 for scored in results:
130 delta = source_deltas.get(scored.fact.id, 0.0)
131 if delta == 0.0: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true
132 adjusted.append(scored)
133 continue
134 next_score = round(max(0.0, scored.score + delta), 6)
135 adjusted.append(
136 scored.model_copy(
137 update={
138 "score": next_score,
139 "score_breakdown": scored.score_breakdown.model_copy(
140 update={"weighted_total": next_score}
141 ),
142 }
143 )
144 )
145 results = adjusted
147 return results
150# ---------------------------------------------------------------------------
151# Token-budget greedy packing
152# ---------------------------------------------------------------------------
155def _greedy_pack(
156 candidates: list[ScoredFact],
157 token_budget: int,
158) -> tuple[list[ScoredFact], int, bool]:
159 """Sort by score desc; include facts until budget exhausted.
161 Returns (packed_facts, tokens_used, truncated).
162 """
163 candidates.sort(key=lambda c: c.score, reverse=True)
164 packed: list[ScoredFact] = []
165 tokens_used = 0
166 truncated = False
168 for candidate in candidates:
169 if tokens_used + candidate.token_estimate > token_budget:
170 truncated = True
171 continue # skip this fact (too large), keep trying smaller ones
172 packed.append(candidate)
173 tokens_used += candidate.token_estimate
175 # Secondary sort within packed: preserve score order
176 packed.sort(key=lambda c: c.score, reverse=True)
177 return packed, tokens_used, truncated