refactor(retrieval): extract candidate projection service

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.

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)