diff --git a/docs/plans/2026-04-28-028-refactor-retrieval-candidate-service-plan.md b/docs/plans/2026-04-28-028-refactor-retrieval-candidate-service-plan.md new file mode 100644 index 0000000..c5b6ff2 --- /dev/null +++ b/docs/plans/2026-04-28-028-refactor-retrieval-candidate-service-plan.md @@ -0,0 +1,166 @@ +--- +title: Refactor Retrieval Candidate Projection Service +created: 2026-04-28 +status: completed +type: refactor +origin: "$compound-engineering:lfg 抽离 RetrievalService 的 candidate projection helper 到 RetrievalCandidateService，保留 RetrievalService 和 MemoryOrchestrator 兼容 wrapper" +--- + +# Refactor Retrieval Candidate Projection Service + +## Problem Frame + +`RetrievalService` is now the largest service after the orchestrator cleanup: +999 lines on `master`. The boundary assessment in +`docs/architecture/retrieval-service-boundary-assessment.md` identifies the +lowest-risk next split: candidate projection helpers that convert raw storage +records into scored and access-controlled retrieval candidates. + +This phase should extract those helpers into `RetrievalCandidateService` while +keeping `RetrievalService` and `MemoryOrchestrator` compatibility wrappers. + +## Requirements + +- R1: Add `src/opencortex/services/retrieval_candidate_service.py`. +- R2: Move candidate projection helper implementations out of + `RetrievalService`: + - `_score_object_record` + - `_record_passes_acl` + - `_matched_record_anchors` + - `_records_to_matched_contexts` +- R3: Keep `RetrievalService` methods with the same names and signatures as + compatibility wrappers. +- R4: Keep `MemoryOrchestrator` wrappers unchanged; callers and tests must still + call or patch the same orchestrator method names. +- R5: Keep `_execute_object_query` orchestration in `RetrievalService`. +- R6: Preserve scoring, ACL, anchor matching, detail-level content loading, and + `MatchedContext` output shape exactly. +- R7: Add focused tests for `RetrievalCandidateService` where existing object + retrieval tests do not directly lock the service boundary. +- R8: Run object rerank/cone, perf, recall planner, memory/e2e, and style gates. + +## Scope Boundaries + +- Do not split probe/planner/runtime binding in this phase. +- Do not split `_execute_object_query`. +- Do not remove compatibility wrappers from `RetrievalService` or + `MemoryOrchestrator`. +- Do not change retrieval ranking weights, ACL rules, detail-level behavior, or + explain metadata. +- Do not alter HTTP routes or request/response schemas. +- Do not touch frontend code. + +## Current Code Evidence + +- `src/opencortex/services/retrieval_service.py` is 999 lines. +- The candidate helpers currently live in `RetrievalService` near the middle of + the file and are called by `_execute_object_query`. +- `tests/test_perf_fixes.py` patches orchestrator wrappers such as + `_execute_object_query` and `_aggregate_results`. +- `tests/test_object_rerank.py` and `tests/test_object_cone.py` call + `_execute_object_query` through `MemoryOrchestrator`. +- `tests/test_context_manager.py` has multiple direct object-query tests. + +## Key Technical Decisions + +- Add a lazy `_retrieval_candidate_service` property on `RetrievalService`. +- Move implementations into `RetrievalCandidateService`, which holds a + back-reference to the parent `RetrievalService`. +- Keep `RetrievalService` wrappers and preserve `_execute_object_query`'s + existing compatibility path through `MemoryOrchestrator` wrapper calls. +- Let `RetrievalCandidateService` reach orchestrator-owned filesystem through + the parent service (`self._service._fs`) for L2 content fallback. +- Use `TYPE_CHECKING` imports to avoid runtime cycles. + +## Implementation Units + +### U1. Add RetrievalCandidateService and lazy property + +**Goal:** Establish the new owner for candidate scoring/projection helpers. + +**Files:** +- Add: `src/opencortex/services/retrieval_candidate_service.py` +- Modify: `src/opencortex/services/retrieval_service.py` + +**Approach:** +- Add `RetrievalCandidateService(retrieval_service)` with `_service` + back-reference. +- Add `RetrievalService._retrieval_candidate_service` lazy property. +- Move imports needed only by candidate helpers from `retrieval_service.py` if + they become unused there. + +**Test Scenarios:** +- Importing `RetrievalService` still works. +- `RetrievalService.__new__` style fixtures can access wrapper methods without + eager service construction. + +### U2. Move scoring, ACL, anchor, and context projection helpers + +**Goal:** Move helper bodies while preserving compatibility wrappers. + +**Files:** +- Modify: `src/opencortex/services/retrieval_candidate_service.py` +- Modify: `src/opencortex/services/retrieval_service.py` +- Add: `tests/test_retrieval_candidate_service.py` + +**Approach:** +- Move `_score_object_record`, `_record_passes_acl`, `_matched_record_anchors`, + and `_records_to_matched_contexts` implementations into + `RetrievalCandidateService`. +- Replace `RetrievalService` methods with thin delegates. +- Keep static-like call compatibility for `_record_passes_acl` and + `_matched_record_anchors` by leaving wrappers callable on the service. + +**Test Scenarios:** +- `_record_passes_acl` preserves private user and project visibility rules. +- `_matched_record_anchors` returns normalized intersection values capped to 8. +- `_records_to_matched_contexts` preserves `MatchedContext` fields and L2 content + fallback behavior. +- Existing object rerank/cone tests still pass. + +### U3. Validation, review, browser gate, and PR + +**Goal:** Complete the LFG pipeline with focused verification. + +**Validation Commands:** +- `uv run --group dev pytest tests/test_retrieval_candidate_service.py -q` +- `uv run --group dev pytest tests/test_object_rerank.py tests/test_object_cone.py -q` +- `uv run --group dev pytest tests/test_perf_fixes.py -q` +- `uv run --group dev pytest tests/test_recall_planner.py -q` +- `uv run --group dev pytest tests/test_memory_service.py tests/test_e2e_phase1.py -q` +- `uv run --group dev ruff check .` +- `uv run --group dev ruff format --check .` + +## Risks + +| Risk | Mitigation | +|------|------------| +| Retrieval scoring drifts | Move code mechanically and run object rerank/cone tests | +| Tests patch wrappers | Keep both RetrievalService and MemoryOrchestrator wrapper names | +| L2 content fallback loses filesystem access | Service reaches parent retrieval service `_fs` property | +| Import cycle between services | Use `TYPE_CHECKING` and local/lazy service imports | +| Split expands scope into `_execute_object_query` | Keep main method in place and only delegate helper calls | + +## Observed Results + +- Added `src/opencortex/services/retrieval_candidate_service.py`. +- `RetrievalService` compatibility wrappers remain in place. +- `MemoryOrchestrator` compatibility wrappers remain unchanged. +- `_execute_object_query` remains owned by `RetrievalService`. +- `RetrievalService` reduced from 999 lines to 886 lines. +- `RetrievalCandidateService` is 225 lines. + +## Validation Results + +- `uv run --group dev pytest tests/test_retrieval_candidate_service.py -q`: + passed, 3 tests. +- `uv run --group dev pytest tests/test_object_rerank.py tests/test_object_cone.py -q`: + passed, 2 tests. +- `uv run --group dev pytest tests/test_perf_fixes.py -q`: passed, 13 tests + with one pre-existing fastembed warning. +- `uv run --group dev pytest tests/test_recall_planner.py -q`: passed, + 25 tests. +- `uv run --group dev pytest tests/test_memory_service.py tests/test_e2e_phase1.py -q`: + passed, 54 tests. +- `uv run --group dev ruff check .`: passed. +- `uv run --group dev ruff format --check .`: passed. diff --git a/src/opencortex/services/retrieval_candidate_service.py b/src/opencortex/services/retrieval_candidate_service.py new file mode 100644 index 0000000..280186e --- /dev/null +++ b/src/opencortex/services/retrieval_candidate_service.py @@ -0,0 +1,225 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Candidate scoring and projection helpers for retrieval.""" + +from __future__ import annotations + +import asyncio +import math +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +from opencortex.intent import RetrievalPlan +from opencortex.intent.retrieval_support import ( + anchor_rerank_bonus, + record_anchor_groups, +) +from opencortex.retrieve.types import ( + ContextType, + DetailLevel, + MatchedContext, + TypedQuery, +) + +if TYPE_CHECKING: + from opencortex.services.retrieval_service import RetrievalService + + +class RetrievalCandidateService: + """Own candidate scoring, ACL, anchor matching, and context projection.""" + + def __init__(self, retrieval_service: "RetrievalService") -> None: + self._service = retrieval_service + + def _score_object_record( + self, + *, + record: Dict[str, Any], + typed_query: TypedQuery, + retrieve_plan: Optional[RetrievalPlan], + query_anchor_groups: Dict[str, set[str]], + probe_candidate_ranks: Dict[str, int], + cone_weight: float, + uri_path_costs: Optional[Dict[str, float]] = None, + ) -> tuple[float, str]: + """Fuse URI path score (primary) with object-aware boosts.""" + leaf_uri = str(record.get("uri", "") or "") + if uri_path_costs is not None and leaf_uri in uri_path_costs: + score = 1.0 - uri_path_costs[leaf_uri] + else: + score = float(record.get("_score", record.get("score", 0.0)) or 0.0) + reasons: List[str] = [] + target_kinds = ( + [kind.value for kind in retrieve_plan.target_memory_kinds] + if retrieve_plan is not None + else [] + ) + record_kind = str(record.get("memory_kind", "")) + if record_kind in target_kinds: + kind_rank = target_kinds.index(record_kind) + score += 0.14 * (len(target_kinds) - kind_rank) / max(len(target_kinds), 1) + reasons.append("kind") + + anchor_bonus, anchor_reasons = anchor_rerank_bonus( + query_anchor_groups=query_anchor_groups, + record_anchor_groups=record_anchor_groups(record), + ) + if anchor_bonus > 0: + score += anchor_bonus + reasons.extend(anchor_reasons) + + probe_rank = probe_candidate_ranks.get(str(record.get("uri", "") or "")) + if probe_rank is not None: + score += max(0.04, 0.14 - min(probe_rank, 5) * 0.02) + reasons.append("probe") + + if typed_query.target_directories and any( + str(record.get("uri", "")).startswith(prefix) + for prefix in typed_query.target_directories + ): + score += 0.06 + reasons.append("scope") + + if typed_query.target_doc_id and ( + str(record.get("source_doc_id", "")) == typed_query.target_doc_id + ): + score += 0.08 + reasons.append("doc") + + reward = float(record.get("reward_score", 0.0) or 0.0) + if reward: + score += max(min(0.06, reward * 0.03), -0.03) + reasons.append("reward") + + active_count = int(record.get("active_count", 0) or 0) + if active_count > 0: + score += min(0.05, math.log1p(active_count) * 0.01) + reasons.append("hot") + + cone_bonus = float(record.get("_cone_bonus", 0.0) or 0.0) + if cone_weight > 0.0 and cone_bonus > 0.0: + score += min(0.30, cone_weight * min(1.0, cone_bonus)) + reasons.append("cone") + + return score, ",".join(reasons) or "semantic" + + @staticmethod + def _record_passes_acl( + record: Dict[str, Any], + tenant_id: str, + user_id: str, + project_id: str, + ) -> bool: + """Return True if record passes tenant/scope/project access control.""" + r_tenant = str(record.get("source_tenant_id", "") or "") + if tenant_id and r_tenant and r_tenant != tenant_id: + return False + if record.get("scope") == "private" and record.get("source_user_id") != user_id: + return False + r_project = str(record.get("project_id", "") or "") + return not ( + project_id + and project_id != "public" + and r_project not in (project_id, "public", "") + ) + + @staticmethod + def _matched_record_anchors( + *, + record: Dict[str, Any], + query_anchor_groups: Dict[str, set[str]], + ) -> List[str]: + """Return normalized query anchors that concretely matched this record.""" + if not query_anchor_groups: + return [] + matched: List[str] = [] + record_groups = record_anchor_groups(record) + for kind, query_values in query_anchor_groups.items(): + record_values = record_groups.get(kind, set()) + for value in sorted(query_values.intersection(record_values)): + if value not in matched: + matched.append(value) + return matched[:8] + + async def _records_to_matched_contexts( + self, + *, + candidates: List[Dict[str, Any]], + context_type: ContextType, + detail_level: DetailLevel, + ) -> List[MatchedContext]: + """Convert raw store records into MatchedContext objects.""" + + async def _build_one(record: Dict[str, Any]) -> MatchedContext: + uri = str(record.get("uri", "")) + overview = None + if detail_level in (DetailLevel.L1, DetailLevel.L2): + overview = str(record.get("overview", "") or "") or None + + content = None + if detail_level == DetailLevel.L2: + content = str(record.get("content", "") or "") or None + if content is None and self._service._fs: + try: + content = await self._service._fs.read_file(f"{uri}/content.md") + except Exception: + content = None + + effective_type = context_type + if context_type == ContextType.ANY: + try: + effective_type = ContextType( + str(record.get("context_type", "memory")) + ) + except ValueError: + effective_type = ContextType.MEMORY + + return MatchedContext( + uri=uri, + context_type=effective_type, + is_leaf=bool(record.get("is_leaf", False)), + abstract=str(record.get("abstract", "") or ""), + overview=overview, + content=content, + keywords=str(record.get("keywords", "") or ""), + category=str(record.get("category", "") or ""), + score=float( + record.get("_final_score", record.get("_score", 0.0)) or 0.0 + ), + match_reason=str(record.get("_match_reason", "") or ""), + session_id=str(record.get("session_id", "") or ""), + source_doc_id=record.get("source_doc_id"), + source_doc_title=record.get("source_doc_title"), + source_section_path=record.get("source_section_path"), + source_uri=( + dict(record.get("meta") or {}).get("source_uri") + if isinstance(record.get("meta"), dict) + else None + ), + msg_range=( + dict(record.get("meta") or {}).get("msg_range") + if isinstance(record.get("meta"), dict) + else None + ), + recomposition_stage=( + dict(record.get("meta") or {}).get("recomposition_stage") + if isinstance(record.get("meta"), dict) + else None + ), + layer=( + dict(record.get("meta") or {}).get("layer") + if isinstance(record.get("meta"), dict) + else None + ), + matched_anchors=list(record.get("_matched_anchors", []) or []), + cone_used=bool(record.get("_cone_used", False)), + path_source=record.get("_path_source") or None, + path_cost=( + float(record["_path_cost"]) + if record.get("_path_cost") is not None + else None + ), + path_breakdown=record.get("_path_breakdown") or None, + relations=[], + ) + + matches = await asyncio.gather(*[_build_one(record) for record in candidates]) + return list(matches) diff --git a/src/opencortex/services/retrieval_service.py b/src/opencortex/services/retrieval_service.py index 2b85067..9e5b2fe 100644 --- a/src/opencortex/services/retrieval_service.py +++ b/src/opencortex/services/retrieval_service.py @@ -10,7 +10,6 @@ import asyncio import logging -import math import time from typing import TYPE_CHECKING, Any, Dict, List, Optional @@ -26,12 +25,10 @@ SearchResult, ) from opencortex.intent.retrieval_support import ( - anchor_rerank_bonus, build_probe_scope_input, build_scope_filter, build_start_point_filter, merge_filter_clauses, - record_anchor_groups, ) from opencortex.intent.retrieval_support import ( probe_candidate_ranks as build_probe_candidate_ranks, @@ -112,6 +109,19 @@ def _analyzer(self) -> Any: def _llm_completion(self) -> Any: return self._orch._llm_completion + @property + def _retrieval_candidate_service(self) -> "RetrievalCandidateService": + """Lazy-built service for candidate scoring/projection helpers.""" + from opencortex.services.retrieval_candidate_service import ( + RetrievalCandidateService, + ) + + cached = getattr(self, "_retrieval_candidate_service_instance", None) + if cached is None: + cached = RetrievalCandidateService(self) + self._retrieval_candidate_service_instance = cached + return cached + def _ensure_init(self) -> None: self._orch._ensure_init() @@ -317,9 +327,7 @@ async def _apply_cone_rerank( collection = self._get_collection() if not self._entity_index.is_ready(collection): - await self._entity_index.build_for_collection( - self._storage, collection - ) + await self._entity_index.build_for_collection(self._storage, collection) if not self._entity_index.is_ready(collection): return records, False @@ -383,65 +391,15 @@ def _score_object_record( uri_path_costs: Optional[Dict[str, float]] = None, ) -> tuple[float, str]: """Fuse URI path score (primary) with object-aware boosts.""" - leaf_uri = str(record.get("uri", "") or "") - if uri_path_costs is not None and leaf_uri in uri_path_costs: - score = 1.0 - uri_path_costs[leaf_uri] - else: - score = float(record.get("_score", record.get("score", 0.0)) or 0.0) - reasons: List[str] = [] - target_kinds = ( - [kind.value for kind in retrieve_plan.target_memory_kinds] - if retrieve_plan is not None - else [] - ) - record_kind = str(record.get("memory_kind", "")) - if record_kind in target_kinds: - kind_rank = target_kinds.index(record_kind) - score += 0.14 * (len(target_kinds) - kind_rank) / max(len(target_kinds), 1) - reasons.append("kind") - - anchor_bonus, anchor_reasons = anchor_rerank_bonus( + return self._retrieval_candidate_service._score_object_record( + record=record, + typed_query=typed_query, + retrieve_plan=retrieve_plan, query_anchor_groups=query_anchor_groups, - record_anchor_groups=record_anchor_groups(record), + probe_candidate_ranks=probe_candidate_ranks, + cone_weight=cone_weight, + uri_path_costs=uri_path_costs, ) - if anchor_bonus > 0: - score += anchor_bonus - reasons.extend(anchor_reasons) - - probe_rank = probe_candidate_ranks.get(str(record.get("uri", "") or "")) - if probe_rank is not None: - score += max(0.04, 0.14 - min(probe_rank, 5) * 0.02) - reasons.append("probe") - - if typed_query.target_directories and any( - str(record.get("uri", "")).startswith(prefix) - for prefix in typed_query.target_directories - ): - score += 0.06 - reasons.append("scope") - - if typed_query.target_doc_id and ( - str(record.get("source_doc_id", "")) == typed_query.target_doc_id - ): - score += 0.08 - reasons.append("doc") - - reward = float(record.get("reward_score", 0.0) or 0.0) - if reward: - score += max(min(0.06, reward * 0.03), -0.03) - reasons.append("reward") - - active_count = int(record.get("active_count", 0) or 0) - if active_count > 0: - score += min(0.05, math.log1p(active_count) * 0.01) - reasons.append("hot") - - cone_bonus = float(record.get("_cone_bonus", 0.0) or 0.0) - if cone_weight > 0.0 and cone_bonus > 0.0: - score += min(0.30, cone_weight * min(1.0, cone_bonus)) - reasons.append("cone") - - return score, ",".join(reasons) or "semantic" @staticmethod def _record_passes_acl( @@ -451,19 +409,16 @@ def _record_passes_acl( project_id: str, ) -> bool: """Return True if record passes tenant/scope/project access control.""" - r_tenant = str(record.get("source_tenant_id", "") or "") - if tenant_id and r_tenant and r_tenant != tenant_id: - return False - if record.get("scope") == "private" and record.get("source_user_id") != user_id: - return False - r_project = str(record.get("project_id", "") or "") - if ( - project_id - and project_id != "public" - and r_project not in (project_id, "public", "") - ): - return False - return True + from opencortex.services.retrieval_candidate_service import ( + RetrievalCandidateService, + ) + + return RetrievalCandidateService._record_passes_acl( + record, + tenant_id, + user_id, + project_id, + ) @staticmethod def _matched_record_anchors( @@ -472,16 +427,14 @@ def _matched_record_anchors( query_anchor_groups: Dict[str, set[str]], ) -> List[str]: """Return normalized query anchors that concretely matched this record.""" - if not query_anchor_groups: - return [] - matched: List[str] = [] - record_groups = record_anchor_groups(record) - for kind, query_values in query_anchor_groups.items(): - record_values = record_groups.get(kind, set()) - for value in sorted(query_values.intersection(record_values)): - if value not in matched: - matched.append(value) - return matched[:8] + from opencortex.services.retrieval_candidate_service import ( + RetrievalCandidateService, + ) + + return RetrievalCandidateService._matched_record_anchors( + record=record, + query_anchor_groups=query_anchor_groups, + ) async def _records_to_matched_contexts( self, @@ -491,79 +444,11 @@ async def _records_to_matched_contexts( detail_level: DetailLevel, ) -> List[MatchedContext]: """Convert raw store records into MatchedContext objects.""" - - async def _build_one(record: Dict[str, Any]) -> MatchedContext: - uri = str(record.get("uri", "")) - overview = None - if detail_level in (DetailLevel.L1, DetailLevel.L2): - overview = str(record.get("overview", "") or "") or None - - content = None - if detail_level == DetailLevel.L2: - content = str(record.get("content", "") or "") or None - if content is None and self._fs: - try: - content = await self._fs.read_file(f"{uri}/content.md") - except Exception: - content = None - - effective_type = context_type - if context_type == ContextType.ANY: - try: - effective_type = ContextType(str(record.get("context_type", "memory"))) - except ValueError: - effective_type = ContextType.MEMORY - - return MatchedContext( - uri=uri, - context_type=effective_type, - is_leaf=bool(record.get("is_leaf", False)), - abstract=str(record.get("abstract", "") or ""), - overview=overview, - content=content, - keywords=str(record.get("keywords", "") or ""), - category=str(record.get("category", "") or ""), - score=float( - record.get("_final_score", record.get("_score", 0.0)) or 0.0 - ), - match_reason=str(record.get("_match_reason", "") or ""), - session_id=str(record.get("session_id", "") or ""), - source_doc_id=record.get("source_doc_id"), - source_doc_title=record.get("source_doc_title"), - source_section_path=record.get("source_section_path"), - source_uri=( - dict(record.get("meta") or {}).get("source_uri") - if isinstance(record.get("meta"), dict) - else None - ), - msg_range=( - dict(record.get("meta") or {}).get("msg_range") - if isinstance(record.get("meta"), dict) - else None - ), - recomposition_stage=( - dict(record.get("meta") or {}).get("recomposition_stage") - if isinstance(record.get("meta"), dict) - else None - ), - layer=( - dict(record.get("meta") or {}).get("layer") - if isinstance(record.get("meta"), dict) - else None - ), - matched_anchors=list(record.get("_matched_anchors", []) or []), - cone_used=bool(record.get("_cone_used", False)), - path_source=record.get("_path_source") or None, - path_cost=( - float(record["_path_cost"]) - if record.get("_path_cost") is not None - else None - ), - path_breakdown=record.get("_path_breakdown") or None, - relations=[], - ) - - return list(await asyncio.gather(*[_build_one(record) for record in candidates])) + return await self._retrieval_candidate_service._records_to_matched_contexts( + candidates=candidates, + context_type=context_type, + detail_level=detail_level, + ) async def _execute_object_query( self, @@ -613,11 +498,13 @@ async def _execute_object_query( "conds": parent_uris, } elif retrieve_plan.scope_level == ScopeLevel.SESSION_ONLY: - session_ids = sorted({ - sp.session_id - for sp in probe_result.starting_points - if sp.session_id - }) + session_ids = sorted( + { + sp.session_id + for sp in probe_result.starting_points + if sp.session_id + } + ) if session_ids: scope_only_filter = { "op": "must", @@ -625,11 +512,13 @@ async def _execute_object_query( "conds": session_ids, } elif retrieve_plan.scope_level == ScopeLevel.DOCUMENT_ONLY: - doc_ids = sorted({ - sp.source_doc_id - for sp in probe_result.starting_points - if sp.source_doc_id - }) + doc_ids = sorted( + { + sp.source_doc_id + for sp in probe_result.starting_points + if sp.source_doc_id + } + ) if doc_ids: scope_only_filter = { "op": "must", @@ -649,7 +538,11 @@ async def _execute_object_query( search_filter, start_point_filter, scope_only_filter, - {"op": "must", "field": "retrieval_surface", "conds": ["anchor_projection"]}, + { + "op": "must", + "field": "retrieval_surface", + "conds": ["anchor_projection"], + }, ) fp_filter_merged = merge_filter_clauses( search_filter, @@ -715,7 +608,9 @@ async def _execute_object_query( if isinstance(search_results[0], Exception): logger.debug("[RetrievalService] leaf search failed: %s", search_results[0]) if isinstance(search_results[1], Exception): - logger.debug("[RetrievalService] anchor search failed: %s", search_results[1]) + logger.debug( + "[RetrievalService] anchor search failed: %s", search_results[1] + ) if isinstance(search_results[2], Exception): logger.debug("[RetrievalService] fp search failed: %s", search_results[2]) @@ -730,9 +625,7 @@ def _get_target_uri(hit: Dict[str, Any]) -> str: known_leaf_uris = {str(r.get("uri", "")) for r in leaf_hits if r.get("uri")} projected_uris = { - _get_target_uri(h) - for h in anchor_hits + fp_hits - if _get_target_uri(h) + _get_target_uri(h) for h in anchor_hits + fp_hits if _get_target_uri(h) } missing_uris = [u for u in projected_uris if u and u not in known_leaf_uris] diff --git a/tests/test_retrieval_candidate_service.py b/tests/test_retrieval_candidate_service.py new file mode 100644 index 0000000..13a3e1c --- /dev/null +++ b/tests/test_retrieval_candidate_service.py @@ -0,0 +1,137 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Tests for ``RetrievalCandidateService``.""" + +from __future__ import annotations + +import asyncio +import unittest +from collections.abc import Awaitable +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from opencortex.retrieve.types import ContextType, DetailLevel +from opencortex.services.retrieval_candidate_service import RetrievalCandidateService + + +class TestRetrievalCandidateService(unittest.TestCase): + """Verify candidate projection helper behavior.""" + + def _run(self, coro: Awaitable[Any]) -> Any: + return asyncio.run(coro) + + def _make_service(self) -> RetrievalCandidateService: + retrieval_service = MagicMock() + retrieval_service._fs = None + return RetrievalCandidateService(retrieval_service) + + def test_record_passes_acl_preserves_private_and_project_rules(self) -> None: + """ACL allows own private/project records and rejects mismatches.""" + service = self._make_service() + self.assertTrue( + service._record_passes_acl( + { + "scope": "private", + "source_tenant_id": "tenant-a", + "source_user_id": "user-a", + "project_id": "project-a", + }, + "tenant-a", + "user-a", + "project-a", + ) + ) + self.assertFalse( + service._record_passes_acl( + { + "scope": "private", + "source_tenant_id": "tenant-a", + "source_user_id": "user-b", + "project_id": "project-a", + }, + "tenant-a", + "user-a", + "project-a", + ) + ) + self.assertFalse( + service._record_passes_acl( + { + "scope": "shared", + "source_tenant_id": "tenant-b", + "source_user_id": "user-a", + "project_id": "project-a", + }, + "tenant-a", + "user-a", + "project-a", + ) + ) + self.assertFalse( + service._record_passes_acl( + { + "scope": "shared", + "source_tenant_id": "tenant-a", + "source_user_id": "user-a", + "project_id": "project-b", + }, + "tenant-a", + "user-a", + "project-a", + ) + ) + + def test_matched_record_anchors_returns_intersections_capped_to_eight(self) -> None: + """Anchor projection returns deterministic matching query anchors.""" + service = self._make_service() + record = { + "abstract_json": { + "anchors": [ + {"anchor_type": "entity", "text": f"v{i}"} for i in range(10) + ] + } + } + query_groups = {"entity": {f"v{i}" for i in range(10)}} + + result = service._matched_record_anchors( + record=record, + query_anchor_groups=query_groups, + ) + + self.assertEqual(result, [f"v{i}" for i in range(8)]) + + def test_records_to_matched_contexts_uses_l2_filesystem_fallback(self) -> None: + """L2 projection reads content from CortexFS when payload lacks content.""" + retrieval_service = MagicMock() + retrieval_service._fs.read_file = AsyncMock(return_value="file content") + service = RetrievalCandidateService(retrieval_service) + + result = self._run( + service._records_to_matched_contexts( + candidates=[ + { + "uri": "opencortex://tenant/user/memories/events/item", + "context_type": "memory", + "is_leaf": True, + "abstract": "abstract", + "overview": "overview", + "_final_score": 0.75, + "_match_reason": "semantic", + "meta": {"source_uri": "source", "msg_range": [1, 2]}, + } + ], + context_type=ContextType.ANY, + detail_level=DetailLevel.L2, + ) + ) + + self.assertEqual(len(result), 1) + match = result[0] + self.assertEqual(match.context_type, ContextType.MEMORY) + self.assertEqual(match.content, "file content") + self.assertEqual(match.overview, "overview") + self.assertEqual(match.score, 0.75) + self.assertEqual(match.source_uri, "source") + self.assertEqual(match.msg_range, [1, 2]) + retrieval_service._fs.read_file.assert_awaited_once_with( + "opencortex://tenant/user/memories/events/item/content.md" + )