diff --git a/docs/adr-160-recall-profiles.md b/docs/adr-160-recall-profiles.md index 56037e2..4ca8a97 100644 --- a/docs/adr-160-recall-profiles.md +++ b/docs/adr-160-recall-profiles.md @@ -331,6 +331,92 @@ promotion, review state, or certification authority changes. packet shape is byte-for-byte unchanged (no `evidence` / `warning_codes` keys on legacy items). +## Supplement (issue #190, ENG-RECALL-003D): admission-first relationship expansion + +Issue #190 extends the governed/exploratory shadow profiles through the +existing bounded graph/tunnel relationship expansion, without reintroducing +the legacy blended trust/ranking model. Relationship expansion becomes an +**admission-first relevance mechanism**: a relationship can make a memory +*relevant*, it can never make that memory trusted, epistemically supported, +review-approved, or admissible. This remains shadow-only — it authorizes no +governed/exploratory production serving and no #162 certification/cutover. + +1. **Admission precedes expansion.** The candidate-profile pipeline order is + normative: direct semantic candidates → exact V2 admission on the direct + candidates → seeds chosen **only** from admitted direct candidates → + bounded graph/tunnel neighbor discovery → exact V2 admission on **every** + expanded neighbor → relationship-aware relevance → separated utility + ranking → budget packing. A withheld direct hit can never seed expansion + (zero admitted seeds means no expansion run at all), and seed admission + never transfers to a neighbor: each neighbor is admitted through the same + `decide_recall_admission` gate over its own resolved `risk_aware_shadow_v1` + decision, so only `current + exact-surface allow` enters the packet and + `missing | stale | mismatched | unsupported` and + `review_required | withhold | not_applicable` stay fail-closed. + +2. **Discovery under the same hard boundaries.** Neighbor discovery reuses + the shared bounded mechanics (depth-1 only, per-seed and total graph caps, + total tunnel cap, deterministic ordering/tie-breaking) and adds the same + live-proposal corpus window the profile's own direct retrieval applies + (`recall_signals.live_proposal_expression`) as the discovery prefilter — + the one prefilter the #190 contract permits precisely because it can + neither admit, widen, nor hide policy-relevant candidate state: it is the + identical window, and active/closed/conflicted rows are inevitably + withheld by the V2 gate anyway. Tenant/read-eligibility/workspace + boundaries are enforced inside discovery itself; the candidate tunnel + fetch orders `created_at desc, id asc` (importance-free) because utility + signals may order only already-admitted items, never decide which + neighbors a bounded window discovers. + +3. **Relationship relevance is versioned and utility-free.** One pure + contract (`relationship_recall.compute_relationship_relevance`, version + `relationship-relevance-v1`) computes relevance from exactly four inputs: + the item's direct semantic score (when it was a direct hit), the + source-seed relevance that justified expansion, the strongest graph edge + (bounded to `[0, 1]`, per-edge weights clamped), and tunnel membership. + Importance, source trust, memory confidence, human verification, review + state, exposure counters, and epistemic/risk state are not inputs — they + cannot move relevance by construction. The combination is + `clamp01(max(direct_score, w_semantic·semantic + w_graph·edge + + w_tunnel·tunnel))` over the existing relationship weights minus the + importance term: an unlinked direct item's relevance is exactly its + similarity (pre-#190 values), links never demote a direct hit, and a + relationship can only derive relevance from the bounded source-seed/ + relationship contract. Admitted items reached through expansion carry a + structured `relationship` block (origin decomposition, direct/seed + scores, edge types, tunnel labels, per-component contributions, + `relevance_score`); the final rank feeds that value into the unchanged + `compute_signal_rank_score` with utility — never a new blended scalar. + +4. **Evidence identity is untouched by relationships.** An admitted expanded + item presents the same `admission.v2` / `evidence` identity as any direct + item (the #188 invariant): `admission.v2` is the binding that admitted + *that neighbor*, `evidence.*` is its pure projection, top-level + `epistemic_state` mirrors it, and relationship metadata never rewrites + evidence. A `supports` edge cannot produce `supported`; a `contradicts` + edge is relevance, never conflict resolution — full conflict-preserving/ + diversity packing remains follow-up work. + +5. **Diagnostics and boundedness.** Withheld expanded neighbors remain + auditable and content-safe: `admission_diagnostics` entries carry the + expansion `origin` (`direct`, `graph`, `tunnel`, `graph+tunnel`) alongside + the V2 resolution status and exact surface decision, and each packet + carries a bounded `expansion` summary (contract version, seed/neighbor/ + admitted/withheld counts). Expansion performs no provider call and one + bounded bulk V2 resolution for the newly discovered neighbor set per + packet (query count constant in neighbor count, on top of the + already-bounded discovery queries); the packet-level `v2_resolution` + summary totals both windows under the same policy identity. + +6. **Legacy stays compatibility-only.** `expand_recall_candidates` (the + legacy blend, importance included, `semantic-v3`) remains byte-for-byte + unchanged and remains the only expansion `POST /v1/recall` runs. + `CERTIFIED_SERVING_PROFILES` stays `{"legacy"}`, the shadow surface stays + read-only and reviewer + tenant-policy gated, no MCP profile selection is + enabled, and #161 corroboration, semantic Context Ledger receipts, + dogfood ranking tuning, and #162 certification/cutover all remain + follow-up. + ## Feedback-loop safeguards (issue #160) * utility excludes exposure counters (`recall_count`, @@ -343,14 +429,16 @@ promotion, review state, or certification authority changes. ## Known limitations / follow-ups (issue #160 remains open) -This ADR records the #160 slice plus the #186 V2-binding and #188 -evidence-presentation supplements. +This ADR records the #160 slice plus the #186 V2-binding, #188 +evidence-presentation, and #190 admission-first relationship-expansion +supplements. Deliberately deferred, tracked by the issue: -* **Signal-aware graph/tunnel expansion** — expansion is legacy-only; - admission must precede expansion and the rescorer still speaks the blended - score. Governed/exploratory evaluate direct semantic hits only until - expansion learns the signal model. +* **Conflict-preserving/diversity packing over expanded candidates** — since + #190 the candidate profiles expand through graph/tunnel relationships + (admission-first, `relationship-relevance-v1`), but packing is still + purely rank-ordered: evidence-root grouping and conflict-pair preservation + over direct + expanded candidates are follow-up. * **#157 enrichment on non-V2 paths** — the V2 resolver supplies risk/epistemic/retention state for candidate admission in bulk (`effective_assessment_selection_bulk`), and since #188 the served items of @@ -375,12 +463,14 @@ Deliberately deferred, tracked by the issue: shadow-comparison surface (the reason it exists). Since #186 this includes keeping the V2 row corpus fresh: candidate packets bind only to *current* persisted decisions, so dogfooding pairs the #158 simulate+persist pass - with the #160 comparison. + with the #160 comparison. Since #190 the compared packets include + relationship expansion, so dogfood measures its contribution too. * **Certification/default cutover** — flipping `CERTIFIED_SERVING_PROFILES` (and then `recall_default_profile`) is gated on a *fresh* accepted #162 - certification of the now-integrated policy (V2-bound admission); #162D/#176 - terminated NOT_CERTIFIED against the pre-#186 integration and nothing here - authorizes #161/#162 production enablement. + certification of the now-integrated policy (V2-bound admission + + admission-first expansion); #162D/#176 terminated NOT_CERTIFIED against the + pre-#186 integration and nothing here authorizes #161/#162 production + enablement. * **`omitted_by_admission` is response-only** — gate-level withholding counts by reason code are returned to the caller and logged; `recall_logs` has no JSON omission column yet (note: under shadow-only rollout these counts diff --git a/engram/api/routes/recall_shadow.py b/engram/api/routes/recall_shadow.py index 27cf7c0..6909c8b 100644 --- a/engram/api/routes/recall_shadow.py +++ b/engram/api/routes/recall_shadow.py @@ -76,6 +76,9 @@ class RecallShadowPacket(BaseModel): ``admission_diagnostics`` and ``v2_resolution`` are the issue #186 binding: bounded, content-free per-candidate withhold diagnostics and the #158 V2 resolution summary. Legacy evaluates them to ``[]`` / ``None``. + ``expansion`` is the issue #190 addition: the bounded admission-first + relationship-expansion summary (contract version plus seed/neighbor/ + admission counts). Legacy evaluates it to ``None``. """ profile: str @@ -87,6 +90,7 @@ class RecallShadowPacket(BaseModel): omitted_by_admission: dict[str, int] admission_diagnostics: list[dict[str, Any]] = Field(default_factory=list) v2_resolution: dict[str, Any] | None = None + expansion: dict[str, Any] | None = None effective_byte_budget: int | None effective_token_budget: int | None effective_item_budget: int | None diff --git a/engram/recall.py b/engram/recall.py index bbedf07..1efde3b 100644 --- a/engram/recall.py +++ b/engram/recall.py @@ -26,16 +26,17 @@ from __future__ import annotations import logging +import math from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from uuid import UUID -from sqlalchemy import ColumnElement, and_, case, func, literal, or_, select, update +from sqlalchemy import ColumnElement, case, func, literal, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from engram import db as db_module -from engram import recall_signals, semantic +from engram import recall_signals, relationship_recall, semantic from engram.config import settings from engram.embeddings import generate_embedding from engram.jobs import enqueue_job @@ -995,23 +996,27 @@ def _enforce_semantic_budget( def _semantic_base_item_fields( item: MemoryItem, *, - distance: float, - similarity: float, + distance: float | None, + similarity: float | None, ) -> dict[str, Any]: """Per-item fields shared by every semantic profile's served items. Both the legacy blend path and the signal path build on this so the served-decision fields (ENG-CONTEXT-001) stay contract-aligned: a new field lands in one place and every profile serves it. Scoring/reasons/ - warnings differ per profile and are added by the caller. + warnings differ per profile and are added by the caller. ``distance``/ + ``similarity`` are ``None`` for items reached only through relationship + expansion (no query vector was ever compared against them — issue #190); + their relevance lives in ``relevance_score`` and the structured + ``relationship`` block. """ return { "id": str(item.id), "kind": item.kind, "content": item.content, "review_status": item.review_status, - "distance": round(distance, 4), - "similarity_score": round(similarity, 4), + "distance": round(distance, 4) if distance is not None else None, + "similarity_score": round(similarity, 4) if similarity is not None else None, "pinned": item.pinned, "importance": item.importance, "source_trust": item.source_trust, @@ -1025,24 +1030,6 @@ def _semantic_base_item_fields( } -def _live_proposal_expression() -> ColumnElement[bool]: - """The mechanically-expressible domain of the #158 V2 admission policy. - - ``risk_aware_shadow_v1`` decides over live proposals (an active, - superseded, or closed item is ``not_live`` and blocked on every surface), - and unresolved conflicts are blocked before any evidence is examined. - Both facts are pure SQL, so they are applied *before* the bounded HNSW - window (issue #186) — rows the V2 gate would inevitably withhold can - never occupy the candidate window and starve eligible proposals. - """ - return and_( - MemoryItem.review_status == "proposed", - MemoryItem.valid_to.is_(None), - MemoryItem.superseded_by.is_(None), - MemoryItem.conflict_resolution_status.is_distinct_from("unresolved"), - ) - - def _signal_corpus_eligibility( profile: RecallProfileSpec, stay_kinds: set[str], @@ -1053,14 +1040,16 @@ def _signal_corpus_eligibility( into SQL *before* the bounded HNSW window. Since issue #186 the signal profiles are V2-bound: their admission authority is the exact #158 per-surface decision, whose expressible domain is the live-proposal - corpus (:func:`_live_proposal_expression`) — everything else (risk, - epistemic state, calibration, observation windows, external disputes) is - the post-retrieval V2 gate's job. ``stay_kinds`` is retained for - signature compatibility with future corpus doctrines; the disputed - stay-kind doctrine is not part of the V2-bound window (disputed items are - never live proposals). + corpus (:func:`engram.recall_signals.live_proposal_expression`) — + everything else (risk, epistemic state, calibration, observation windows, + external disputes) is the post-retrieval V2 gate's job. ``stay_kinds`` is + retained for signature compatibility with future corpus doctrines; the + disputed stay-kind doctrine is not part of the V2-bound window (disputed + items are never live proposals). Issue #190 reuses this exact predicate + as the relationship-expansion discovery window so direct retrieval and + neighbor discovery can never disagree about the corpus. """ - return _live_proposal_expression() + return recall_signals.live_proposal_expression() @dataclass @@ -1069,12 +1058,30 @@ class SignalAdmissionOutcome: ``admission_diagnostics`` carries one bounded, content-free entry per withheld candidate — identity and codes only — so operators can see the - exact local-versus-V2 disagreement that withheld it (issue #186).""" + exact local-versus-V2 disagreement that withheld it (issue #186). Since + issue #190 each entry also names the ``origin`` that surfaced the + candidate (direct semantic hit vs graph/tunnel expansion), and + ``expansion`` summarizes the admission-first expansion run itself + (contract version, seed/neighbor/admission counts).""" items: list[dict[str, Any]] omitted_by_admission: dict[str, int] admission_diagnostics: list[dict[str, Any]] = field(default_factory=list) v2_resolution: dict[str, Any] | None = None + expansion: dict[str, Any] | None = None + + +@dataclass +class _AdmittedSignalItem: + """One admitted candidate item and the context its ranking consumed.""" + + item: MemoryItem + item_dict: dict[str, Any] + # Direct semantic similarity; None for items reached only through + # relationship expansion. + similarity: float | None + distance: float | None + created_ts: float async def _admit_and_rank_signal_items( @@ -1082,29 +1089,39 @@ async def _admit_and_rank_signal_items( *, profile: RecallProfileSpec, memory_context: ResolvedMemoryContext, + workspace_id: str | None, candidates: list[dict[str, Any]], item_by_id: dict[UUID, MemoryItem], stay_kinds: set[str], now: datetime, ) -> SignalAdmissionOutcome: - """V2-bound admission + separated-signal ranking (issues #160 / #186). + """V2-bound admission + separated-signal ranking (issues #160 / #186 / #190). Admission runs on the retrieved candidate window — after relevance retrieval (whose SQL already excluded mechanically-ineligible rows — see - :func:`_signal_corpus_eligibility`), before ranking, packing, and (by not - running at all for signal profiles) before any graph/tunnel expansion, so - nothing ineligible can enter the packet through a side door. - - The positive admission authority is the exact #158 ``risk_aware_shadow_v1`` + :func:`_signal_corpus_eligibility`) and before ranking and packing, so + nothing ineligible can enter the packet through a side door. The + positive admission authority is the exact #158 ``risk_aware_shadow_v1`` per-surface decision, resolved for the whole window by the shared bulk resolver (``admission_shadow.resolve_bulk_v2_decisions`` — the same evaluation the #158 simulator runs). Recall-local rules (the #159 - blocked/stale binding, lifecycle facts) can only withhold. Withheld items - are counted by reason code (``omitted_by_admission``) and itemized - content-free in ``admission_diagnostics``. - - Returns the admitted items sorted by signal rank plus the omission - counts, diagnostics, and V2 resolution summary. + blocked/stale binding, lifecycle facts) can only withhold. + + Since issue #190 the same ordering governs relationship expansion, which + runs strictly between direct admission and ranking: + + 1. exact V2 admission on the direct candidates; + 2. seeds are chosen only from admitted direct candidates; + 3. bounded graph/tunnel neighbor discovery under tenant/scope/RLS and + the same live-proposal corpus window; + 4. exact V2 admission on every expanded neighbor — seed admission never + transfers; + 5. relationship-aware relevance (versioned, importance-free) feeds the + separated utility ranking; budget packing follows unchanged. + + Withheld items (direct and expanded) are counted by reason code + (``omitted_by_admission``) and itemized content-free in + ``admission_diagnostics`` with their origin. """ items = list(item_by_id.values()) bindings = await recall_signals.load_admission_bindings( @@ -1123,7 +1140,7 @@ async def _admit_and_rank_signal_items( ) v2_summary = resolution.summary() - admitted: list[tuple[dict[str, Any], float, float]] = [] + admitted: list[_AdmittedSignalItem] = [] omitted: dict[str, int] = {} diagnostics: list[dict[str, Any]] = [] for cand in candidates: @@ -1167,15 +1184,349 @@ async def _admit_and_rank_signal_items( ) created = cand.get("created_at") created_ts = created.timestamp() if created is not None else 0.0 - admitted.append((item_dict, distance, created_ts)) + admitted.append( + _AdmittedSignalItem( + item=item, + item_dict=item_dict, + similarity=similarity, + distance=distance, + created_ts=created_ts, + ) + ) - # Deterministic order: signal rank desc, then closer vector, then newer. - admitted.sort(key=lambda entry: (-entry[0]["score"], entry[1], -entry[2])) + # Admission-first relationship expansion (issue #190). Only V2-bound + # candidate profiles expand — the legacy profile keeps its own + # compatibility expansion path in evaluate_semantic_profile below. + expansion_summary: dict[str, Any] | None = None + if settings.relationship_expansion_enabled and profile.v2_surface is not None and admitted: + expansion_run = await _expand_signal_candidates( + session, + profile=profile, + memory_context=memory_context, + workspace_id=workspace_id, + admitted=admitted, + direct_candidate_ids=set(item_by_id), + stay_kinds=stay_kinds, + now=now, + ) + admitted = expansion_run.admitted + for code, count in expansion_run.omitted_by_admission.items(): + omitted[code] = omitted.get(code, 0) + count + diagnostics.extend(expansion_run.admission_diagnostics) + expansion_summary = expansion_run.expansion + if expansion_run.v2_resolution is not None: + assert v2_summary is not None # the direct window resolved above + v2_summary = _merge_v2_resolution_summaries(v2_summary, expansion_run.v2_resolution) + + # Deterministic order: signal rank desc, then closer vector (direct hits + # ahead of expansion-only items at equal rank), then newer, then id. + admitted.sort( + key=lambda entry: ( + -entry.item_dict["score"], + entry.distance if entry.distance is not None else math.inf, + -entry.created_ts, + entry.item_dict["id"], + ) + ) + if expansion_summary is not None: + # The expansion path's ceiling — the same bound (and the same + # "applies whenever the expansion path runs, discoveries or not") + # the legacy expansion path applies after rescoring. + admitted = admitted[: settings.recall_candidate_ceiling] return SignalAdmissionOutcome( - items=[entry[0] for entry in admitted], + items=[entry.item_dict for entry in admitted], omitted_by_admission=omitted, admission_diagnostics=diagnostics, v2_resolution=v2_summary, + expansion=expansion_summary, + ) + + +def _merge_v2_resolution_summaries( + primary: dict[str, Any], neighbor: dict[str, Any] +) -> dict[str, Any]: + """Combine the direct-window and expanded-neighbor resolution summaries. + + Both resolutions ran the same policy artifact (same profile key/version/ + digest), so the merged summary simply totals the windows: counts and + queries add, status counts merge — the packet-level evidence that the + expanded neighbors were resolved by the same shared bulk resolver, never + a second policy or a per-neighbor lookup. + """ + status_counts: dict[str, int] = dict(primary["resolution_status_counts"]) + for status, count in neighbor["resolution_status_counts"].items(): + status_counts[status] = status_counts.get(status, 0) + count + return { + **primary, + "resolved_count": primary["resolved_count"] + neighbor["resolved_count"], + "resolution_status_counts": dict(sorted(status_counts.items())), + "query_count": primary["query_count"] + neighbor["query_count"], + } + + +def _expansion_origin( + item_id: UUID, discovery: relationship_recall.CandidateNeighborDiscovery +) -> str: + """Which expansion surface(s) reached this neighbor — the diagnostic + category that keeps direct withholds distinguishable from graph-, + tunnel-, and graph+tunnel-expanded withholds (issue #190). + + Callers only pass ids that came out of the discovery maps, so the + neither-map case is a contract break, not a policy outcome — fail loudly + rather than mislabel a withhold. + """ + in_graph = item_id in discovery.graph_links + in_tunnel = item_id in discovery.tunnel_links + if in_graph and in_tunnel: + return "graph+tunnel" + if in_graph: + return "graph" + if in_tunnel: + return "tunnel" + raise ValueError(f"expansion origin requested for an undiscovered item: {item_id}") + + +def _relationship_reason_lines( + discovery: relationship_recall.CandidateNeighborDiscovery, item_id: UUID +) -> list[str]: + """Human-readable relationship reasons, deterministic in output order.""" + lines: list[str] = [] + for link in sorted( + discovery.graph_links.get(item_id, []), + key=lambda link: (-link.weight, link.edge_type, str(link.neighbor_id)), + ): + reason = f"linked via {link.edge_type}" + if reason not in lines: + lines.append(reason) + for tlink in sorted( + discovery.tunnel_links.get(item_id, []), + key=lambda link: (link.tunnel_label, str(link.neighbor_id)), + ): + reason = f'same tunnel "{tlink.tunnel_label}"' + if reason not in lines: + lines.append(reason) + return lines + + +def _candidate_relevance( + *, + item_id: UUID, + similarity: float | None, + discovery: relationship_recall.CandidateNeighborDiscovery, + seed_similarity: dict[UUID, float], +) -> relationship_recall.RelationshipRelevance: + """Relationship-aware relevance for one linked candidate item. + + Source-seed attribution: graph and tunnel links alike carry the exact + admitted seed that reached the item (graph: the edge's seed endpoint; + tunnel: a seed whose tunnel membership exposed the (wing, room) the item + was pulled from), so ``source_seed_score`` is the strongest similarity + among the seeds that actually reached it — an unrelated stronger seed in + the packet contributes nothing. Both are relevance inputs only: nothing + here touches admission or evidence state. + """ + graph_links = discovery.graph_links.get(item_id, []) + tunnel_links = discovery.tunnel_links.get(item_id, []) + graph_seed_scores = [ + seed_similarity[link.seed_id] for link in graph_links if link.seed_id in seed_similarity + ] + tunnel_seed_scores = [ + seed_similarity[link.seed_id] for link in tunnel_links if link.seed_id in seed_similarity + ] + source_seed_score = max(graph_seed_scores + tunnel_seed_scores, default=0.0) + return relationship_recall.compute_relationship_relevance( + direct_semantic_score=similarity, + source_seed_score=source_seed_score, + graph_links=[(link.edge_type, link.weight) for link in graph_links], + tunnel_labels=[link.tunnel_label for link in tunnel_links], + ) + + +@dataclass +class _SignalExpansionOutcome: + """What one admission-first expansion run added to a signal packet.""" + + admitted: list[_AdmittedSignalItem] + omitted_by_admission: dict[str, int] + admission_diagnostics: list[dict[str, Any]] + v2_resolution: dict[str, Any] | None + expansion: dict[str, Any] + + +async def _expand_signal_candidates( + session: AsyncSession, + *, + profile: RecallProfileSpec, + memory_context: ResolvedMemoryContext, + workspace_id: str | None, + admitted: list[_AdmittedSignalItem], + direct_candidate_ids: set[UUID], + stay_kinds: set[str], + now: datetime, +) -> _SignalExpansionOutcome: + """Admission-first graph/tunnel expansion for one V2-bound packet. + + ``admitted`` are the direct candidates the exact V2 surface already + admitted, in retrieval order (the final signal-rank sort runs after + expansion) — only they may seed discovery. The whole direct candidate + window is passed as the enrichment set: already-evaluated direct items + linked to an admitted seed gain graph/tunnel origin metadata (no second + admission, no capacity cost), so a direct candidate — including one the + exact V2 surface withheld on the direct path — can never occupy a + bounded expansion slot. Newly discovered neighbors (outside that + window) are resolved through the same shared bulk resolver in ONE + bounded call and admitted through the same ``decide_recall_admission`` + gate: no per-neighbor lookup, no provider call, no second policy. Every + linked admitted item — enriched direct items and admitted neighbors + alike — is re-scored through the versioned relationship-relevance + contract (``relationship-relevance-v1``) feeding the unchanged + separated-utility rank; utility (importance/freshness) is computed by + the signal model exactly as for direct items and never enters relevance. + """ + seeds = admitted[: settings.recall_semantic_expansion_seed_limit] + seed_ids = [entry.item.id for entry in seeds] + seed_similarity = { + entry.item.id: entry.similarity for entry in seeds if entry.similarity is not None + } + + discovery = await relationship_recall.discover_candidate_neighbors( + session, + memory_context=memory_context, + workspace_id=workspace_id, + seed_ids=seed_ids, + seed_items=[entry.item for entry in seeds], + exclude_ids=direct_candidate_ids, + enrichment_ids=direct_candidate_ids, + ) + + new_neighbor_ids = (set(discovery.graph_links) | set(discovery.tunnel_links)) - set(seed_ids) + new_neighbor_ids -= direct_candidate_ids + + omitted: dict[str, int] = {} + diagnostics: list[dict[str, Any]] = [] + neighbor_summary: dict[str, Any] | None = None + new_entries: list[_AdmittedSignalItem] = [] + neighbor_items = [ + discovery.neighbor_items[item_id] + for item_id in sorted(new_neighbor_ids) + if item_id in discovery.neighbor_items + ] + if neighbor_items: + # One bounded bulk resolution for the whole newly discovered neighbor + # set (issue #190's no-N+1 contract): support + selection + one + # latest-row lookup, query count constant in neighbor count. + neighbor_bindings = await recall_signals.load_admission_bindings( + session, tenant_id=str(memory_context.tenant_id), items=neighbor_items + ) + from engram.admission_shadow import resolve_bulk_v2_decisions + + neighbor_resolution = await resolve_bulk_v2_decisions( + session, + items=neighbor_items, + context=memory_context, + evaluation_time=now, + ) + neighbor_summary = neighbor_resolution.summary() + for item in neighbor_items: + decision = recall_signals.decide_recall_admission( + item, + profile=profile, + stay_kinds=stay_kinds, + assessment=neighbor_bindings.get(item.id), + v2_resolution=neighbor_resolution.items.get(item.id), + ) + if decision.decision == "withhold": + code = decision.reason_codes[0] + omitted[code] = omitted.get(code, 0) + 1 + diagnostics.append( + _admission_diagnostic( + item, + profile=profile, + decision=decision, + assessment=neighbor_bindings.get(item.id), + origin=_expansion_origin(item.id, discovery), + ) + ) + continue + relevance = _candidate_relevance( + item_id=item.id, + similarity=None, + discovery=discovery, + seed_similarity=seed_similarity, + ) + item_dict = _semantic_base_item_fields(item, distance=None, similarity=None) + item_dict.update( + recall_signals.signal_item_fields( + item, decision=decision, now=now, relevance=relevance.relevance_score + ) + ) + item_dict["reasons"].extend(_relationship_reason_lines(discovery, item.id)) + item_dict["relationship"] = relevance.payload() + new_entries.append( + _AdmittedSignalItem( + item=item, + item_dict=item_dict, + similarity=None, + distance=None, + created_ts=( + item.created_at.timestamp() if item.created_at is not None else 0.0 + ), + ) + ) + + # Origin-merging enrichment: an admitted direct item (seed or not) that + # discovery linked keeps its direct fields (distance/similarity, + # admission, evidence — all unchanged) and gains the structured + # relationship block plus the merged relevance. Links can only raise a + # direct hit's relevance (max floor in the relevance contract), never + # demote it or touch its admission/evidence identity. + for entry in admitted: + linked = entry.item.id in discovery.graph_links or entry.item.id in discovery.tunnel_links + if not linked: + continue + relevance = _candidate_relevance( + item_id=entry.item.id, + similarity=entry.similarity, + discovery=discovery, + seed_similarity=seed_similarity, + ) + entry.item_dict["relevance_score"] = relevance.relevance_score + entry.item_dict["score"] = recall_signals.compute_signal_rank_score( + similarity=relevance.relevance_score, utility=entry.item_dict["utility_score"] + ) + # Refresh the relevance reason line signal_item_fields built — + # matched by its prefix rather than position so a reason reorder + # upstream can never rewrite the wrong line. + reasons = entry.item_dict["reasons"] + relevance_reason = f"relevance {relevance.relevance_score:.2f}" + for index, reason in enumerate(reasons): + if reason.startswith("relevance "): + reasons[index] = relevance_reason + break + else: + reasons.insert(0, relevance_reason) + entry.item_dict["reasons"].extend(_relationship_reason_lines(discovery, entry.item.id)) + entry.item_dict["relationship"] = relevance.payload() + + # Accounting describes genuinely new expansion candidates only: every + # count below excludes the already-evaluated direct window (seeds and + # withheld direct candidates alike), which received enrichment links at + # zero capacity cost and is never re-admitted here. + expansion_summary = { + "version": relationship_recall.RELATIONSHIP_RELEVANCE_VERSION, + "seed_count": len(seeds), + "discovered_neighbors": len(new_neighbor_ids), + "graph_neighbors": len(set(discovery.graph_links) - direct_candidate_ids), + "tunnel_neighbors": len(set(discovery.tunnel_links) - direct_candidate_ids), + "admitted_expanded": len(new_entries), + "withheld_expanded": len(diagnostics), + } + return _SignalExpansionOutcome( + admitted=admitted + new_entries, + omitted_by_admission=omitted, + admission_diagnostics=diagnostics, + v2_resolution=neighbor_summary, + expansion=expansion_summary, ) @@ -1185,6 +1536,7 @@ def _admission_diagnostic( profile: RecallProfileSpec, decision: recall_signals.RecallAdmissionDecision, assessment: recall_signals.AdmissionAssessmentBinding | None, + origin: str = "direct", ) -> dict[str, Any]: """One bounded, content-free withheld-candidate diagnostic. @@ -1201,6 +1553,10 @@ def _admission_diagnostic( would decide differently (unavailable V2 state is fail-closed unavailability, not a disagreement; the local hard gate is withhold-only, so "local would admit" means the lifecycle/#159 layers had no objection). + ``origin`` names what surfaced the candidate — ``direct`` for the + semantic window, ``graph`` / ``tunnel`` / ``graph+tunnel`` for items + reached through relationship expansion (issue #190), keeping the + withholds distinguishable for #162 evaluation. """ binding = decision.v2 v2_status = binding.resolution_status if binding is not None else "missing" @@ -1215,6 +1571,7 @@ def _admission_diagnostic( return { "item_id": str(item.id), "profile": profile.key, + "origin": origin, "decision": decision.decision, "reason_codes": list(decision.reason_codes), "surface": profile.v2_surface, @@ -1251,6 +1608,11 @@ class SemanticPacketEvaluation: # diagnostics for withheld candidates, and the resolution summary. admission_diagnostics: list[dict[str, Any]] = field(default_factory=list) v2_resolution: dict[str, Any] | None = None + # Admission-first relationship-expansion context (issue #190): the + # bounded summary of the expansion run (contract version, seed/neighbor/ + # admission counts). None on the legacy profile and on packets that did + # not expand. + expansion: dict[str, Any] | None = None def finalize_counts(self) -> None: self.item_count = len(self.items) @@ -1411,17 +1773,20 @@ async def evaluate_semantic_profile( omitted_by_admission: dict[str, int] = {} admission_diagnostics: list[dict[str, Any]] = [] v2_resolution_summary: dict[str, Any] | None = None + expansion_summary: dict[str, Any] | None = None if profile.signals_enabled: - # 4. V2-bound admission + separated-signal ranking (issues #160/#186). - # Relationship expansion deliberately does not run for signal - # profiles: admission must be enforced before graph/tunnel - # expansion, and the expansion rescorer still speaks the legacy - # blended score. Teaching expansion the signal model is follow-up - # work; until then these profiles serve only direct semantic hits. + # 4. V2-bound admission + separated-signal ranking (issues #160/#186), + # with admission-first relationship expansion (issue #190): only + # direct candidates the exact V2 surface admitted may seed bounded + # graph/tunnel discovery, every expanded neighbor is independently + # admitted through the same surface, and relationship-aware + # relevance (versioned, importance-free) feeds the separated + # utility ranking. Budget packing below is unchanged. admission_outcome = await _admit_and_rank_signal_items( session, profile=profile, memory_context=memory_context, + workspace_id=workspace_id, candidates=candidates, item_by_id=item_by_id, stay_kinds=stay_kinds, @@ -1431,6 +1796,7 @@ async def evaluate_semantic_profile( omitted_by_admission = admission_outcome.omitted_by_admission admission_diagnostics = admission_outcome.admission_diagnostics v2_resolution_summary = admission_outcome.v2_resolution + expansion_summary = admission_outcome.expansion else: # 4. Build per-item response dicts in trust-weighted order (legacy # profile — pre-#160 behavior, byte-for-byte). The candidate dicts @@ -1503,6 +1869,7 @@ async def evaluate_semantic_profile( item_budget=item_budget, admission_diagnostics=admission_diagnostics, v2_resolution=v2_resolution_summary, + expansion=expansion_summary, ) evaluation.finalize_counts() return evaluation diff --git a/engram/recall_shadow.py b/engram/recall_shadow.py index 569c189..b7ffb5e 100644 --- a/engram/recall_shadow.py +++ b/engram/recall_shadow.py @@ -89,8 +89,10 @@ def _packet_payload(evaluation: recall_module.SemanticPacketEvaluation) -> dict[ V2-bound candidate packets additionally carry the bounded admission diagnostics (one content-free entry per withheld candidate) and the V2 resolution summary — the operator-facing evidence of exactly which #158 - decisions admitted or withheld each candidate (issue #186). Legacy - evaluates to an empty diagnostics list / ``None`` summary. + decisions admitted or withheld each candidate (issue #186). Since issue + #190 each packet also carries the bounded relationship-expansion summary + (contract version, seed/neighbor/admission counts). Legacy evaluates to + an empty diagnostics list / ``None`` summaries. """ profile = evaluation.profile return { @@ -105,6 +107,7 @@ def _packet_payload(evaluation: recall_module.SemanticPacketEvaluation) -> dict[ "omitted_by_admission": dict(sorted(evaluation.omitted_by_admission.items())), "admission_diagnostics": evaluation.admission_diagnostics, "v2_resolution": evaluation.v2_resolution, + "expansion": evaluation.expansion, "effective_byte_budget": evaluation.byte_budget, "effective_token_budget": evaluation.token_budget, "effective_item_budget": evaluation.item_budget, diff --git a/engram/recall_signals.py b/engram/recall_signals.py index 9f260b0..fdd450b 100644 --- a/engram/recall_signals.py +++ b/engram/recall_signals.py @@ -44,6 +44,7 @@ from datetime import datetime from typing import Any, Final, Literal, Protocol, cast +from sqlalchemy import ColumnElement, and_ from sqlalchemy.ext.asyncio import AsyncSession from engram.admission_policy import AdmissionPolicyDecision @@ -60,6 +61,27 @@ # ``profile_key``); pinned against drift by the unit tests. V2_ADMISSION_PROFILE_KEY: Final[Literal["risk_aware_shadow_v1"]] = "risk_aware_shadow_v1" + +def live_proposal_expression() -> ColumnElement[bool]: + """The mechanically-expressible domain of the #158 V2 admission policy. + + ``risk_aware_shadow_v1`` decides over live proposals (an active, + superseded, or closed item is ``not_live`` and blocked on every surface), + and unresolved conflicts are blocked before any evidence is examined. + Both facts are pure SQL, so the V2-bound retrieval window applies them + *before* the bounded HNSW window (issue #186) — rows the V2 gate would + inevitably withhold can never occupy the candidate window and starve + eligible proposals. Issue #190 reuses the exact predicate as the + candidate-profile relationship-expansion discovery prefilter, so direct + retrieval and neighbor discovery can never disagree about the corpus. + """ + return and_( + MemoryItem.review_status == "proposed", + MemoryItem.valid_to.is_(None), + MemoryItem.superseded_by.is_(None), + MemoryItem.conflict_resolution_status.is_distinct_from("unresolved"), + ) + EpistemicState = Literal["supported", "contested", "insufficient_evidence", "unknown"] # ---- utility weights ---- @@ -809,8 +831,9 @@ def signal_item_fields( item: MemoryItem, *, decision: RecallAdmissionDecision, - similarity: float, + similarity: float | None = None, now: datetime, + relevance: float | None = None, ) -> dict[str, Any]: """Build the additive per-item signal fields for an admitted item. @@ -818,6 +841,15 @@ def signal_item_fields( the admission receipt) that ``execute_semantic_recall`` merges into the served item dict. No blended ``trust_score`` is produced or accepted here. + ``relevance`` is the item's final bounded relevance value. It defaults to + the direct semantic ``similarity``; relationship expansion (issue #190) + passes the versioned relationship-aware relevance for items reached + through graph/tunnel links (see + ``relationship_recall.compute_relationship_relevance``) — ranking + consumes the same value that is published as ``relevance_score``, so + every packet rank stays reproducible from the published relevance + + utility inputs. + Evidence authority (issue #188): on a V2-bound profile (``decision.v2`` present) the served epistemic state and the structured ``evidence`` block are exact projections of the already-bound V2 fresh @@ -827,6 +859,10 @@ def signal_item_fields( derivation (no candidate profile uses it today; it exists for legacy local-profile compatibility and is byte-stable). """ + if relevance is None: + if similarity is None: + raise ValueError("signal_item_fields requires similarity or relevance") + relevance = similarity utility = compute_utility_score( importance=item.importance, created_at=item.created_at, @@ -852,15 +888,15 @@ def signal_item_fields( assessment_status=decision.assessment_status, risk_state=risk_state, ) - rank = compute_signal_rank_score(similarity=similarity, utility=utility) + rank = compute_signal_rank_score(similarity=relevance, utility=utility) reasons = [ - f"relevance {similarity:.2f}", + f"relevance {relevance:.2f}", f"utility {utility:.2f}", f"admission {decision.profile}:{','.join(decision.reason_codes)}", ] fields = { "score": rank, - "relevance_score": round(similarity, 4), + "relevance_score": round(relevance, 4), "utility_score": utility, "epistemic_state": epistemic_state, "warning_codes": codes, @@ -933,6 +969,7 @@ async def load_admission_bindings( "compute_utility_score", "decide_recall_admission", "derive_epistemic_state", + "live_proposal_expression", "load_admission_bindings", "signal_item_fields", "structured_warning_codes", diff --git a/engram/relationship_recall.py b/engram/relationship_recall.py index 01aa00d..13158e9 100644 --- a/engram/relationship_recall.py +++ b/engram/relationship_recall.py @@ -10,40 +10,54 @@ "tunnel membership" is any tunnel whose source/target (wing, room) coordinates match the memory's own (wing, room). -Pipeline (called from engram.recall.execute_semantic_recall, between -semantic.search() and budget packing): +Two consumers share the bounded discovery mechanics: + +* **Legacy expansion** (:func:`expand_recall_candidates`, ENG-AUD-012) — the + compatibility path ``POST /v1/recall`` serves today: semantic candidates (already scored, sorted desc) -> bounded seed selection (recall_semantic_expansion_seed_limit) -> graph expansion (depth 1, bounded, deterministic) -> tunnel expansion (bounded, deterministic) -> merge (dedupe by id, track origin + relationship metadata) - -> relationship-aware rescoring + -> relationship-aware rescoring (legacy blend, importance included) -> ceiling truncation (recall_candidate_ceiling) -Every expanded candidate is re-filtered through the exact same trust -predicate semantic recall itself uses (tenant + read_eligibility_expression + -active/proposed review status + optional workspace scope) — expansion is -never an eligibility bypass. No recursive traversal: graph/tunnel neighbors -are found only for the original semantic seeds, never for neighbors of -neighbors. +* **Candidate-profile expansion** (issue #190 / ENG-RECALL-003D, + :func:`discover_candidate_neighbors` + + :func:`compute_relationship_relevance`) — the admission-first path the + governed/exploratory shadow profiles use. Discovery is the *same* bounded + depth-1 mechanics under the *same* hard boundaries, but it never scores or + admits anything: only V2-admitted direct items may seed it, every expanded + neighbor is admitted independently by the caller through the exact #158 + surface decision, and relevance is computed by the versioned pure helper + that excludes importance/trust/confidence/verification/review/exposure + inputs entirely (utility is applied separately, after admission). + +Every expanded candidate is re-filtered through read eligibility (tenant + +read_eligibility_expression + workspace scope + a review-window corpus +predicate) — expansion is never an eligibility bypass. No recursive +traversal: graph/tunnel neighbors are found only for the original seeds, +never for neighbors of neighbors. """ from __future__ import annotations from collections import defaultdict +from collections.abc import Sequence from dataclasses import dataclass, field from datetime import datetime -from typing import Any +from typing import Any, Final, Literal from uuid import UUID -from sqlalchemy import Select, or_, select +from sqlalchemy import ColumnElement, Select, or_, select from sqlalchemy.ext.asyncio import AsyncSession from engram.config import settings from engram.memory_access import read_eligibility_expression from engram.memory_context import ResolvedMemoryContext from engram.models import MemoryEdge, MemoryItem, Tunnel +from engram.recall_signals import live_proposal_expression # Recall-pipeline scoring version (distinct from engram.semantic.SEMANTIC_SCORING_VERSION, # which covers /v1/search's own ranking, unaffected by this module). Bumped @@ -52,6 +66,16 @@ # (requirement 6) — auditable via recall_logs.scoring_version. RECALL_SCORING_VERSION = "semantic-v3" +# Contract identity of the candidate-profile (V2-bound) relationship-relevance +# algorithm (issue #190 / ENG-RECALL-003D). Distinct from the legacy blend +# above: importance is excluded (utility, applied after admission), and +# graph/tunnel contribution is bounded to [0, 1]. Later #162 certification +# artifacts reference this version to identify exactly which algorithm +# produced a candidate packet's relevance values. +RELATIONSHIP_RELEVANCE_VERSION: Final[Literal["relationship-relevance-v1"]] = ( + "relationship-relevance-v1" +) + # Static edge_type -> strength mapping (requirement 9), used when an edge row # doesn't carry its own ``weight``. If the graph model gains real per-edge # weights later, MemoryEdge.weight already takes precedence over this table. @@ -84,12 +108,22 @@ class _GraphLink: neighbor_id: UUID edge_type: str weight: float + # The seed this link was reached from. Informational for the legacy blend + # (which never reads it); the candidate-profile path (issue #190) uses it + # for precise per-neighbor source-seed relevance attribution. + seed_id: UUID | None = None @dataclass class _TunnelLink: neighbor_id: UUID tunnel_label: str + # A seed whose tunnel membership exposed the (wing, room) this neighbor + # was pulled from (None on the frozen legacy path, whose blend never + # reads it). The candidate-profile path (issue #190) uses it for precise + # per-neighbor source-seed relevance attribution — mirrors + # ``_GraphLink.seed_id``. + seed_id: UUID | None = None @dataclass @@ -109,11 +143,27 @@ def _eligible_items_stmt( *, memory_context: ResolvedMemoryContext, workspace_id: str | None, + corpus_eligibility: ColumnElement[bool] | None = None, ) -> Select[tuple[MemoryItem]]: + """The complete neighbor predicate: tenant/read eligibility plus the + consumer's review-window corpus predicate. + + ``corpus_eligibility=None`` is the legacy window (active + proposed). + The candidate-profile path passes the V2 live-proposal predicate + (``recall_signals.live_proposal_expression``) — the exact window its own + direct retrieval uses, so discovery can neither admit, widen, nor hide + policy-relevant state the direct path would have shown (issue #190: a + prefilter is allowed only under that invariant; the exact V2 decision + stays authoritative). + """ stmt = select(MemoryItem).where( MemoryItem.id.in_(ids), MemoryItem.valid_to.is_(None), - MemoryItem.review_status.in_(_EXPANSION_REVIEW_STATUSES), + ( + corpus_eligibility + if corpus_eligibility is not None + else MemoryItem.review_status.in_(_EXPANSION_REVIEW_STATUSES) + ), read_eligibility_expression(memory_context), ) if workspace_id is not None: @@ -127,15 +177,29 @@ async def _fetch_graph_neighbors( memory_context: ResolvedMemoryContext, workspace_id: str | None, seed_ids: list[UUID], + corpus_eligibility: ColumnElement[bool] | None = None, + enrichment_ids: set[UUID] | None = None, ) -> dict[UUID, list[_GraphLink]]: """Depth-1, bounded, deterministic graph expansion. Returns eligible neighbor_id -> list of links (a neighbor can be reached via more than one edge/seed; all are kept for explainability, but scoring uses only the strongest — see :func:`_relationship_bonus`). + + ``enrichment_ids`` (the candidate-profile path passes the whole direct + candidate window, seeds included) splits two concerns the legacy path + never had: links to already-evaluated items are *enrichment* — recorded + for origin-merging at zero capacity cost, since the caller already + admission-evaluated them and never re-evaluates them here — while only + genuinely new neighbor ids compete for ``max_graph_neighbors_per_item`` + and ``max_graph_expanded_items``. Without that split, a direct candidate + (including one the exact V2 surface withheld) could occupy a bounded + graph slot purely because it was semantically near the query and suppress + a genuinely new expansion candidate (issue #190). """ if not seed_ids: return {} + enrichment = enrichment_ids or None seed_id_set = set(seed_ids) stmt = select(MemoryEdge).where( @@ -166,6 +230,7 @@ async def _fetch_graph_neighbors( potential_neighbor_ids, memory_context=memory_context, workspace_id=workspace_id, + corpus_eligibility=corpus_eligibility, ) ) ).scalars() @@ -179,6 +244,9 @@ async def _fetch_graph_neighbors( # just because it was already found semantically (requirement 5: origin # tags like "semantic+graph") — it simply doesn't count against the # max_graph_expanded_items budget below, since it's not a *new* addition. + # On the candidate-profile path (enrichment set present) that zero-cost + # treatment extends to every already-evaluated direct candidate, and the + # per-seed cap counts only genuinely new neighbors. per_seed: dict[UUID, list[tuple[float, str, UUID]]] = defaultdict(list) for edge in edges: weight = effective_edge_weight(edge.edge_type, edge.weight) @@ -190,9 +258,22 @@ async def _fetch_graph_neighbors( candidate_links: dict[UUID, list[_GraphLink]] = defaultdict(list) for seed_id in seed_ids: bucket = sorted(per_seed.get(seed_id, []), key=lambda t: (-t[0], t[1], str(t[2]))) - for weight, edge_type, neighbor_id in bucket[: settings.max_graph_neighbors_per_item]: + if enrichment is None: + selected = bucket[: settings.max_graph_neighbors_per_item] + else: + selected = [] + new_taken = 0 + for weight, edge_type, neighbor_id in bucket: + if neighbor_id in enrichment: + selected.append((weight, edge_type, neighbor_id)) + elif new_taken < settings.max_graph_neighbors_per_item: + selected.append((weight, edge_type, neighbor_id)) + new_taken += 1 + for weight, edge_type, neighbor_id in selected: candidate_links[neighbor_id].append( - _GraphLink(neighbor_id=neighbor_id, edge_type=edge_type, weight=weight) + _GraphLink( + neighbor_id=neighbor_id, edge_type=edge_type, weight=weight, seed_id=seed_id + ) ) if not candidate_links: @@ -201,35 +282,59 @@ async def _fetch_graph_neighbors( # Existing semantic seeds are enriched unconditionally (no budget cost — # they're already part of the result set). Only genuinely new neighbors # compete for the max_graph_expanded_items cap, strongest first - # (requirement 8: bounded graph additions). + # (requirement 8: bounded graph additions). With an enrichment set, the + # unconditional tier is the whole already-evaluated direct window; the + # caps then bound genuinely new neighbors only. linked_ids = set(candidate_links) - seed_neighbor_ids = linked_ids & seed_id_set + if enrichment is None: + enriched_ids = linked_ids & seed_id_set + capped_ids = linked_ids - seed_id_set + else: + enriched_ids = linked_ids & enrichment + capped_ids = linked_ids - enrichment new_neighbor_ids = sorted( - linked_ids - seed_id_set, + capped_ids, key=lambda nid: ( -max(link.weight for link in candidate_links[nid]), str(nid), ), )[: settings.max_graph_expanded_items] - return {nid: candidate_links[nid] for nid in (*seed_neighbor_ids, *new_neighbor_ids)} + return {nid: candidate_links[nid] for nid in (*enriched_ids, *new_neighbor_ids)} -async def _fetch_tunnel_neighbors( +@dataclass(frozen=True) +class _TunnelTarget: + """One (wing, room) target tunnels expose, with every seed that reaches it. + + ``label`` keeps the legacy dedup semantics (last matching seed/tunnel + write wins). ``seed_ids`` accumulates *all* seeds whose tunnel membership + exposes the target, because every item pulled from that target is + genuinely reachable from each of them — the provenance the + candidate-profile path binds into ``_TunnelLink.seed_id`` so tunnel + relevance attributes only seeds that actually reached an item, never an + unrelated packet-level best seed (issue #190). + """ + + label: str + seed_ids: frozenset[UUID] + + +async def _tunnel_targets( session: AsyncSession, *, memory_context: ResolvedMemoryContext, - workspace_id: str | None, seed_items: list[MemoryItem], - exclude_ids: set[UUID], -) -> dict[UUID, list[_TunnelLink]]: - """Bounded, deterministic tunnel expansion. +) -> dict[tuple[str, str | None], _TunnelTarget]: + """Resolve the (wing, room) targets tunnels expose for these seeds. A seed's tunnel membership is any ``Tunnel`` row whose source or target (wing, room) matches the seed's own (wing, room); the *other* endpoint of - that tunnel names the neighboring (wing, room) to pull bounded items from. - No full tunnel scan: each matched (wing, room) is fetched with its own - small LIMIT query, not a wing-wide table scan. + that tunnel names the neighboring (wing, room) to pull items from. + Returns ``(target_wing, target_room) -> label + the seeds that reach it``, + deduped across seeds/tunnels. Shared by the legacy and candidate-profile + fetchers so the two can never disagree about tunnel topology; the frozen + legacy fetcher ignores the seed attribution (its blend never reads it). """ wings = {item.wing for item in seed_items if item.wing} if not wings: @@ -247,8 +352,7 @@ async def _fetch_tunnel_neighbors( if not tunnels: return {} - # (target_wing, target_room) -> label, deduped across seeds/tunnels. - targets: dict[tuple[str, str | None], str] = {} + targets: dict[tuple[str, str | None], _TunnelTarget] = {} for item in seed_items: if not item.wing: continue @@ -257,20 +361,50 @@ async def _fetch_tunnel_neighbors( if tunnel.source_wing == item.wing and ( tunnel.source_room is None or tunnel.source_room == item.room ): - targets[(tunnel.target_wing, tunnel.target_room)] = label + key = (tunnel.target_wing, tunnel.target_room) + prior = targets.get(key) + targets[key] = _TunnelTarget( + label=label, + seed_ids=(prior.seed_ids if prior is not None else frozenset()) + | {item.id}, + ) if tunnel.target_wing == item.wing and ( tunnel.target_room is None or tunnel.target_room == item.room ): - targets[(tunnel.source_wing, tunnel.source_room)] = label + key = (tunnel.source_wing, tunnel.source_room) + prior = targets.get(key) + targets[key] = _TunnelTarget( + label=label, + seed_ids=(prior.seed_ids if prior is not None else frozenset()) + | {item.id}, + ) + return targets + + +async def _fetch_tunnel_neighbors( + session: AsyncSession, + *, + memory_context: ResolvedMemoryContext, + workspace_id: str | None, + seed_items: list[MemoryItem], + exclude_ids: set[UUID], +) -> dict[UUID, list[_TunnelLink]]: + """Bounded, deterministic tunnel expansion (legacy window/ordering). + Each matched (wing, room) is fetched with its own small LIMIT query, not + a wing-wide table scan. Discovery order is importance-first — a utility + ordering the legacy compatibility path has always used. + """ + targets = await _tunnel_targets(session, memory_context=memory_context, seed_items=seed_items) if not targets: return {} candidate_links: dict[UUID, list[_TunnelLink]] = defaultdict(list) remaining = settings.max_tunnel_additions - for (target_wing, target_room), label in sorted( + for (target_wing, target_room), target in sorted( targets.items(), key=lambda kv: (kv[0][0], kv[0][1] or "") ): + label = target.label if remaining <= 0: break filters: list[Any] = [ @@ -305,6 +439,134 @@ async def _fetch_tunnel_neighbors( return dict(candidate_links) +def _add_candidate_tunnel_links( + links_by_neighbor: dict[UUID, list[_TunnelLink]], + neighbor_id: UUID, + label: str, + seed_ids: list[UUID], +) -> bool: + """Record one tunnel link per reaching seed, deduped by (label, seed_id). + + Returns whether the neighbor was linked for the first time — the signal + the bounded new-neighbor budget keys on: re-seeing an already-linked + neighbor through another target adds source attribution, never budget. + """ + links = links_by_neighbor[neighbor_id] + newly_linked = not links + for seed_id in seed_ids: + link = _TunnelLink(neighbor_id=neighbor_id, tunnel_label=label, seed_id=seed_id) + if link not in links: + links.append(link) + return newly_linked + + +async def _fetch_candidate_tunnel_neighbors( + session: AsyncSession, + *, + memory_context: ResolvedMemoryContext, + workspace_id: str | None, + seed_items: list[MemoryItem], + exclude_ids: set[UUID], + enrichment_ids: set[UUID] | None = None, +) -> dict[UUID, list[_TunnelLink]]: + """Bounded, deterministic tunnel discovery for V2-bound profiles (#190). + + Same tunnel topology and the same hard boundaries (tenant/read + eligibility, explicit workspace restriction, live-proposal corpus + window, per-target and total caps), but the discovery order is + importance-free (``created_at desc, id asc``): utility signals may order + only already-admitted items, never influence which neighbors a candidate + profile discovers within its bounded window. + + Source-seed attribution (the tunnel analogue of ``_GraphLink.seed_id``): + every neighbor pulled from a target (wing, room) is linked once per + admitted seed whose tunnel membership exposed that target, deduped by + (label, seed_id) — so relationship relevance can attribute exactly the + seeds that actually reached the item, never an unrelated packet-level + best seed (issue #190). + + ``enrichment_ids`` (the whole direct candidate window) receives the same + direct-vs-new split the graph fetcher applies: already-evaluated direct + items sitting in a tunneled (wing, room) collect tunnel-origin metadata + for origin-merging at zero budget cost — no second admission, no + consumption of the new-neighbor tunnel caps — while genuinely new + neighbors keep competing inside the bounded per-target/total windows + (issue #190: ``semantic+tunnel`` and ``semantic+graph+tunnel`` origins + must be representable). Enrichment queries apply the identical + eligibility predicates, so tunnel visibility is never broadened. + + Deliberately a near-sibling of :func:`_fetch_tunnel_neighbors` rather + than a parameterized shared helper: the legacy fetcher is frozen for + byte-compatibility, and its importance-first ordering is exactly the + utility ordering the candidate contract forbids — the two policies are + kept in sibling functions so neither can regress the other. + """ + targets = await _tunnel_targets(session, memory_context=memory_context, seed_items=seed_items) + if not targets: + return {} + + candidate_links: dict[UUID, list[_TunnelLink]] = defaultdict(list) + remaining = settings.max_tunnel_additions + for (target_wing, target_room), target in sorted( + targets.items(), key=lambda kv: (kv[0][0], kv[0][1] or "") + ): + label = target.label + # Deterministic attribution order: seed ids ascending. + target_seed_ids = sorted(target.seed_ids) + if enrichment_ids: + # Direct-item enrichment: bounded by the already-evaluated direct + # window itself, so it needs no LIMIT and never touches the + # new-neighbor budget. Same predicates as the discovery query — + # enrichment is not an eligibility bypass. + enrich_filters: list[Any] = [ + MemoryItem.id.in_(enrichment_ids), + MemoryItem.wing == target_wing, + live_proposal_expression(), + read_eligibility_expression(memory_context), + ] + if target_room is not None: + enrich_filters.append(MemoryItem.room == target_room) + if workspace_id is not None: + enrich_filters.append(MemoryItem.workspace_id == workspace_id) + enrich_stmt = ( + select(MemoryItem) + .where(*enrich_filters) + .order_by(MemoryItem.created_at.desc(), MemoryItem.id.asc()) + ) + for row in (await session.execute(enrich_stmt)).scalars().all(): + _add_candidate_tunnel_links(candidate_links, row.id, label, target_seed_ids) + if remaining <= 0: + # Enrichment is budget-free, so later targets still get their + # enrichment pass; with no enrichment set the rest of this loop + # body is a no-op from here on, exactly like the legacy break. + continue + filters: list[Any] = [ + MemoryItem.wing == target_wing, + live_proposal_expression(), + read_eligibility_expression(memory_context), + ] + if target_room is not None: + filters.append(MemoryItem.room == target_room) + if exclude_ids: + filters.append(MemoryItem.id.notin_(exclude_ids)) + if workspace_id is not None: + filters.append(MemoryItem.workspace_id == workspace_id) + + stmt = ( + select(MemoryItem) + .where(*filters) + .order_by(MemoryItem.created_at.desc(), MemoryItem.id.asc()) + .limit(min(settings.max_tunnel_neighbors_per_item, remaining)) + ) + + rows = list((await session.execute(stmt)).scalars().all()) + for row in rows: + if _add_candidate_tunnel_links(candidate_links, row.id, label, target_seed_ids): + remaining -= 1 + + return dict(candidate_links) + + def _relationship_bonus(links: list[_GraphLink]) -> float: """Strongest edge wins — a node with many weak edges shouldn't outscore one strong, directly relevant edge (requirement 11).""" @@ -357,6 +619,229 @@ def _origin_label(origins: set[str]) -> str: return "+".join(sorted(origins, key=lambda o: {"semantic": 0, "graph": 1, "tunnel": 2}[o])) +# ---- candidate-profile relationship relevance (issue #190 / ENG-RECALL-003D) ---- +# +# Relationship is a *relevance* signal only. It can never make a memory +# trusted, epistemically supported, review-approved, or admissible — those +# are the exact V2 surface decision's job, resolved independently per item. + + +def _clamp01(value: float) -> float: + return max(0.0, min(1.0, value)) + + +@dataclass(frozen=True) +class RelationshipRelevance: + """Structured, versioned relationship-aware relevance for one item. + + The deterministic explanation block candidate-profile items carry when + relationship expansion reached them (issue #190): origin decomposition, + the direct semantic relevance when the item was also a direct hit, the + source-seed relevance that justified expansion when it was not, the + strongest graph contribution with its edge types, tunnel labels, the + per-component contributions, and the final bounded ``relevance_score`` + the candidate profile ranks with. + """ + + version: str + origins: tuple[str, ...] + direct: bool + direct_semantic_score: float | None + source_seed_score: float + graph_contribution: float + graph_edge_types: tuple[str, ...] + tunnel_labels: tuple[str, ...] + relevance_score: float + components: dict[str, float] + + def payload(self) -> dict[str, Any]: + return { + "version": self.version, + "origins": list(self.origins), + "direct": self.direct, + "direct_semantic_score": ( + round(self.direct_semantic_score, 4) + if self.direct_semantic_score is not None + else None + ), + "source_seed_score": round(self.source_seed_score, 4), + "graph_contribution": round(self.graph_contribution, 4), + "graph_edge_types": list(self.graph_edge_types), + "tunnel_labels": list(self.tunnel_labels), + "relevance_score": self.relevance_score, + "components": {key: round(value, 4) for key, value in self.components.items()}, + } + + +def compute_relationship_relevance( + *, + direct_semantic_score: float | None, + source_seed_score: float, + graph_links: Sequence[tuple[str, float]] = (), + tunnel_labels: Sequence[str] = (), +) -> RelationshipRelevance: + """The one pure, versioned relationship-relevance helper (issue #190). + + Combines the item's own direct semantic relevance (when it was a direct + hit) with the bounded graph/tunnel contribution derived from the seeds + that reached it, reusing the relationship weighting configuration with + the importance term removed:: + + blend = w_semantic * semantic + w_graph * strongest_edge + + w_tunnel * tunnel_membership + relevance = clamp01(max(direct_score or 0, blend)) + + Invariants (mechanically pinned by unit tests): + + * importance, source trust, memory confidence, human verification, + review state, exposure counters, and epistemic/risk state are not + inputs — they cannot move relationship relevance by construction; + * graph/tunnel contribution is bounded to ``[0, 1]`` (per-edge weights + are clamped, so a stored weight > 1 cannot inflate it); + * a direct hit's relevance is at least its own semantic similarity + (``max`` floor) — being additionally relationship-linked can never + demote it, and an unlinked direct item's relevance is exactly its + similarity, byte-compatible with the pre-#190 signal contract; + * a relationship can only derive relevance from the source-seed score + and its own bounded contribution — never create relevance outside + that contract; + * identical inputs produce identical relevance (deterministic, and the + weight sum w_semantic + w_graph + w_tunnel < 1 keeps the blend in + range even at maximal inputs). + """ + graph_links = tuple(graph_links) + tunnel_labels = tuple(tunnel_labels) + graph_contribution = max((_clamp01(weight) for _, weight in graph_links), default=0.0) + tunnel_membership = 1.0 if tunnel_labels else 0.0 + semantic_component = ( + direct_semantic_score if direct_semantic_score is not None else source_seed_score + ) + components = { + "semantic": settings.relationship_score_weight_semantic * semantic_component, + "graph": settings.relationship_score_weight_relationship * graph_contribution, + "tunnel": settings.relationship_score_weight_tunnel * tunnel_membership, + } + blend = sum(components.values()) + direct_floor = direct_semantic_score if direct_semantic_score is not None else 0.0 + relevance_score = round(_clamp01(max(direct_floor, blend)), 4) + + origins: list[str] = [] + if direct_semantic_score is not None: + origins.append("semantic") + if graph_links: + origins.append("graph") + if tunnel_labels: + origins.append("tunnel") + return RelationshipRelevance( + version=RELATIONSHIP_RELEVANCE_VERSION, + origins=tuple(origins), + direct=direct_semantic_score is not None, + direct_semantic_score=direct_semantic_score, + source_seed_score=source_seed_score, + graph_contribution=graph_contribution, + graph_edge_types=tuple(sorted({edge_type for edge_type, _ in graph_links})), + tunnel_labels=tuple(sorted(set(tunnel_labels))), + relevance_score=relevance_score, + components=components, + ) + + +# ---- candidate-profile neighbor discovery (issue #190) ---- + + +@dataclass +class CandidateNeighborDiscovery: + """Bounded graph+tunnel neighbor discovery for one candidate packet. + + ``graph_links`` / ``tunnel_links`` include already-evaluated direct + candidates reachable from an admitted seed (origin-merging enrichment — + zero capacity cost, never re-admitted). Every link carries the admitted + seed it was reached from (``seed_id`` — the exact provenance + relationship relevance attributes, never a packet-level best seed). + ``neighbor_items`` carries the backing ``MemoryItem`` rows for every + *genuinely new* discovered id only + (never the seeds or the enrichment window) — fetched in one bounded bulk + query with read-eligibility defense in depth. Discovery confers no + admission: every discovered neighbor must be independently admitted by + the caller through the exact V2 surface. + """ + + graph_links: dict[UUID, list[_GraphLink]] + tunnel_links: dict[UUID, list[_TunnelLink]] + neighbor_items: dict[UUID, MemoryItem] + + +async def discover_candidate_neighbors( + session: AsyncSession, + *, + memory_context: ResolvedMemoryContext, + workspace_id: str | None, + seed_ids: list[UUID], + seed_items: list[MemoryItem], + exclude_ids: set[UUID], + enrichment_ids: set[UUID] | None = None, +) -> CandidateNeighborDiscovery: + """Admission-first bounded neighbor discovery for a V2-bound profile. + + Runs AFTER the caller admitted the direct candidates (the seeds) through + the exact V2 surface — a withheld direct hit can never seed discovery + because it is never passed in here. Mechanics are the shared bounded + graph/tunnel core: depth-1 only, per-seed and total graph caps, total + tunnel cap, same-tenant edges/tunnels only, read eligibility, explicit + workspace restriction with no unscoped fallback, and the V2 + live-proposal corpus window as the discovery prefilter (the exact window + the profile's own direct retrieval uses — it can never widen or hide + policy-relevant candidate state). + + ``exclude_ids`` (the direct candidate window) are excluded from tunnel + fetches and from the returned neighbor rows: they were already + admission-evaluated as direct candidates, so expansion never + re-evaluates or double-diagnoses them. ``enrichment_ids`` (normally the + same direct window) additionally lets those already-evaluated items + collect graph/tunnel origin metadata from admitted seeds — the + direct-vs-new split that keeps every bounded window reserved for + genuinely new neighbors (issue #190). + """ + graph_links = await _fetch_graph_neighbors( + session, + memory_context=memory_context, + workspace_id=workspace_id, + seed_ids=seed_ids, + corpus_eligibility=live_proposal_expression(), + enrichment_ids=enrichment_ids, + ) + tunnel_links = await _fetch_candidate_tunnel_neighbors( + session, + memory_context=memory_context, + workspace_id=workspace_id, + seed_items=seed_items, + exclude_ids=exclude_ids, + enrichment_ids=enrichment_ids, + ) + + discovered_ids = (set(graph_links) | set(tunnel_links)) - set(seed_ids) - exclude_ids + if enrichment_ids: + discovered_ids -= enrichment_ids + neighbor_items: dict[UUID, MemoryItem] = {} + if discovered_ids: + neighbor_stmt = select(MemoryItem).where( + MemoryItem.id.in_(discovered_ids), + live_proposal_expression(), + read_eligibility_expression(memory_context), + ) + if workspace_id is not None: + # Same explicit-restriction defense in depth the discovery + # queries enforce — never an unscoped fallback. + neighbor_stmt = neighbor_stmt.where(MemoryItem.workspace_id == workspace_id) + rows = await session.execute(neighbor_stmt) + neighbor_items = {row.id: row for row in rows.scalars().all()} + return CandidateNeighborDiscovery( + graph_links=graph_links, + tunnel_links=tunnel_links, + neighbor_items=neighbor_items, + ) + + async def expand_recall_candidates( session: AsyncSession, *, diff --git a/sdk/engram-client/tests/test_models.py b/sdk/engram-client/tests/test_models.py index f7e0602..8448b44 100644 --- a/sdk/engram-client/tests/test_models.py +++ b/sdk/engram-client/tests/test_models.py @@ -72,14 +72,18 @@ def test_classify_legacy_confidence_is_canonical_alias() -> None: def test_recall_response_accepts_structured_evidence_blocks() -> None: """Issue #188: candidate-profile items carry the structured ``evidence`` - block and mirrored top-level ``epistemic_state``. The untyped item dicts - must pass the model through unchanged — unknown/null states stay - structured, never flattened into a numeric confidence field.""" + block and mirrored top-level ``epistemic_state``. Issue #190 adds the + structured ``relationship`` relevance block on expansion-reached items. + The untyped item dicts must pass the model through unchanged — + unknown/null states stay structured, never flattened into a numeric + confidence field.""" item = { "id": "0197c0de-0000-7000-8000-000000000001", "kind": "fact", "content": "served candidate item", "epistemic_state": "unknown", + "relevance_score": 0.735, + "utility_score": 0.6, "warning_codes": ["unreviewed", "evidence_unknown", "risk_unknown"], "evidence": { "source": "v2_fresh_evaluation", @@ -93,6 +97,18 @@ def test_recall_response_accepts_structured_evidence_blocks() -> None: "retention_state": "unknown", "effective_assessment_refs": [], }, + "relationship": { + "version": "relationship-relevance-v1", + "origins": ["graph"], + "direct": False, + "direct_semantic_score": None, + "source_seed_score": 0.9, + "graph_contribution": 0.9, + "graph_edge_types": ["derived_from"], + "tunnel_labels": [], + "relevance_score": 0.735, + "components": {"semantic": 0.63, "graph": 0.135, "tunnel": 0.0}, + }, } response = RecallResponse( working_set="[fact] served candidate item", @@ -107,6 +123,8 @@ def test_recall_response_accepts_structured_evidence_blocks() -> None: served = response.items[0] assert served["evidence"]["epistemic_state"] == served["epistemic_state"] assert served["evidence"]["source"] == "v2_fresh_evaluation" + assert served["relationship"]["version"] == "relationship-relevance-v1" + assert served["relevance_score"] == served["relationship"]["relevance_score"] # No numeric confidence was invented anywhere in the item. assert "trust_score" not in served assert "confidence" not in served diff --git a/tests/test_recall_profile_semantic.py b/tests/test_recall_profile_semantic.py index 3ecfb66..83f9ce4 100644 --- a/tests/test_recall_profile_semantic.py +++ b/tests/test_recall_profile_semantic.py @@ -114,8 +114,16 @@ async def _clean_db(): await conn.execute(text("DELETE FROM jobs")) await conn.execute(text("DELETE FROM item_events")) await conn.execute(text("DELETE FROM classification_runs")) + # Issue #190 expansion fixtures: edges cascade with their items, but + # tunnels, workspaces/principals, and cross-tenant rows need explicit + # cleanup (after memory_items, which reference them). + await conn.execute(text("DELETE FROM memory_edges")) + await conn.execute(text("DELETE FROM tunnels")) + await conn.execute(text("DELETE FROM tenants WHERE slug LIKE 'exp190-%'")) await conn.execute(text("DELETE FROM memory_embeddings")) await conn.execute(text("DELETE FROM memory_items")) + await conn.execute(text("DELETE FROM workspaces WHERE slug LIKE 'exp190-%'")) + await conn.execute(text("DELETE FROM principals WHERE name LIKE 'exp190-%'")) # Tenant shadow policy fails closed after every test. await conn.execute( text( @@ -1636,3 +1644,1408 @@ async def test_shadow_compare_with_evidence_remains_read_only(client, monkeypatc assert await _recall_log_count() == logs_before assert await _recall_counts([item["id"]]) == counts_before + + +# ---- admission-first relationship expansion (issue #190 / ENG-RECALL-003D) ---- +# +# The candidate profiles expand through the bounded graph/tunnel mechanics, +# strictly after direct V2 admission: only admitted direct items seed +# discovery, every expanded neighbor is independently admitted through the +# exact V2 surface, and relationship-aware relevance (versioned, +# importance-free) feeds the separated utility ranking. All fixtures below +# make neighbors *expansion-only* by deleting their embedding row — semantic +# retrieval can then never surface them, so their presence in a packet is +# mechanical proof the expansion path (and only it) reached them. + + +async def _link_items( + source_id: str, target_id: str, edge_type: str, *, weight: float | None = None +) -> None: + async with _test_engine.begin() as conn: + tenant_id = await conn.scalar( + text("SELECT tenant_id::text FROM memory_items WHERE id = :id"), {"id": source_id} + ) + await conn.execute( + text( + "INSERT INTO memory_edges (id, tenant_id, source_item_id, target_item_id, " + "edge_type, weight) " + "VALUES (gen_random_uuid(), :tenant, :src, :tgt, :et, :w)" + ), + {"tenant": tenant_id, "src": source_id, "tgt": target_id, "et": edge_type, "w": weight}, + ) + + +async def _unlink_items(source_id: str, target_id: str) -> None: + async with _test_engine.begin() as conn: + await conn.execute( + text( + "DELETE FROM memory_edges WHERE source_item_id = :src AND target_item_id = :tgt" + ), + {"src": source_id, "tgt": target_id}, + ) + + +async def _mk_tunnel(source_wing: str, target_wing: str, *, label: str | None = None) -> None: + async with _test_engine.begin() as conn: + tenant_id = await conn.scalar(text("SELECT id::text FROM tenants WHERE slug = 'default'")) + await conn.execute( + text( + "INSERT INTO tunnels (id, tenant_id, source_wing, target_wing, label) " + "VALUES (gen_random_uuid(), :tenant, :sw, :tw, :label)" + ), + {"tenant": tenant_id, "sw": source_wing, "tw": target_wing, "label": label}, + ) + + +async def _make_expansion_only(item_id: str) -> None: + """Remove the item's embedding so only relationship expansion can reach it.""" + async with _test_engine.begin() as conn: + await conn.execute( + text("DELETE FROM memory_embeddings WHERE memory_item_id = :id"), {"id": item_id} + ) + + +async def _update_item(item_id: str, **assignments: Any) -> None: + if not assignments: + return + clause = ", ".join(f"{column} = :{column}" for column in assignments) + async with _test_engine.begin() as conn: + await conn.execute( + text(f"UPDATE memory_items SET {clause} WHERE id = :id"), + {"id": item_id, **assignments}, + ) + + +# A third deterministic angle (cosine ~0.7071 against the query vector): +# strictly weaker than _NEAR_VEC, still well inside the retrieval window. +_LOWER_VEC = [0.7071067811865476, 0.7071067811865476] + [0.0] * 1534 + + +def _patch_keyed_embeddings( + monkeypatch: pytest.MonkeyPatch, vectors: dict[str, list[float]] +) -> None: + """Deterministic per-content embeddings for the source-attribution + regressions: exact contents in ``vectors`` get their assigned vector, + everything else falls back to the prefix-based module fake.""" + import engram.embeddings as embeddings_mod + from engram import recall as recall_mod + + async def keyed_embedding( + text_value: str, *_args: object, **_kwargs: object + ) -> list[float] | None: + if text_value in vectors: + return vectors[text_value] + return _fake_embedding_for(text_value) + + monkeypatch.setattr(recall_mod, "generate_embedding", keyed_embedding) + monkeypatch.setattr(memory_routes, "generate_embedding", keyed_embedding) + monkeypatch.setattr(embeddings_mod, "generate_embedding", keyed_embedding) + + +async def _set_item_embedding(item_id: str, vector: list[float]) -> None: + """Rewrite one stored embedding — the surgical "vary only this seed" + knob for attribution regressions (item content and V2 binding intact).""" + literal = "[" + ", ".join(repr(component) for component in vector) + "]" + async with _test_engine.begin() as conn: + await conn.execute( + text( + "UPDATE memory_embeddings SET embedding = CAST(:vec AS vector) " + "WHERE memory_item_id = :id" + ), + {"id": item_id, "vec": literal}, + ) + + +async def _fabricate_v2_row( + item_id: str, + *, + schema_version: str | None = None, + policy_config_digest: str | None = None, +) -> None: + """Append a newer V2-profile row with a chosen identity defect. + + The latest-row lookup (evaluated_at desc, id desc) resolves this row, so + a wrong schema_version makes the item ``unsupported`` and a wrong policy + digest makes it ``mismatched`` — otherwise shaped exactly like a real + persisted V2 shadow row (the DB CHECK contract validates that shape). + """ + from engram.admission_policy import V2_SCHEMA_VERSION + from engram.models import AdmissionAssessment as AssessmentRow + + async with _test_session_factory() as session: + row = ( + ( + await session.execute( + text( + "SELECT tenant_id::text, content_hash FROM memory_items WHERE id = :id" + ), + {"id": item_id}, + ) + ) + .mappings() + .one() + ) + session.add( + AssessmentRow( + id=uuid4(), + tenant_id=row["tenant_id"], + memory_item_id=item_id, + schema_version=schema_version or V2_SCHEMA_VERSION, + mode="shadow", + trigger_type="test", + trigger_id=f"test:{uuid4()}", + invocation_source="test", + evaluated_at=datetime.now(UTC) + timedelta(seconds=1), + item_content_hash=row["content_hash"], + input_digest="sha256:" + "0" * 64, + policy_profile_key="risk_aware_shadow_v1", + policy_contract_version=_V2_POLICY.policy_version, + policy_config_digest=policy_config_digest or _V2_POLICY.artifact_digest, + selected_basis=None, + outcome="would_admit", + blocker_codes=[], + reason_codes=[], + decision_inputs={}, + available_memory_assessment_refs=[], + # The v2 shadow-contract CHECK requires the V2 column set. + risk_state="low", + epistemic_state="supported", + retention_state="retain", + effective_memory_assessment_refs=[], + highest_admission_tier="semantic_governed", + surface_decisions={ + "semantic_exploratory": "allow", + "semantic_governed": "allow", + "startup": "withhold", + }, + conflict_recheck_status="not_run", + next_actions=[], + decision_hash="sha256:" + "0" * 64, + ) + ) + await session.commit() + + +async def _seed_qualified( + client: AsyncClient, content: str, **payload: Any +) -> dict[str, Any]: + """A live proposal with a current qualifying V2 row (low risk, supported).""" + item = await _remember(client, content, source_type="extraction", **payload) + await _persist_v2_row(item["id"]) + return item + + +async def _candidate_packet( + client: AsyncClient, profile: str = "governed", **extra: Any +) -> dict[str, Any]: + shadow = await _shadow_compare(client, profiles=[profile], **extra) + return shadow["candidates"][0] + + +def _relationship(item: dict[str, Any]) -> dict[str, Any]: + relationship = item.get("relationship") + assert relationship is not None, "admitted expanded item must carry the relationship block" + return relationship + + +# ---- admission-before-expansion ---- + + +async def test_withheld_direct_candidate_cannot_seed_graph_expansion( + client, monkeypatch +): + """Required test 1: a direct candidate the exact V2 surface withholds + never seeds expansion — its qualified graph neighbor is absent, has no + diagnostic, and the packet records that no expansion ran at all.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _remember(client, "semantic target withheld seed", source_type="extraction") + await _persist_v2_row(seed["id"], risk=None) # governed: review_required + neighbor = await _seed_qualified(client, "graph neighbor of withheld seed") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from", weight=1.0) + + governed = await _candidate_packet(client) + assert governed["item_count"] == 0 + diagnostics = {d["item_id"]: d for d in governed["admission_diagnostics"]} + # The seed itself is a DIRECT withhold; the neighbor was never discovered. + assert diagnostics[seed["id"]]["origin"] == "direct" + assert diagnostics[seed["id"]]["v2_surface_decision"] == "review_required" + assert neighbor["id"] not in diagnostics + assert governed["expansion"] is None # zero admitted seeds -> no expansion + + +async def test_withheld_direct_candidate_cannot_seed_tunnel_expansion( + client, monkeypatch +): + """Required test 2: same boundary for tunnels — a withheld seed's wing + membership reveals nothing.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _remember( + client, + "semantic target withheld tunnel seed", + source_type="extraction", + wing="WithheldWing", + room="src", + ) + await _persist_v2_row(seed["id"], risk=None) + neighbor = await _seed_qualified( + client, "tunnel neighbor of withheld seed", wing="TunnelTarget", room="dst" + ) + await _make_expansion_only(neighbor["id"]) + await _mk_tunnel("WithheldWing", "TunnelTarget", label="blocked") + + governed = await _candidate_packet(client) + assert governed["item_count"] == 0 + diagnostics = {d["item_id"] for d in governed["admission_diagnostics"]} + assert neighbor["id"] not in diagnostics + assert governed["expansion"] is None + + +async def test_admitted_seed_discovers_qualified_graph_neighbor(client, monkeypatch): + """Required tests 3 + 5: an admitted direct seed discovers an eligible + graph neighbor, and that neighbor is admitted only through its own + current + allow decision on the exact governed surface.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target graph seed") + neighbor = await _seed_qualified(client, "qualified graph neighbor content") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from") + + governed = await _candidate_packet(client) + by_id = {item["id"]: item for item in governed["items"]} + assert set(by_id) == {seed["id"], neighbor["id"]} + expanded = by_id[neighbor["id"]] + # The neighbor presents the full admitted-candidate evidence identity. + assert expanded["admission"]["surface"] == "semantic_governed" + assert expanded["admission"]["surface_decision"] == "allow" + assert expanded["admission"]["v2"]["resolution_status"] == "current" + _evidence_identity_asserts(expanded) + # Expansion-only: no vector was ever compared against it. + assert expanded["distance"] is None + assert expanded["similarity_score"] is None + relationship = _relationship(expanded) + assert relationship["version"] == "relationship-relevance-v1" + assert relationship["origins"] == ["graph"] + assert relationship["direct"] is False + assert relationship["graph_edge_types"] == ["derived_from"] + assert relationship["relevance_score"] == expanded["relevance_score"] + assert any("linked via derived_from" in reason for reason in expanded["reasons"]) + assert governed["expansion"] == { + "version": "relationship-relevance-v1", + "seed_count": 1, + "discovered_neighbors": 1, + "graph_neighbors": 1, + "tunnel_neighbors": 0, + "admitted_expanded": 1, + "withheld_expanded": 0, + } + assert governed["v2_resolution"]["resolved_count"] == 2 + assert governed["v2_resolution"]["resolution_status_counts"] == {"current": 2} + + +async def test_admitted_seed_discovers_qualified_tunnel_neighbor(client, monkeypatch): + """Required test 4: an admitted seed's tunnel membership discovers an + eligible neighbor in the tunneled (wing, room).""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + await _seed_qualified( + client, "semantic target tunnel seed", wing="TunnelSeed", room="ops" + ) + neighbor = await _seed_qualified( + client, "qualified tunnel neighbor content", wing="TunnelFar", room="run" + ) + await _make_expansion_only(neighbor["id"]) + await _mk_tunnel("TunnelSeed", "TunnelFar", label="ops-link") + + governed = await _candidate_packet(client) + by_id = {item["id"]: item for item in governed["items"]} + assert neighbor["id"] in by_id + expanded = by_id[neighbor["id"]] + _evidence_identity_asserts(expanded) + relationship = _relationship(expanded) + assert relationship["origins"] == ["tunnel"] + assert relationship["tunnel_labels"] == ["ops-link"] + assert relationship["graph_edge_types"] == [] + assert any('same tunnel "ops-link"' in r for r in expanded["reasons"]) + assert governed["expansion"]["tunnel_neighbors"] == 1 + assert governed["expansion"]["admitted_expanded"] == 1 + + +# ---- tunnel source-seed attribution (#191 merge-blocker fix) ---- + + +async def test_tunnel_neighbor_attributes_only_the_seed_that_reaches_it( + client, monkeypatch +): + """Tunnel relevance derives from the admitted seed whose tunnel + membership actually discovered the neighbor — never from the strongest + admitted seed in the packet. A (similarity 1.0, no tunnel relationship + to C) leaves C attributed entirely to B (lower similarity, + tunnel-connected to C), and varying only A's similarity while holding B + and the topology fixed changes nothing about C at all.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_keyed_embeddings( + monkeypatch, + { + "attribution unrelated high seed": _TARGET_VEC, # A: similarity 1.0 + "attribution connected low seed": _NEAR_VEC, # B: similarity ~0.92 + }, + ) + await _enable_tenant_shadow_policy() + + high = await _seed_qualified(client, "attribution unrelated high seed") + connected = await _seed_qualified( + client, "attribution connected low seed", wing="AttrSeed", room="src" + ) + neighbor = await _seed_qualified( + client, "attribution tunnel neighbor", wing="AttrFar", room="dst" + ) + await _make_expansion_only(neighbor["id"]) + await _mk_tunnel("AttrSeed", "AttrFar", label="attr-link") + + governed = await _candidate_packet(client) + by_id = {item["id"]: item for item in governed["items"]} + high_similarity = by_id[high["id"]]["similarity_score"] + connected_similarity = by_id[connected["id"]]["similarity_score"] + assert high_similarity > connected_similarity # A is the packet's best seed + expanded = by_id[neighbor["id"]] + relationship = _relationship(expanded) + assert relationship["origins"] == ["tunnel"] + assert relationship["source_seed_score"] == connected_similarity + assert relationship["source_seed_score"] != high_similarity + + def _snapshot(packet: dict[str, Any]) -> dict[str, Any]: + item = next(i for i in packet["items"] if i["id"] == neighbor["id"]) + return { + "relationship": item["relationship"], + "relevance": item["relevance_score"], + "utility": item["utility_score"], + "score": item["score"], + "admission": item["admission"], + "evidence": item["evidence"], + } + + before = _snapshot(governed) + # Vary ONLY A: its similarity drops well below B's. B and the tunnel + # topology are untouched, and A must remain an admitted (still unrelated) + # seed so the variation is real. + await _set_item_embedding(high["id"], _LOWER_VEC) + varied_packet = await _candidate_packet(client) + assert high["id"] in {item["id"] for item in varied_packet["items"]} + varied = _snapshot(varied_packet) + assert varied == before + + +async def test_tunnel_multi_seed_source_attribution_strongest_connected_wins( + client, monkeypatch +): + """Multi-source tunnel attribution: when two admitted seeds both + genuinely reach C through their own tunnels, C uses the strongest of + those two; remove the stronger seed's tunnel and attribution drops to + the remaining connected seed exactly — the unrelated-seed rule seen from + the other side.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_keyed_embeddings( + monkeypatch, + { + "attribution multi strong seed": _TARGET_VEC, # similarity 1.0 + "attribution multi weak seed": _NEAR_VEC, # similarity ~0.92 + }, + ) + await _enable_tenant_shadow_policy() + + strong = await _seed_qualified( + client, "attribution multi strong seed", wing="AttrStrong", room="src" + ) + weak = await _seed_qualified( + client, "attribution multi weak seed", wing="AttrWeak", room="src" + ) + neighbor = await _seed_qualified( + client, "attribution multi tunnel neighbor", wing="AttrFarWing", room="dst" + ) + await _make_expansion_only(neighbor["id"]) + await _mk_tunnel("AttrStrong", "AttrFarWing", label="strong-link") + await _mk_tunnel("AttrWeak", "AttrFarWing", label="weak-link") + + governed = await _candidate_packet(client) + by_id = {item["id"]: item for item in governed["items"]} + relationship = _relationship(by_id[neighbor["id"]]) + assert relationship["origins"] == ["tunnel"] + # Both seeds actually reach C: the strongest connected seed wins. + assert relationship["source_seed_score"] == by_id[strong["id"]]["similarity_score"] + assert relationship["source_seed_score"] != by_id[weak["id"]]["similarity_score"] + + # Remove ONLY the strong seed's tunnel: the weak seed becomes the sole + # actually-connected source and attribution follows it exactly. + async with _test_engine.begin() as conn: + await conn.execute( + text( + "DELETE FROM tunnels " + "WHERE source_wing = 'AttrStrong' AND target_wing = 'AttrFarWing'" + ) + ) + after = await _candidate_packet(client) + by_id_after = {item["id"]: item for item in after["items"]} + relationship_after = _relationship(by_id_after[neighbor["id"]]) + assert relationship_after["source_seed_score"] == by_id_after[weak["id"]][ + "similarity_score" + ] + assert relationship_after["relevance_score"] < relationship["relevance_score"] + + +async def test_withheld_direct_candidate_does_not_consume_graph_capacity( + client, monkeypatch +): + """The direct-vs-new capacity split (#191 fix 1), mechanically: with a + one-slot graph window, a V2-withheld direct candidate holding the + strongest edge (A -> B, weight 1.0) must not consume the bounded slot — + the genuinely new, weaker-edge neighbor (A -> C, weight 0.9) is still + discovered and independently admitted.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + monkeypatch.setattr(settings, "max_graph_neighbors_per_item", 1) + monkeypatch.setattr(settings, "max_graph_expanded_items", 1) + + seed_a = await _seed_qualified(client, "semantic target capacity seed") + direct_b = await _remember(client, "semantic target withheld direct", source_type="extraction") + await _persist_v2_row(direct_b["id"], risk=None) # governed: review_required + neighbor_c = await _seed_qualified(client, "capacity expansion neighbor") + await _make_expansion_only(neighbor_c["id"]) + await _link_items(seed_a["id"], direct_b["id"], "derived_from", weight=1.0) + await _link_items(seed_a["id"], neighbor_c["id"], "derived_from", weight=0.9) + + governed = await _candidate_packet(client) + by_id = {item["id"]: item for item in governed["items"]} + # B is represented only by its direct-path withhold diagnostic. + assert direct_b["id"] not in by_id + b_diagnostics = [ + d for d in governed["admission_diagnostics"] if d["item_id"] == direct_b["id"] + ] + assert len(b_diagnostics) == 1 + assert b_diagnostics[0]["origin"] == "direct" + assert b_diagnostics[0]["v2_surface_decision"] == "review_required" + # B consumed no expansion capacity: C was discovered through the one-slot + # window despite B's stronger edge, and is served on its own exact V2 + # admission (current + allow). + expanded_c = by_id[neighbor_c["id"]] + assert _relationship(expanded_c)["origins"] == ["graph"] + assert expanded_c["admission"]["surface_decision"] == "allow" + assert expanded_c["admission"]["v2"]["resolution_status"] == "current" + assert expanded_c["distance"] is None # expansion-only: never vector-scored + _evidence_identity_asserts(expanded_c) + # Accounting counts genuinely new expansion candidates only — B is not a + # graph neighbor despite being edge-linked to the seed. + assert governed["expansion"] == { + "version": "relationship-relevance-v1", + "seed_count": 1, + "discovered_neighbors": 1, + "graph_neighbors": 1, + "tunnel_neighbors": 0, + "admitted_expanded": 1, + "withheld_expanded": 0, + } + # Direct window (A + B) plus the one neighbor window (C): B was never + # resolved a second time through expansion. + assert governed["v2_resolution"]["resolved_count"] == 3 + assert governed["v2_resolution"]["resolution_status_counts"] == {"current": 3} + + +# ---- independent neighbor admission ---- + + +async def test_neighbor_withheld_by_surface_despite_strongest_edge(client, monkeypatch): + """Required test 6: a maximal-weight edge cannot turn a review_required + governed decision into an admission; the withhold is diagnosed with its + graph origin.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target strong edge seed") + neighbor = await _remember(client, "review routed graph neighbor", source_type="extraction") + await _persist_v2_row(neighbor["id"], risk=None) # governed: review_required + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "supports", weight=1.0) + + governed = await _candidate_packet(client) + assert {item["id"] for item in governed["items"]} == {seed["id"]} + assert governed["omitted_by_admission"].get("v2_surface_review_required") == 1 + diagnostic = next( + d for d in governed["admission_diagnostics"] if d["item_id"] == neighbor["id"] + ) + assert diagnostic["origin"] == "graph" + assert diagnostic["v2_resolution_status"] == "current" + assert diagnostic["v2_surface_decision"] == "review_required" + assert governed["expansion"]["withheld_expanded"] == 1 + + +async def test_exploratory_high_risk_neighbor_admitted_on_its_own_surface( + client, monkeypatch +): + """Required test 7: a high-risk neighbor enters the exploratory packet + only because the exact exploratory surface allows it, with the canonical + risk warning preserved; governed withholds the same neighbor.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target risk seed") + neighbor = await _remember(client, "high risk graph neighbor", source_type="extraction") + await _persist_v2_row(neighbor["id"], risk="high", epistemic_state="supported") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "references") + + governed = await _candidate_packet(client) + assert neighbor["id"] not in {item["id"] for item in governed["items"]} + gov_diag = {d["item_id"]: d for d in governed["admission_diagnostics"]} + assert gov_diag[neighbor["id"]]["v2_surface_decision"] == "review_required" + + exploratory = await _candidate_packet(client, profile="exploratory") + by_id = {item["id"]: item for item in exploratory["items"]} + assert neighbor["id"] in by_id + item = by_id[neighbor["id"]] + assert item["admission"]["surface"] == "semantic_exploratory" + assert "risk_high" in item["warning_codes"] + _evidence_identity_asserts(item) + assert _relationship(item)["origins"] == ["graph"] + + +async def test_noncurrent_expanded_neighbors_fail_closed_with_diagnostics( + client, monkeypatch +): + """Required test 8: missing | stale | mismatched | unsupported expanded + neighbors are withheld, diagnostic-only, each with its own resolution + status and origin.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target noncurrent seed") + + missing = await _remember(client, "never assessed neighbor", source_type="extraction") + await _make_expansion_only(missing["id"]) + await _link_items(seed["id"], missing["id"], "references") + + stale = await _remember(client, "stale neighbor", source_type="extraction") + await _persist_v2_row(stale["id"]) + await _update_item(stale["id"], human_verified=True) # changes the fresh hash + await _make_expansion_only(stale["id"]) + await _link_items(seed["id"], stale["id"], "references") + + unsupported = await _remember(client, "unsupported neighbor", source_type="extraction") + await _persist_v2_row(unsupported["id"]) + await _fabricate_v2_row(unsupported["id"], schema_version="engram.admission-assessment.v1") + await _make_expansion_only(unsupported["id"]) + await _link_items(seed["id"], unsupported["id"], "references") + + mismatched = await _remember(client, "mismatched neighbor", source_type="extraction") + await _persist_v2_row(mismatched["id"]) + await _fabricate_v2_row(mismatched["id"], policy_config_digest="sha256:" + "f" * 64) + await _make_expansion_only(mismatched["id"]) + await _link_items(seed["id"], mismatched["id"], "references") + + governed = await _candidate_packet(client) + assert {item["id"] for item in governed["items"]} == {seed["id"]} + diagnostics = {d["item_id"]: d for d in governed["admission_diagnostics"]} + expected = { + missing["id"]: ("missing", "v2_decision_missing"), + stale["id"]: ("stale", "v2_decision_stale"), + unsupported["id"]: ("unsupported", "v2_decision_unsupported"), + mismatched["id"]: ("mismatched", "v2_decision_mismatched"), + } + for item_id, (status, reason) in expected.items(): + assert diagnostics[item_id]["origin"] == "graph", item_id + assert diagnostics[item_id]["v2_resolution_status"] == status, item_id + assert diagnostics[item_id]["reason_codes"] == [reason], item_id + assert governed["expansion"]["withheld_expanded"] == 4 + assert governed["expansion"]["admitted_expanded"] == 0 + + +async def test_supports_edge_cannot_upgrade_epistemic_state(client, monkeypatch): + """Required test 9: a maximal supports edge leaves the V2 epistemic state + unknown — relationship is relevance, never evidence.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target epistemic seed") + neighbor = await _remember(client, "unknown evidence neighbor", source_type="extraction") + await _persist_v2_row(neighbor["id"], risk=None) # epistemic: unknown + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "supports", weight=1.0) + + exploratory = await _candidate_packet(client, profile="exploratory") + by_id = {item["id"]: item for item in exploratory["items"]} + assert neighbor["id"] in by_id # the exploratory surface allows it + item = by_id[neighbor["id"]] + assert item["epistemic_state"] == "unknown" + assert item["evidence"]["epistemic_state"] == "unknown" + assert "evidence_unknown" in item["warning_codes"] + assert _relationship(item)["graph_edge_types"] == ["supports"] + assert item["admission"]["surface_decision"] == "allow" + + +async def test_high_importance_neighbor_cannot_bypass_v2_withholding( + client, monkeypatch +): + """Required test 10: importance is utility — it can never buy admission.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target importance seed") + neighbor = await _remember( + client, "important but unqualified neighbor", source_type="extraction", importance=1.0 + ) + await _persist_v2_row(neighbor["id"], risk=None) + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from", weight=1.0) + + governed = await _candidate_packet(client) + assert neighbor["id"] not in {item["id"] for item in governed["items"]} + assert governed["omitted_by_admission"].get("v2_surface_review_required") == 1 + + +# ---- relevance/utility separation ---- + + +async def test_importance_moves_utility_and_rank_but_not_relevance(client, monkeypatch): + """Required test 11: changing only the neighbor's importance changes its + utility and final rank, never its relationship relevance, evidence + state, or admission.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target utility seed") + neighbor = await _seed_qualified(client, "utility probe neighbor", importance=0.1) + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from") + + def _snapshot(packet: dict[str, Any]) -> dict[str, Any]: + item = next(i for i in packet["items"] if i["id"] == neighbor["id"]) + return { + "relevance": item["relevance_score"], + "relationship": item["relationship"], + "evidence": item["evidence"], + "admission": item["admission"], + "utility": item["utility_score"], + "score": item["score"], + } + + before = _snapshot(await _candidate_packet(client, profile="exploratory")) + await _update_item(neighbor["id"], importance=0.9) + after = _snapshot(await _candidate_packet(client, profile="exploratory")) + + assert before["relevance"] == after["relevance"] + assert before["relationship"] == after["relationship"] + assert before["evidence"] == after["evidence"] + assert before["admission"] == after["admission"] + assert before["utility"] < after["utility"] + assert before["score"] < after["score"] + + +async def test_edge_strength_moves_relevance_but_not_utility_or_admission( + client, monkeypatch +): + """Required test 12: changing only the edge weight changes relationship + relevance, never utility, epistemic state, or admission.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target edge weight seed") + neighbor = await _seed_qualified(client, "edge weight probe neighbor") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "supports", weight=0.3) + + def _snapshot(packet: dict[str, Any]) -> dict[str, Any]: + item = next(i for i in packet["items"] if i["id"] == neighbor["id"]) + return { + "relevance": item["relevance_score"], + "graph_contribution": item["relationship"]["graph_contribution"], + "utility": item["utility_score"], + "evidence": item["evidence"], + "admission": item["admission"], + } + + weak = _snapshot(await _candidate_packet(client, profile="exploratory")) + await _unlink_items(seed["id"], neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "supports", weight=0.9) + strong = _snapshot(await _candidate_packet(client, profile="exploratory")) + + assert weak["graph_contribution"] < strong["graph_contribution"] + assert weak["relevance"] < strong["relevance"] + assert weak["utility"] == strong["utility"] + assert weak["evidence"] == strong["evidence"] + assert weak["admission"] == strong["admission"] + + +async def test_epistemic_and_utility_inputs_cannot_move_relationship_relevance( + client, monkeypatch +): + """Required test 13: source trust, memory confidence, human verification, + and exposure counters are neither relevance nor admission inputs — for a + fixed expansion/admission binding they change nothing. (human_verified + IS an input to the V2 decision hash, so flipping it re-binds the row; + re-persisting restores the identical binding and proves relevance never + moved.)""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target exclusion seed") + neighbor = await _seed_qualified(client, "exclusion probe neighbor") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from") + + def _snapshot(packet: dict[str, Any]) -> dict[str, Any]: + item = next(i for i in packet["items"] if i["id"] == neighbor["id"]) + return { + "relevance": item["relevance_score"], + "relationship": item["relationship"], + "evidence": item["evidence"], + "admission": item["admission"], + } + + before = _snapshot(await _candidate_packet(client)) + await _update_item( + neighbor["id"], + source_trust=0.05, + memory_confidence=0.05, + recall_count=99, + last_recalled_at=datetime.now(UTC), + ) + after = _snapshot(await _candidate_packet(client)) + assert before == after + + # human_verified moves the V2 decision hash (it is epistemic input to + # the policy), so the binding re-persists first; with the binding fixed + # again, relevance is still exactly what it was. + await _update_item(neighbor["id"], human_verified=True) + async with _test_engine.begin() as conn: + # classification_runs is one-per-item; dropping the old run cascades + # its #157 assessment so the re-persist can rebuild both fresh. + await conn.execute( + text("DELETE FROM classification_runs WHERE memory_item_id = :id"), + {"id": neighbor["id"]}, + ) + await _persist_v2_row(neighbor["id"]) + rebound = _snapshot(await _candidate_packet(client)) + assert rebound["relevance"] == before["relevance"] + assert rebound["relationship"] == before["relationship"] + assert rebound["admission"]["decision"] == "admit" + + +async def test_direct_and_expanded_origin_merge_is_deterministic(client, monkeypatch): + """Required test 14: a direct hit that is also a graph neighbor merges + origins deterministically and exposes the structured components.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed_a = await _seed_qualified(client, "semantic target merge a") + seed_b = await _seed_qualified(client, "semantic target merge b") + await _link_items(seed_b["id"], seed_a["id"], "supports", weight=0.6) + + first = await _candidate_packet(client) + second = await _candidate_packet(client) + by_id = {item["id"]: item for item in first["items"]} + merged = by_id[seed_a["id"]] + # The merged item keeps its direct fields and gains the graph origin. + assert merged["distance"] is not None + assert merged["similarity_score"] is not None + relationship = _relationship(merged) + assert relationship["origins"] == ["semantic", "graph"] + assert relationship["direct"] is True + assert relationship["direct_semantic_score"] == merged["similarity_score"] + assert relationship["graph_edge_types"] == ["supports"] + assert set(relationship["components"]) == {"semantic", "graph", "tunnel"} + assert any("linked via supports" in r for r in merged["reasons"]) + # Deterministic: identical packet on re-evaluation. + assert first["items"] == second["items"] + assert [i["id"] for i in first["items"]] == [i["id"] for i in second["items"]] + + +async def test_direct_candidate_merges_semantic_tunnel_origin(client, monkeypatch): + """The ``semantic+tunnel`` origin merge (#191 fix 2): an admitted direct + item sitting in a tunnel-linked wing collects tunnel-origin metadata + through another admitted seed's tunnel membership — direct fields, + admission, and evidence identity unchanged, no second V2 admission, and + no tunnel new-neighbor accounting.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified( + client, "semantic target tunnel origin seed", wing="OrigSeed", room="src" + ) + direct = await _seed_qualified( + client, "semantic target tunnel origin direct", wing="OrigFar", room="dst" + ) + await _mk_tunnel("OrigSeed", "OrigFar", label="origin-link") + + governed = await _candidate_packet(client) + by_id = {item["id"]: item for item in governed["items"]} + assert set(by_id) == {seed["id"], direct["id"]} + item = by_id[direct["id"]] + relationship = _relationship(item) + assert relationship["origins"] == ["semantic", "tunnel"] + assert relationship["direct"] is True + assert relationship["tunnel_labels"] == ["origin-link"] + assert relationship["graph_edge_types"] == [] + assert any('same tunnel "origin-link"' in r for r in item["reasons"]) + # The direct identity is retained untouched. + assert item["distance"] is not None + assert item["similarity_score"] is not None + assert relationship["direct_semantic_score"] == item["similarity_score"] + assert item["admission"]["surface_decision"] == "allow" + _evidence_identity_asserts(item) + # No second V2 admission for the direct item: the neighbor window is + # empty (nothing genuinely new was discovered), so exactly the two + # direct candidates were resolved. + assert governed["v2_resolution"]["resolved_count"] == 2 + assert governed["v2_resolution"]["resolution_status_counts"] == {"current": 2} + # And the direct item never counts toward the new-neighbor totals. + assert governed["expansion"] == { + "version": "relationship-relevance-v1", + "seed_count": 2, + "discovered_neighbors": 0, + "graph_neighbors": 0, + "tunnel_neighbors": 0, + "admitted_expanded": 0, + "withheld_expanded": 0, + } + + +async def test_direct_candidate_merges_semantic_graph_tunnel_origin(client, monkeypatch): + """The ``semantic+graph+tunnel`` merge: pins the complete origin-merging + contract — one admitted direct item reached by an edge from another + admitted seed AND sitting in that seed's tunneled wing carries all three + origins while keeping its direct admission/evidence identity.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified( + client, "semantic target triple origin seed", wing="TriSeed", room="src" + ) + direct = await _seed_qualified( + client, "semantic target triple origin direct", wing="TriFar", room="dst" + ) + await _link_items(seed["id"], direct["id"], "supports", weight=0.7) + await _mk_tunnel("TriSeed", "TriFar", label="tri-link") + + governed = await _candidate_packet(client) + by_id = {item["id"]: item for item in governed["items"]} + assert set(by_id) == {seed["id"], direct["id"]} + item = by_id[direct["id"]] + relationship = _relationship(item) + assert relationship["origins"] == ["semantic", "graph", "tunnel"] + assert relationship["direct"] is True + assert relationship["graph_edge_types"] == ["supports"] + assert relationship["tunnel_labels"] == ["tri-link"] + assert set(relationship["components"]) == {"semantic", "graph", "tunnel"} + assert any("linked via supports" in r for r in item["reasons"]) + assert any('same tunnel "tri-link"' in r for r in item["reasons"]) + assert item["distance"] is not None + assert item["admission"]["surface_decision"] == "allow" + _evidence_identity_asserts(item) + assert governed["v2_resolution"]["resolved_count"] == 2 + assert governed["expansion"]["graph_neighbors"] == 0 + assert governed["expansion"]["tunnel_neighbors"] == 0 + assert governed["expansion"]["admitted_expanded"] == 0 + + +async def test_candidate_rank_reproducible_from_published_inputs(client, monkeypatch): + """Required test 15: every packet item's rank is recomputable from its + published relevance + utility — direct, expanded, and merged alike.""" + from engram.recall_signals import compute_signal_rank_score + + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target reproducibility seed") + graph_neighbor = await _seed_qualified(client, "reproducibility graph neighbor") + await _make_expansion_only(graph_neighbor["id"]) + await _link_items(seed["id"], graph_neighbor["id"], "derived_from") + + for profile in ("governed", "exploratory"): + packet = await _candidate_packet(client, profile=profile) + assert len(packet["items"]) == 2 + for item in packet["items"]: + assert item["score"] == compute_signal_rank_score( + similarity=item["relevance_score"], utility=item["utility_score"] + ) + + +# ---- security / RLS boundaries through edges and tunnels ---- + + +async def test_cross_tenant_neighbor_is_never_discoverable(client, monkeypatch): + """Required test 16: an edge row can name a foreign item, but the + neighbor is never discovered or diagnosed.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + from engram.models import MemoryItem as MemoryItemRow + from engram.models import Principal as PrincipalRow + from engram.models import Tenant as TenantRow + + seed = await _seed_qualified(client, "semantic target cross tenant seed") + async with _test_session_factory() as session: + tenant = TenantRow(name="exp190 other", slug=f"exp190-{uuid4().hex[:8]}") + session.add(tenant) + await session.flush() + principal = PrincipalRow(tenant_id=tenant.id, name="exp190-agent", type="agent") + session.add(principal) + await session.flush() + foreign = MemoryItemRow( + tenant_id=tenant.id, + principal_id=principal.id, + content="foreign secret", + content_hash=f"h-{uuid4()}", + kind="fact", + visibility="tenant", + review_status="proposed", + ) + session.add(foreign) + await session.commit() + foreign_id = str(foreign.id) + # The edge row lives in the caller's tenant but points at the foreign item. + await _link_items(seed["id"], foreign_id, "derived_from", weight=1.0) + + governed = await _candidate_packet(client) + assert foreign_id not in {item["id"] for item in governed["items"]} + assert foreign_id not in {d["item_id"] for d in governed["admission_diagnostics"]} + + +async def test_private_neighbor_is_undiscoverable_and_undiagnosable(client, monkeypatch): + """Required test 17: another principal's private neighbor is invisible — + not in the packet, not even diagnosable by identity.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target private seed") + neighbor = await _remember( + client, + "private neighbor of seed", + source_type="extraction", + visibility="private", + ) + await _persist_v2_row(neighbor["id"]) + async with _test_engine.begin() as conn: + other_principal = str(uuid4()) + principal_name = f"exp190-private-{other_principal[:8]}" + tenant_id = await conn.scalar( + text("SELECT tenant_id::text FROM memory_items WHERE id = :id"), {"id": seed["id"]} + ) + await conn.execute( + text( + "INSERT INTO principals (id, tenant_id, name, type) " + "VALUES (:pid, :tid, :pname, 'agent')" + ), + {"pid": other_principal, "tid": tenant_id, "pname": principal_name}, + ) + await conn.execute( + text("UPDATE memory_items SET principal_id = :pid WHERE id = :id"), + {"pid": other_principal, "id": neighbor["id"]}, + ) + await _link_items(seed["id"], neighbor["id"], "derived_from", weight=1.0) + + governed = await _candidate_packet(client) + assert neighbor["id"] not in {item["id"] for item in governed["items"]} + assert neighbor["id"] not in {d["item_id"] for d in governed["admission_diagnostics"]} + # An inaccessible seed reveals nothing either: the only admitted item is + # the seed itself (required test 19's non-disclosure property). + assert governed["expansion"]["discovered_neighbors"] == 0 + + +async def test_out_of_workspace_neighbor_unreachable_through_edge(client, monkeypatch): + """Required test 18: a workspace-scoped comparison cannot reach a + qualified neighbor outside the workspace, even through a visible edge.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + async with _test_engine.begin() as conn: + tenant_id = await conn.scalar(text("SELECT id::text FROM tenants WHERE slug = 'default'")) + admin_id = await conn.scalar( + text("SELECT p.id::text FROM principals p JOIN tenants t ON t.id = p.tenant_id " + "WHERE t.slug = 'default' AND p.name = 'admin'") + ) + workspace_id = str(uuid4()) + workspace_slug = f"exp190-ws-{workspace_id[:8]}" + await conn.execute( + text( + "INSERT INTO workspaces (id, tenant_id, name, slug) " + "VALUES (:id, :tid, 'exp190 ws', :slug)" + ), + {"id": workspace_id, "tid": tenant_id, "slug": workspace_slug}, + ) + await conn.execute( + text( + "INSERT INTO workspace_members (id, workspace_id, principal_id, role) " + "VALUES (gen_random_uuid(), :ws, :pid, 'member')" + ), + {"ws": workspace_id, "pid": admin_id}, + ) + + seed = await _seed_qualified( + client, "semantic target workspace seed", workspace=workspace_slug + ) + neighbor = await _seed_qualified(client, "out of workspace neighbor") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from", weight=1.0) + + scoped = await _candidate_packet(client, workspace=workspace_slug) + assert {item["id"] for item in scoped["items"]} == {seed["id"]} + assert neighbor["id"] not in {d["item_id"] for d in scoped["admission_diagnostics"]} + assert scoped["expansion"]["discovered_neighbors"] == 0 + + +async def test_inaccessible_item_cannot_be_used_to_infer_neighbors(client, monkeypatch): + """Required test 19: an item the caller cannot read sits between the + admitted seed and further neighbors. Neither the inaccessible item nor + its own neighbors may be discovered, diagnosed, or inferred — the packet + and its diagnostics reveal nothing about them.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target inaccessible seed") + # An edge-visible but unreadable middle item (another principal's + # private proposal), itself linked to a fully qualified neighbor. + middle = await _remember( + client, "private middle item", source_type="extraction", visibility="private" + ) + far = await _seed_qualified(client, "qualified far neighbor of private middle") + await _make_expansion_only(middle["id"]) + await _make_expansion_only(far["id"]) + async with _test_engine.begin() as conn: + other_principal = str(uuid4()) + tenant_id = await conn.scalar( + text("SELECT tenant_id::text FROM memory_items WHERE id = :id"), {"id": seed["id"]} + ) + await conn.execute( + text( + "INSERT INTO principals (id, tenant_id, name, type) " + "VALUES (:pid, :tid, :pname, 'agent')" + ), + { + "pid": other_principal, + "tid": tenant_id, + "pname": f"exp190-infer-{other_principal[:8]}", + }, + ) + await conn.execute( + text("UPDATE memory_items SET principal_id = :pid WHERE id = :id"), + {"pid": other_principal, "id": middle["id"]}, + ) + await _link_items(seed["id"], middle["id"], "derived_from", weight=1.0) + await _link_items(middle["id"], far["id"], "derived_from", weight=1.0) + + governed = await _candidate_packet(client) + assert {item["id"] for item in governed["items"]} == {seed["id"]} + diagnosed = {d["item_id"] for d in governed["admission_diagnostics"]} + assert middle["id"] not in diagnosed + assert far["id"] not in diagnosed + assert governed["expansion"]["discovered_neighbors"] == 0 + + +# ---- performance / boundedness ---- + + +async def test_expansion_adds_no_provider_call(client, monkeypatch): + """Required test 20: with graph+tunnel expansion active, the comparison + still makes exactly one provider call — the shared query embedding.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target provider seed") + neighbor = await _seed_qualified(client, "provider probe neighbor") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from") + tunnel_neighbor = await _seed_qualified( + client, "provider tunnel neighbor", wing="ProvWing", room="a" + ) + await _seed_qualified( + client, "semantic target provider wing seed", wing="ProvWing", room="a" + ) + await _make_expansion_only(tunnel_neighbor["id"]) + await _mk_tunnel("ProvWing", "ProvOtherWing") + + # Seed first, THEN start counting: only the comparison itself is measured. + provider_calls = {"count": 0} + + async def counting_embedding(text_value: str, *_args: object, **_kwargs: object) -> list[float]: + provider_calls["count"] += 1 + return _fake_embedding_for(text_value) + + import engram.embeddings as embeddings_mod + from engram import recall as recall_mod + from engram.api.routes import memory as memory_routes + + monkeypatch.setattr(recall_mod, "generate_embedding", counting_embedding) + monkeypatch.setattr(memory_routes, "generate_embedding", counting_embedding) + monkeypatch.setattr(embeddings_mod, "generate_embedding", counting_embedding) + + await _shadow_compare(client, profiles=["governed", "exploratory"]) + assert provider_calls["count"] == 1 + + +async def test_expanded_neighbor_resolution_is_bulk_without_nplus1(client, monkeypatch): + """Required test 21: one bulk resolution per window — query count stays + constant as the neighbor count grows within the configured bounds.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + import engram.admission_shadow as admission_shadow_mod + + resolve_calls = {"count": 0} + original_resolve = admission_shadow_mod.resolve_bulk_v2_decisions + + async def counting_resolve(*args: Any, **kwargs: Any) -> Any: + resolve_calls["count"] += 1 + return await original_resolve(*args, **kwargs) + + monkeypatch.setattr(admission_shadow_mod, "resolve_bulk_v2_decisions", counting_resolve) + monkeypatch.setattr(settings, "max_graph_neighbors_per_item", 10) + + seed = await _seed_qualified(client, "semantic target bulk seed") + + async def add_neighbors(count: int, prefix: str) -> list[str]: + ids = [] + for i in range(count): + neighbor = await _seed_qualified(client, f"{prefix} neighbor {i:02d}") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "references") + ids.append(neighbor["id"]) + return ids + + few = await add_neighbors(2, "bulk few") + packet_few = await _candidate_packet(client) + assert len(few) == packet_few["expansion"]["admitted_expanded"] + + await add_neighbors(4, "bulk many") # 6 neighbors total, within the caps + packet_many = await _candidate_packet(client) + assert packet_many["expansion"]["admitted_expanded"] == 6 + + # Direct window + neighbor window: exactly two bulk resolutions per + # packet regardless of neighbor count. + assert resolve_calls["count"] == 4 # two packets x two windows + # Merged per-packet query count is constant in neighbor count. + assert packet_few["v2_resolution"]["query_count"] == packet_many[ + "v2_resolution" + ]["query_count"] + assert packet_many["v2_resolution"]["resolved_count"] == 7 # 1 direct + 6 neighbors + + +async def test_graph_and_tunnel_caps_remain_enforced(client, monkeypatch): + """Required test 22: expansion cannot exceed the configured bounded + windows.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + monkeypatch.setattr(settings, "max_graph_expanded_items", 2) + monkeypatch.setattr(settings, "max_graph_neighbors_per_item", 5) + monkeypatch.setattr(settings, "max_tunnel_additions", 1) + + seed = await _seed_qualified(client, "semantic target caps seed") + for i in range(4): + neighbor = await _seed_qualified(client, f"caps graph neighbor {i}") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from", weight=0.9 - i * 0.1) + tunnel_extra = await _seed_qualified( + client, "caps tunnel neighbor", wing="CapsWing", room="src" + ) + # A second admitted direct seed in the tunneled wing drives discovery. + await _seed_qualified( + client, "semantic target caps tunnel seed", wing="CapsFar", room="dst" + ) + await _make_expansion_only(tunnel_extra["id"]) + await _mk_tunnel("CapsWing", "CapsFar") + + governed = await _candidate_packet(client) + # max_graph_expanded_items=2 of the 4 graph neighbors, strongest first. + assert governed["expansion"]["graph_neighbors"] == 2 + # Tunnel additions capped at 1 (the tunnel neighbor; the seed itself is + # excluded as a direct candidate). + assert governed["expansion"]["tunnel_neighbors"] == 1 + assert governed["expansion"]["discovered_neighbors"] <= 3 + + +async def test_equal_score_tie_ordering_is_deterministic(client, monkeypatch): + """Required test 23: two neighbors with identical relevance, utility, + and timestamps keep a stable order (id-ordered) across evaluations.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target tie seed") + anchors = datetime.now(UTC) - timedelta(days=7) + neighbors = [] + for i in range(2): + neighbor = await _remember(client, f"tie neighbor {i}", source_type="extraction") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "references", weight=0.5) + # Identical timestamps/weights/importance BEFORE persisting the V2 + # row, so both neighbors stay current with identical rank inputs. + await _update_item(neighbor["id"], created_at=anchors, valid_from=anchors) + await _persist_v2_row(neighbor["id"]) + neighbors.append(neighbor) + + first = await _candidate_packet(client) + second = await _candidate_packet(client) + order_first = [item["id"] for item in first["items"]] + order_second = [item["id"] for item in second["items"]] + assert order_first == order_second + scores = [item["score"] for item in first["items"]] + assert scores[1] == scores[2] # the tied neighbors + # Equal on every ranking input: the stable order is the id tiebreak. + assert order_first[1:] == sorted(order_first[1:]) + + +# ---- compatibility / read-only ---- + + +async def test_expansion_shadow_comparison_remains_read_only(client, monkeypatch): + """Required tests 24 + 25: with expansion active the comparison writes + nothing — no recall log, no exposure counter, no review/promotion/ + assessment mutation — and the authoritative legacy packet keeps its + legacy-only shape (blended scoring; no signal/evidence/relationship + blocks). Legacy byte-compatibility itself is pinned by the pre-existing + legacy suites (test_relationship_recall / test_graph_recall / + test_tunnel_recall / legacy semantic tests), which this change runs + unchanged.""" + await _skip_without_db() + settings.embedding_provider = "openai" + _patch_embeddings(monkeypatch) + await _enable_tenant_shadow_policy() + + seed = await _seed_qualified(client, "semantic target readonly seed") + neighbor = await _seed_qualified(client, "readonly neighbor") + await _make_expansion_only(neighbor["id"]) + await _link_items(seed["id"], neighbor["id"], "derived_from") + + async def _snapshot_state() -> tuple[int, list[int], dict[str, Any], int]: + async with _test_session_factory() as session: + logs = int(await session.scalar(text("SELECT count(*) FROM recall_logs"))) + counts = list( + ( + await session.execute( + text( + "SELECT recall_count FROM memory_items " + "WHERE id = ANY(CAST(:ids AS uuid[])) ORDER BY id" + ), + {"ids": [seed["id"], neighbor["id"]]}, + ) + ) + .scalars() + .all() + ) + reviews = dict( + ( + await session.execute( + text("SELECT id::text, review_status FROM memory_items") + ) + ).all() + ) + assessments = int( + await session.scalar(text("SELECT count(*) FROM admission_assessments")) + ) + return logs, counts, reviews, assessments + + before = await _snapshot_state() + shadow = await _shadow_compare(client, profiles=["governed", "exploratory"]) + after = await _snapshot_state() + assert before == after + + legacy_served = await _recall(client) + # The authoritative legacy write path is the only one that may write; + # its counts prove the comparison itself wrote nothing (the shadow left + # exactly one recall log and +1 exposure counters, not two of each). + async with _test_session_factory() as session: + logs = int(await session.scalar(text("SELECT count(*) FROM recall_logs"))) + counts = list( + ( + await session.execute( + text( + "SELECT recall_count FROM memory_items " + "WHERE id = ANY(CAST(:ids AS uuid[])) ORDER BY id" + ), + {"ids": [seed["id"], neighbor["id"]]}, + ) + ) + .scalars() + .all() + ) + assert logs == before[0] + 1 + assert counts == [count + 1 for count in before[1]] + # The candidate packet did expand (the boundary under test is real). + governed = next(c for c in shadow["candidates"] if c["profile"] == "governed") + assert governed["expansion"]["admitted_expanded"] == 1 + # The legacy served packet is untouched by the candidate expansion + # machinery: blended scoring, trust_score present, and — where the legacy + # compatibility expansion surfaces the same neighbor — purely legacy + # fields (no signal/evidence/relationship blocks). + assert legacy_served["scoring_version"] == "semantic-v3" + assert legacy_served["recall_profile"] == "legacy" + for item in legacy_served["items"]: + assert "trust_score" in item + assert "relationship" not in item + assert "evidence" not in item + assert "warning_codes" not in item + # And production remains legacy-only (required test 26). + from engram.recall_profiles import CERTIFIED_SERVING_PROFILES + + assert sorted(CERTIFIED_SERVING_PROFILES) == ["legacy"] + resp = await client.post( + "/v1/recall", + json={"mode": "semantic", "query": "q", "recall_profile": "governed"}, + ) + assert resp.status_code == 422 diff --git a/tests/test_relationship_relevance.py b/tests/test_relationship_relevance.py new file mode 100644 index 0000000..21c8065 --- /dev/null +++ b/tests/test_relationship_relevance.py @@ -0,0 +1,203 @@ +"""Unit tests for the candidate-profile relationship-relevance contract +(issue #190 / ENG-RECALL-003D — ``relationship_recall.compute_relationship_relevance``). + +Pure-function coverage, no DB: the invariants the issue states mechanically — + +* importance/source trust/memory confidence/human verification/review state/ + exposure counters/epistemic state are not inputs, so they cannot move + relationship relevance by construction; +* graph/tunnel contribution is bounded to ``[0, 1]`` even for stored edge + weights above 1; +* a direct hit's relevance is never demoted by links and an unlinked direct + item's relevance is exactly its similarity (byte-compatibility with the + pre-#190 signal contract); +* a relationship can only derive relevance from the source-seed score and + its own bounded contribution; +* identical inputs produce identical relevance and ordering. +""" + +from __future__ import annotations + +import inspect + +from engram.config import settings +from engram.relationship_recall import ( + RELATIONSHIP_RELEVANCE_VERSION, + compute_relationship_relevance, +) + + +def _weights() -> tuple[float, float, float]: + return ( + settings.relationship_score_weight_semantic, + settings.relationship_score_weight_relationship, + settings.relationship_score_weight_tunnel, + ) + + +def test_version_is_pinned(): + assert RELATIONSHIP_RELEVANCE_VERSION == "relationship-relevance-v1" + + +def test_unlinked_direct_hit_relevance_is_exactly_its_similarity(): + """Byte-compatibility property: with no graph/tunnel links the relevance + contract must not rescale the direct semantic score — #190 packets + without expandable relationships produce the pre-#190 values.""" + for similarity in (0.0, 0.25, 0.9, 1.0): + result = compute_relationship_relevance( + direct_semantic_score=similarity, + source_seed_score=0.0, + ) + assert result.relevance_score == round(similarity, 4) + assert result.origins == ("semantic",) + assert result.direct is True + assert result.direct_semantic_score == similarity + assert result.graph_contribution == 0.0 + assert result.graph_edge_types == () + assert result.tunnel_labels == () + + +def test_links_never_demote_a_direct_hit(): + """Monotonicity: being additionally relationship-linked can only raise a + direct item's relevance, never lower it below its own similarity.""" + similarity = 0.8 + for links in ( + [], + [("mentions", 0.3)], + [("supports", 1.0)], + [("derived_from", 1.0), ("references", 0.6)], + ): + result = compute_relationship_relevance( + direct_semantic_score=similarity, + source_seed_score=1.0, + graph_links=links, + tunnel_labels=["w1<->w2"] if links else [], + ) + assert result.relevance_score >= similarity + assert 0.0 <= result.relevance_score <= 1.0 + + +def test_graph_contribution_is_bounded_to_unit_interval(): + """A stored edge weight above 1 cannot inflate the contribution; the + relevance stays within [0, 1].""" + result = compute_relationship_relevance( + direct_semantic_score=None, + source_seed_score=1.0, + graph_links=[("derived_from", 7.5)], + tunnel_labels=["t"], + ) + assert result.graph_contribution == 1.0 + w_sem, w_graph, w_tunnel = _weights() + assert result.relevance_score == round(w_sem + w_graph + w_tunnel, 4) + assert result.relevance_score <= 1.0 + + +def test_expansion_only_relevance_is_bounded_by_seed_and_relationship_contract(): + """A relationship cannot create relevance outside the bounded + source-seed/relationship contract: the max attainable relevance is the + weighted combination, never more.""" + w_sem, w_graph, w_tunnel = _weights() + maxed = compute_relationship_relevance( + direct_semantic_score=None, + source_seed_score=1.0, + graph_links=[("derived_from", 1.0)], + tunnel_labels=["t"], + ) + assert maxed.relevance_score == round(w_sem + w_graph + w_tunnel, 4) + assert maxed.relevance_score < 1.0 + + # Weaker seed -> strictly weaker relevance under the same relationships. + weaker = compute_relationship_relevance( + direct_semantic_score=None, + source_seed_score=0.5, + graph_links=[("derived_from", 1.0)], + tunnel_labels=["t"], + ) + assert weaker.relevance_score < maxed.relevance_score + + +def test_strongest_edge_wins_and_edge_types_are_sorted_unique(): + result = compute_relationship_relevance( + direct_semantic_score=None, + source_seed_score=0.9, + graph_links=[ + ("mentions", 0.3), + ("supports", 0.6), + ("supports", 0.2), + ("derived_from", 0.95), + ], + ) + assert result.graph_contribution == 0.95 + assert result.graph_edge_types == ("derived_from", "mentions", "supports") + + +def test_origins_are_ordered_and_deduplicated(): + result = compute_relationship_relevance( + direct_semantic_score=0.4, + source_seed_score=0.7, + graph_links=[("supports", 0.6)], + tunnel_labels=["b", "a", "a"], + ) + assert result.origins == ("semantic", "graph", "tunnel") + assert result.tunnel_labels == ("a", "b") + assert result.direct is True + + +def test_identical_inputs_produce_identical_relevance_and_components(): + kwargs = { + "direct_semantic_score": None, + "source_seed_score": 0.77, + "graph_links": [("supports", 0.6), ("references", 0.4)], + "tunnel_labels": ["ops"], + } + first = compute_relationship_relevance(**kwargs) + second = compute_relationship_relevance(**kwargs) + assert first == second + assert first.payload() == second.payload() + assert first.payload()["version"] == RELATIONSHIP_RELEVANCE_VERSION + + +def test_utility_and_epistemic_inputs_are_structurally_absent(): + """The excluded signal families are not parameters: importance, source + trust, memory confidence, human verification, review state, exposure + counters, and epistemic/risk state cannot be expressed at all.""" + parameters = inspect.signature(compute_relationship_relevance).parameters + assert set(parameters) == { + "direct_semantic_score", + "source_seed_score", + "graph_links", + "tunnel_labels", + } + + +def test_payload_exposes_structured_components(): + w_sem, w_graph, w_tunnel = _weights() + result = compute_relationship_relevance( + direct_semantic_score=None, + source_seed_score=0.8, + graph_links=[("supports", 0.5)], + tunnel_labels=["eng<->ops"], + ) + payload = result.payload() + assert payload["components"] == { + "semantic": round(w_sem * 0.8, 4), + "graph": round(w_graph * 0.5, 4), + "tunnel": round(w_tunnel * 1.0, 4), + } + assert payload["direct"] is False + assert payload["direct_semantic_score"] is None + assert payload["source_seed_score"] == 0.8 + assert payload["relevance_score"] == result.relevance_score + + +def test_direct_item_ignores_source_seed_score_for_its_semantic_component(): + """A merged (direct + expanded) item's semantic component is its own + similarity, not the source seed's.""" + w_sem, w_graph, _ = _weights() + result = compute_relationship_relevance( + direct_semantic_score=0.6, + source_seed_score=1.0, + graph_links=[("supports", 0.8)], + ) + assert result.components["semantic"] == round(w_sem * 0.6, 4) + assert result.components["graph"] == round(w_graph * 0.8, 4)