diff --git a/src/backend/integrations/internal_mcp/orchestrator.py b/src/backend/integrations/internal_mcp/orchestrator.py
index 263c71f6c..1e5590b6c 100644
--- a/src/backend/integrations/internal_mcp/orchestrator.py
+++ b/src/backend/integrations/internal_mcp/orchestrator.py
@@ -19714,6 +19714,73 @@ def _apply_turn_scoped_tool_payload_support(
                                 }
                             )
 
+        if (
+            lowered_tool_name == "find_relations_with_argument"
+            and isinstance(matched_tool_defaults, Mapping)
+            and bool(
+                matched_tool_defaults.get(
+                    "__expand_predicate_family_from_vontology",
+                    False,
+                )
+            )
+        ):
+            raw_predicate_filter = payload.get("predicate_filter")
+            predicate_filter = [
+                str(item).strip()
+                for item in (
+                    raw_predicate_filter
+                    if isinstance(raw_predicate_filter, Sequence)
+                    and not isinstance(
+                        raw_predicate_filter,
+                        (str, bytes, bytearray),
+                    )
+                    else []
+                )
+                if isinstance(item, str) and str(item).strip()
+            ]
+            if predicate_filter:
+                from ...services.predicate_family_vontology_service import (
+                    expand_relation_predicate_family,
+                )
+
+                expansion = expand_relation_predicate_family(
+                    predicate_filter,
+                    max_predicates=matched_tool_defaults.get(
+                        "__predicate_family_max_predicates",
+                        32,
+                    ),
+                )
+                expanded_predicate_ids = [
+                    str(item).strip()
+                    for item in (expansion.get("predicate_ids") or [])
+                    if isinstance(item, str) and str(item).strip()
+                ]
+                if expanded_predicate_ids:
+                    payload["predicate_filter"] = expanded_predicate_ids
+                if isinstance(data, MutableMapping):
+                    data["relation_predicate_family_context"] = dict(expansion)
+                    events = data.setdefault("tool_payload_support_events", [])
+                    if isinstance(events, list):
+                        events.append(
+                            {
+                                "tool": lowered_tool_name,
+                                "field": "predicate_filter",
+                                "source": "vontology_predicate_schema",
+                                "seed_predicate_ids": predicate_filter,
+                                "predicate_ids": expanded_predicate_ids,
+                                "family_concept_ids": list(
+                                    expansion.get("family_concept_ids") or []
+                                ),
+                                "status": expansion.get("status"),
+                                "relationship_extent_index_status": expansion.get(
+                                    "relationship_extent_index_status"
+                                ),
+                                "family_discovery_source": expansion.get(
+                                    "family_discovery_source"
+                                ),
+                            }
+                        )
+
         if lowered_tool_name != "search_knowledge_base":
             return
 
@@ -20848,6 +20915,9 @@ def _shape_find_relations_with_argument_payload_for_llm(
         )
         compact_hits: list[dict[str, Any]] = []
         predicate_values: list[str] = []
+        focal_concept_id = str(payload.get("concept_id") or "").strip() or None
+        subject_argument_index = 1
+        first_object_argument_index = 2
 
         def _preview_name(preview: Any) -> str | None:
             if not isinstance(preview, Mapping):
@@ -20889,7 +20959,73 @@ def _preview_type_ids(preview: Any) -> list[str]:
                 if predicate_value not in predicate_values:
                     predicate_values.append(predicate_value)
 
-        for hit in hits[:max_hits]:
+        selected_hits: list[Mapping[str, Any]] = []
+        seen_related_concept_ids: set[str] = set()
+        related_predicates_by_concept_id: dict[str, list[str]] = {}
+        related_directions_by_concept_id: dict[str, list[str]] = {}
+        compacted_duplicate_hit_count = 0
+        for hit in hits:
+            source_id = str(hit.get("source_concept_id") or "").strip()
+            target_preview = hit.get("target_concept_preview")
+            target_id = (
+                _preview_concept_id(target_preview)
+                or str(hit.get("target_value") or "").strip()
+            )
+            raw_indexes = hit.get("argument_indexes")
+            indexes = (
+                [int(index) for index in raw_indexes if isinstance(index, (int, float))]
+                if isinstance(raw_indexes, list)
+                else []
+            )
+            related_concept_id: str | None = None
+            direction_from_focal_entity: str | None = None
+            if (
+                focal_concept_id
+                and source_id == focal_concept_id
+                and target_id.startswith("#V#")
+                and (not indexes or subject_argument_index in indexes)
+            ):
+                related_concept_id = target_id
+                direction_from_focal_entity = "outgoing"
+            elif (
+                focal_concept_id
+                and target_id == focal_concept_id
+                and source_id.startswith("#V#")
+                and (
+                    not indexes
+                    or any(index >= first_object_argument_index for index in indexes)
+                )
+            ):
+                related_concept_id = source_id
+                direction_from_focal_entity = "incoming"
+            if related_concept_id:
+                predicate_id = str(
+                    hit.get("predicate_concept_id") or ""
+                ).strip()
+                related_predicates = related_predicates_by_concept_id.setdefault(
+                    related_concept_id,
+                    [],
+                )
+                if predicate_id and predicate_id not in related_predicates:
+                    related_predicates.append(predicate_id)
+                related_directions = related_directions_by_concept_id.setdefault(
+                    related_concept_id,
+                    [],
+                )
+                if (
+                    direction_from_focal_entity
+                    and direction_from_focal_entity not in related_directions
+                ):
+                    related_directions.append(direction_from_focal_entity)
+                if related_concept_id in seen_related_concept_ids:
+                    compacted_duplicate_hit_count += 1
+                    continue
+                seen_related_concept_ids.add(related_concept_id)
+            if len(selected_hits) >= max_hits:
+                continue
+            selected_hits.append(hit)
+
+        for hit in selected_hits:
             compact_hit: dict[str, Any] = {}
             source_concept_id = hit.get("source_concept_id")
             if isinstance(source_concept_id, str) and source_concept_id.strip():
@@ -20944,6 +21080,62 @@ def _preview_type_ids(preview: Any) -> list[str]:
             if target_type_ids:
                 compact_hit["target_type_ids"] = target_type_ids
 
+            compact_source_id = compact_hit.get("source_concept_id")
+            compact_target_id = compact_hit.get("target_concept_id")
+            compact_indexes = compact_hit.get("argument_indexes") or []
+            if (
+                focal_concept_id
+                and compact_source_id == focal_concept_id
+                and isinstance(compact_target_id, str)
+                and compact_target_id.startswith("#V#")
+                and (not compact_indexes or subject_argument_index in compact_indexes)
+            ):
+                compact_hit["direction_from_focal_entity"] = "outgoing"
+                compact_hit["related_concept_id"] = compact_target_id
+                if target_name:
+                    compact_hit["related_name"] = target_name
+                if target_type_ids:
+                    compact_hit["related_type_ids"] = target_type_ids
+            elif (
+                focal_concept_id
+                and compact_target_id == focal_concept_id
+                and isinstance(compact_source_id, str)
+                and compact_source_id.startswith("#V#")
+                and (
+                    not compact_indexes
+                    or any(
+                        index >= first_object_argument_index
+                        for index in compact_indexes
+                        if isinstance(index, int)
+                    )
+                )
+            ):
+                compact_hit["direction_from_focal_entity"] = "incoming"
+                compact_hit["related_concept_id"] = compact_source_id
+                if source_name:
+                    compact_hit["related_name"] = source_name
+                if source_type_ids:
+                    compact_hit["related_type_ids"] = source_type_ids
+
+            related_concept_id = compact_hit.get("related_concept_id")
+            if isinstance(related_concept_id, str):
+                related_predicate_ids = related_predicates_by_concept_id.get(
+                    related_concept_id,
+                    [],
+                )
+                if related_predicate_ids:
+                    compact_hit["related_predicate_concept_ids"] = list(
+                        related_predicate_ids
+                    )
+                related_directions = related_directions_by_concept_id.get(
+                    related_concept_id,
+                    [],
+                )
+                if related_directions:
+                    compact_hit["directions_from_focal_entity"] = list(
+                        related_directions
+                    )
+
             text_snippet = hit.get("text_snippet")
             if isinstance(text_snippet, str) and text_snippet.strip():
                 compact_hit["text_preview"] = text_snippet.strip()[:max_text_chars]
@@ -20970,7 +21162,7 @@ def _preview_type_ids(preview: Any) -> list[str]:
             if not hits
             else (
                 f"{len(predicate_values)} distinct predicate(s) were observed across "
-                f"{len(compact_hits)} shown relation hit(s)."
+                f"{len(compact_hits)} shown unique related concept(s)."
             )
         )
 
@@ -20984,6 +21176,18 @@ def _preview_type_ids(preview: Any) -> list[str]:
             "hits": compact_hits,
             "retrieval_diagnostics": {"note": diagnostics_note},
         }
+        omitted_related_concept_count = max(
+            0,
+            len(seen_related_concept_ids) - len(compact_hits),
+        )
+        if compacted_duplicate_hit_count:
+            compact_payload["compacted_duplicate_hit_count"] = (
+                compacted_duplicate_hit_count
+            )
+        if omitted_related_concept_count:
+            compact_payload["omitted_related_concept_count"] = (
+                omitted_related_concept_count
+            )
         return {
             key: value
             for key, value in compact_payload.items()
@@ -28074,6 +28278,22 @@ def _build_find_relations_with_argument_follow_up_lines(
         for hit in hits[:8]:
             if not isinstance(hit, Mapping):
                 continue
+            related_name = hit.get("related_name")
+            related_concept_id = hit.get("related_concept_id")
+            if (
+                isinstance(related_name, str)
+                and related_name.strip()
+                and isinstance(related_concept_id, str)
+                and related_concept_id.strip()
+            ):
+                related_label = f"{related_name.strip()} [{related_concept_id.strip()}]"
+            else:
+                related_label = related_name or related_concept_id
+            related_types = [
+                str(type_id).strip()
+                for type_id in (hit.get("related_type_ids") or [])[:3]
+                if isinstance(type_id, str) and str(type_id).strip()
+            ]
             source_label = hit.get("source_name") or hit.get("source_concept_id")
             predicate = hit.get("predicate_concept_id")
             target_label = (
@@ -28090,16 +28310,21 @@ def _build_find_relations_with_argument_follow_up_lines(
                 and isinstance(target_label, str)
                 and target_label.strip()
             ):
-                target_types = [
-                    str(type_id).strip()
-                    for type_id in (hit.get("target_type_ids") or [])[:3]
-                    if isinstance(type_id, str) and str(type_id).strip()
-                ]
                 type_suffix = (
-                    f" [types: {', '.join(target_types)}]" if target_types else ""
+                    f" [related types: {', '.join(related_types)}]"
+                    if related_types
+                    else ""
+                )
+                direction = hit.get("direction_from_focal_entity")
+                related_prefix = (
+                    f"related={related_label.strip()}; "
+                    if isinstance(related_label, str) and related_label.strip()
+                    else ""
                 )
                 fragments.append(
-                    f"{source_label.strip()} via {predicate.strip()} -> {target_label.strip()}{type_suffix}"
+                    f"{related_prefix}{source_label.strip()} via {predicate.strip()} "
+                    f"-> {target_label.strip()} ({direction or 'direction unspecified'})"
+                    f"{type_suffix}"
                 )
                 continue
             if (
diff --git a/src/backend/services/concept_relation_service.py b/src/backend/services/concept_relation_service.py
index fb93ba47a..f83908b33 100644
--- a/src/backend/services/concept_relation_service.py
+++ b/src/backend/services/concept_relation_service.py
@@ -35,6 +35,7 @@
 _ARG_INDEX_FIRST_OBJECT = 2
 _DEFAULT_LIMIT = 200
 _MAX_LIMIT = 500
+_CANONICAL_EXACT_PREDICATE_FALLBACK_MAX_TIME_MS = 8_000
 _UNCERTAINTY_MODE_ASSERTED_ONLY = "asserted_only"
 _UNCERTAINTY_MODE_UNCERTAIN_ONLY = "uncertain_only"
 _UNCERTAINTY_MODE_INCLUDE_UNCERTAIN = "include_uncertain"
@@ -401,30 +402,31 @@ def find_relations_with_argument(
                         True,
                         preview_cache,
                     )
-                hits.append(
-                    {
-                        "source_concept_id": resolved_concept_id,
-                        "predicate_concept_id": predicate_id,
-                        "relation_kind": "binary",
-                        "argument_indexes": matched_indexes,
-                        "target_value": target_value,
-                        "target_concept_preview": target_preview,
-                        "relation_metadata": {
-                            "relation_id": f"struct::{resolved_concept_id}::{predicate_id}::{target_index}",
-                            "updated_at": source_updated_at,
-                            "match_type": "exact",
-                        },
-                        "access_granted": source_preview is not None
-                        or not include_concept_preview,
-                        "follow_up_actions": _build_follow_up_actions(
-                            [target_value],
-                            exclude={resolved_concept_id},
-                        ),
-                        "score": 1.0,
-                        "is_asserted": True,
-                        "relation_state": "asserted",
-                    }
-                )
+                hit = {
+                    "source_concept_id": resolved_concept_id,
+                    "predicate_concept_id": predicate_id,
+                    "relation_kind": "binary",
+                    "argument_indexes": matched_indexes,
+                    "target_value": target_value,
+                    "target_concept_preview": target_preview,
+                    "relation_metadata": {
+                        "relation_id": f"struct::{resolved_concept_id}::{predicate_id}::{target_index}",
+                        "updated_at": source_updated_at,
+                        "match_type": "exact",
+                    },
+                    "access_granted": source_preview is not None
+                    or not include_concept_preview,
+                    "follow_up_actions": _build_follow_up_actions(
+                        [target_value],
+                        exclude={resolved_concept_id},
+                    ),
+                    "score": 1.0,
+                    "is_asserted": True,
+                    "relation_state": "asserted",
+                }
+                if include_concept_preview:
+                    hit["source_concept_preview"] = source_preview
+                hits.append(hit)
 
     if include_asserted_rows and include_structural and include_arg2_or_later:
         incoming_candidates: List[Tuple[str, Any, int, Any]] = []
@@ -483,55 +485,136 @@ def find_relations_with_argument(
                     continue
                 incoming_candidates.append(candidate)
         else:
-            incoming_asserted_binary_diagnostics.update(
-                {
-                    "path": "canonical_aggregation",
-                    "fallback_reason": "relationship_extent_index_unavailable",
-                }
+            exact_predicate_ids = _canonical_exact_predicate_filter_ids(
+                predicate_filter
             )
-            incoming_pipeline = [
-                {"$match": {"relationships": {"$type": "object"}}},
-                {
-                    "$project": {
-                        "concept_id": 1,
-                        "updated_at": 1,
-                        "relationship_items": {"$objectToArray": "$relationships"},
-                    }
-                },
-                {"$unwind": "$relationship_items"},
-                {
-                    "$project": {
-                        "concept_id": 1,
-                        "updated_at": 1,
-                        "predicate": "$relationship_items.k",
-                        "targets": "$relationship_items.v",
+            if exact_predicate_ids:
+                incoming_asserted_binary_diagnostics.update(
+                    {
+                        "path": "canonical_exact_predicate_query",
+                        "fallback_reason": (
+                            "relationship_extent_index_unavailable"
+                        ),
                     }
-                },
-                {"$match": {"targets": resolved_concept_id}},
-            ]
-            for row in ConceptsRepository.aggregate(incoming_pipeline):
-                incoming_asserted_binary_diagnostics[
-                    "canonical_rows_returned"
-                ] += 1
-                source_id = row.get("concept_id")
-                if not isinstance(source_id, str) or not source_id.strip():
-                    continue
-                source_id = source_id.strip()
-                targets = _normalise_relationship_targets(row.get("targets"))
-                if not targets:
-                    continue
-                for target_index, target_value in enumerate(targets):
-                    if target_value != resolved_concept_id:
+                )
+                exact_query = {
+                    "$or": [
+                        {
+                            f"relationships.{predicate_id}": (
+                                resolved_concept_id
+                            )
+                        }
+                        for predicate_id in exact_predicate_ids
+                    ]
+                }
+                exact_projection = {
+                    "_id": 0,
+                    "concept_id": 1,
+                    "updated_at": 1,
+                    **{
+                        f"relationships.{predicate_id}": 1
+                        for predicate_id in exact_predicate_ids
+                    },
+                }
+                exact_documents = ConceptsRepository.find(
+                    exact_query,
+                    exact_projection,
+                    max_time_ms=(
+                        _CANONICAL_EXACT_PREDICATE_FALLBACK_MAX_TIME_MS
+                    ),
+                )
+                for document in exact_documents:
+                    source_id = document.get("concept_id")
+                    if not isinstance(source_id, str) or not source_id.strip():
+                        continue
+                    source_id = source_id.strip()
+                    relationships = document.get("relationships")
+                    if not isinstance(relationships, Mapping):
                         continue
-                    incoming_candidates.append(
-                        (
-                            source_id,
-                            row.get("predicate"),
-                            target_index,
-                            row.get("updated_at"),
+                    for predicate_id in exact_predicate_ids:
+                        targets = _normalise_relationship_targets(
+                            relationships.get(predicate_id)
                         )
+                        if not targets:
+                            continue
+                        incoming_asserted_binary_diagnostics[
+                            "canonical_rows_returned"
+                        ] += 1
+                        for target_index, target_value in enumerate(targets):
+                            if target_value != resolved_concept_id:
+                                continue
+                            incoming_candidates.append(
+                                (
+                                    source_id,
+                                    predicate_id,
+                                    target_index,
+                                    document.get("updated_at"),
+                                )
+                            )
+            else:
+                incoming_asserted_binary_diagnostics.update(
+                    {
+                        "path": "canonical_aggregation",
+                        "fallback_reason": (
+                            "relationship_extent_index_unavailable"
+                        ),
+                    }
+                )
+                incoming_pipeline = [
+                    {"$match": {"relationships": {"$type": "object"}}},
+                    {
+                        "$project": {
+                            "concept_id": 1,
+                            "updated_at": 1,
+                            "relationship_items": {
+                                "$objectToArray": "$relationships"
+                            },
+                        }
+                    },
+                    {"$unwind": "$relationship_items"},
+                    {
+                        "$project": {
+                            "concept_id": 1,
+                            "updated_at": 1,
+                            "predicate": "$relationship_items.k",
+                            "targets": "$relationship_items.v",
+                        }
+                    },
+                    {"$match": {"targets": resolved_concept_id}},
+                ]
+                for row in ConceptsRepository.aggregate(incoming_pipeline):
+                    incoming_asserted_binary_diagnostics[
+                        "canonical_rows_returned"
+                    ] += 1
+                    source_id = row.get("concept_id")
+                    if not isinstance(source_id, str) or not source_id.strip():
+                        continue
+                    source_id = source_id.strip()
+                    targets = _normalise_relationship_targets(
+                        row.get("targets")
                     )
+                    if not targets:
+                        continue
+                    for target_index, target_value in enumerate(targets):
+                        if target_value != resolved_concept_id:
+                            continue
+                        incoming_candidates.append(
+                            (
+                                source_id,
+                                row.get("predicate"),
+                                target_index,
+                                row.get("updated_at"),
+                            )
+                        )
 
+        _prime_concept_preview_cache(
+            [
+                resolved_concept_id,
+                *(source_id for source_id, _, _, _ in incoming_candidates),
+            ],
+            include_preview=include_concept_preview,
+            preview_cache=preview_cache,
+        )
         for source_id, predicate_id, target_index, source_updated_at in (
             incoming_candidates
         ):
@@ -545,38 +628,39 @@ def find_relations_with_argument(
                 include_concept_preview,
                 preview_cache,
             )
-            hits.append(
-                {
-                    "source_concept_id": source_id,
-                    "predicate_concept_id": predicate_id,
-                    "relation_kind": "binary",
-                    "argument_indexes": matched_indexes,
-                    "target_value": resolved_concept_id,
-                    "target_concept_preview": (
-                        _resolve_concept_preview(
-                            resolved_concept_id,
-                            True,
-                            preview_cache,
-                        )
-                        if include_concept_preview
-                        else None
-                    ),
-                    "relation_metadata": {
-                        "relation_id": f"struct::{source_id}::{predicate_id}::incoming::{target_index}",
-                        "updated_at": _isoformat(source_updated_at),
-                        "match_type": "exact",
-                    },
-                    "access_granted": source_preview is not None
-                    or not include_concept_preview,
-                    "follow_up_actions": _build_follow_up_actions(
-                        [source_id],
-                        exclude={resolved_concept_id},
-                    ),
-                    "score": 1.0,
-                    "is_asserted": True,
-                    "relation_state": "asserted",
-                }
-            )
+            hit = {
+                "source_concept_id": source_id,
+                "predicate_concept_id": predicate_id,
+                "relation_kind": "binary",
+                "argument_indexes": matched_indexes,
+                "target_value": resolved_concept_id,
+                "target_concept_preview": (
+                    _resolve_concept_preview(
+                        resolved_concept_id,
+                        True,
+                        preview_cache,
+                    )
+                    if include_concept_preview
+                    else None
+                ),
+                "relation_metadata": {
+                    "relation_id": f"struct::{source_id}::{predicate_id}::incoming::{target_index}",
+                    "updated_at": _isoformat(source_updated_at),
+                    "match_type": "exact",
+                },
+                "access_granted": source_preview is not None
+                or not include_concept_preview,
+                "follow_up_actions": _build_follow_up_actions(
+                    [source_id],
+                    exclude={resolved_concept_id},
+                ),
+                "score": 1.0,
+                "is_asserted": True,
+                "relation_state": "asserted",
+            }
+            if include_concept_preview:
+                hit["source_concept_preview"] = source_preview
+            hits.append(hit)
 
     if include_asserted_rows and include_text and include_arg1:
         source_preview = _resolve_concept_preview(
@@ -614,6 +698,8 @@ def find_relations_with_argument(
                 "is_asserted": True,
                 "relation_state": "asserted",
             }
+            if include_concept_preview:
+                hit["source_concept_preview"] = source_preview
             snippet = _make_argument_match_snippet(
                 text=text_value,
                 needle=resolved_concept_id,
@@ -694,6 +780,7 @@ def find_relations_with_argument(
                     "relation_state": "asserted",
                 }
                 if include_concept_preview:
+                    hit["source_concept_preview"] = source_preview
                     hit["target_concept_preview"] = _resolve_concept_preview(
                         resolved_concept_id,
                         True,
@@ -2779,6 +2866,60 @@ def _resolve_concept_preview(
     return preview
 
 
+def _prime_concept_preview_cache(
+    concept_ids: Iterable[str],
+    *,
+    include_preview: bool,
+    preview_cache: Dict[str, Optional[Dict[str, Any]]],
+) -> None:
+    """Batch preview reads when more than one related concept is unresolved."""
+
+    if not include_preview:
+        return
+    pending_ids = list(
+        dict.fromkeys(
+            concept_id.strip()
+            for concept_id in concept_ids
+            if isinstance(concept_id, str)
+            and concept_id.strip()
+            and concept_id.strip() not in preview_cache
+        )
+    )
+    if len(pending_ids) < 2:
+        return
+
+    accessible_id_set = (
+        filter_accessible_concept_ids(pending_ids)
+        if should_enforce_access_control()
+        else set(pending_ids)
+    )
+    accessible_ids = [
+        concept_id
+        for concept_id in pending_ids
+        if concept_id in accessible_id_set
+    ]
+    for concept_id in pending_ids:
+        if concept_id not in accessible_id_set:
+            preview_cache[concept_id] = None
+
+    with bypass_access_control():
+        documents = list(
+            ConceptsRepository.find(
+                {"concept_id": {"$in": accessible_ids}},
+                _PREVIEW_DOC_PROJECTION,
+            )
+        )
+    for document in documents:
+        concept_id = document.get("concept_id")
+        if not isinstance(concept_id, str) or not concept_id.strip():
+            continue
+        preview_cache[concept_id.strip()] = _build_concept_preview_from_document(
+            document
+        )
+    for concept_id in accessible_ids:
+        preview_cache.setdefault(concept_id, None)
+
+
 def _load_accessible_preview_document(concept_id: str) -> Optional[Dict[str, Any]]:
     if should_enforce_access_control() and not can_access_concept(concept_id):
         return None
@@ -3264,6 +3405,8 @@ def _collect_uncertain_argument_hits_for_subject(
             "relation_state": "uncertain",
             "uncertainty": _coerce_uncertainty_entry(assertion),
         }
+        if include_concept_preview:
+            hit["source_concept_preview"] = source_preview
         if include_concept_preview and target_value.startswith("#V#"):
             hit["target_concept_preview"] = _resolve_concept_preview(
                 target_value,
@@ -3346,6 +3489,7 @@ def _collect_uncertain_argument_hits_for_targets(
                 "uncertainty": _coerce_uncertainty_entry(assertion),
             }
             if include_concept_preview:
+                hit["source_concept_preview"] = source_preview
                 hit["target_concept_preview"] = target_preview
             hits.append(hit)
     return hits
@@ -3393,6 +3537,23 @@ def _normalise_predicate_display_terms(
     return _iter_predicate_filter_tokens(predicate_filter)
 
 
+def _canonical_exact_predicate_filter_ids(
+    predicate_filter: Optional[Sequence[str]],
+) -> List[str]:
+    """Return exact safe field IDs only when the whole filter is canonical."""
+
+    tokens = _iter_predicate_filter_tokens(predicate_filter)
+    if not tokens or any(
+        not token.startswith("#V#")
+        or "." in token
+        or "$" in token
+        or "\x00" in token
+        for token in tokens
+    ):
+        return []
+    return list(dict.fromkeys(tokens))
+
+
 def _predicate_matches_terms(predicate_id: Any, terms: Sequence[str]) -> bool:
     if not terms:
         return True
diff --git a/src/backend/services/predicate_family_vontology_service.py b/src/backend/services/predicate_family_vontology_service.py
new file mode 100644
index 000000000..bcf55fac7
--- /dev/null
+++ b/src/backend/services/predicate_family_vontology_service.py
@@ -0,0 +1,247 @@
+"""Read represented relation-predicate family metadata from Vontology."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from ..db.repositories.concepts_repository import ConceptsRepository
+from .relationship_extent_index_service import query_relationship_extent_index
+
+INVERSE_LINK_PREDICATE_IDS = (
+    "#V#predicate_is_inverse_of_predicate",
+    "#V#inverse_predicates",
+)
+FAMILY_MEMBER_PREDICATE_ID = "#V#has_member_predicate"
+FOCAL_ROLE_PREDICATE_ID = "#V#has_focal_role_predicate"
+RELATED_ROLE_PREDICATE_ID = "#V#has_related_role_predicate"
+REIFIED_RELATION_TYPE_PREDICATE_ID = "#V#has_reified_relation_type"
+_MAX_EXPANDED_PREDICATES = 100
+_CANONICAL_FAMILY_FALLBACK_MAX_TIME_MS = 8_000
+
+
+def _canonical_ids(value: Any) -> list[str]:
+    values = (
+        value
+        if isinstance(value, Sequence)
+        and not isinstance(value, (str, bytes, bytearray))
+        else [value]
+    )
+    result: list[str] = []
+    for item in values:
+        if not isinstance(item, str):
+            continue
+        concept_id = item.strip()
+        if not concept_id.startswith("#V#") or concept_id in result:
+            continue
+        result.append(concept_id)
+    return result
+
+
+def _append_unique(target: list[str], values: Any, *, limit: int) -> None:
+    for concept_id in _canonical_ids(values):
+        if concept_id not in target:
+            target.append(concept_id)
+        if len(target) >= limit:
+            return
+
+
+def _relationships(document: Mapping[str, Any]) -> Mapping[str, Any]:
+    relationships = document.get("relationships")
+    return relationships if isinstance(relationships, Mapping) else {}
+
+
+def _empty_expansion(seeds: list[str]) -> dict[str, Any]:
+    return {
+        "schema_version": "relation_predicate_family_expansion.v1",
+        "seed_predicate_ids": seeds,
+        "predicate_ids": list(seeds),
+        "family_concept_ids": [],
+        "focal_role_predicate_ids": [],
+        "related_role_predicate_ids": [],
+        "reified_relation_type_ids": [],
+        "source": "vontology_predicate_schema",
+    }
+
+
+def expand_relation_predicate_family(
+    predicate_ids: Sequence[str],
+    *,
+    max_predicates: int = 32,
+) -> dict[str, Any]:
+    """Expand predicate IDs through represented inverse and family metadata.
+
+    Predicate documents are resolved by their exact indexed IDs. Family
+    membership is discovered through the target-first relationship extent
+    index, avoiding a scan across dynamic ``relationships`` keys. If that
+    derived index is unavailable, the supplied predicate IDs and any directly
+    represented inverse links remain usable.
+    """
+
+    try:
+        requested_limit = int(max_predicates or 32)
+    except (TypeError, ValueError):
+        requested_limit = 32
+    resolved_limit = max(1, min(requested_limit, _MAX_EXPANDED_PREDICATES))
+    seeds = _canonical_ids(predicate_ids)[:resolved_limit]
+    result = _empty_expansion(seeds)
+    if not seeds:
+        result["status"] = "unchanged"
+        return result
+
+    expanded = list(seeds)
+    family_ids: list[str] = []
+    focal_role_predicate_ids: list[str] = []
+    related_role_predicate_ids: list[str] = []
+    reified_relation_type_ids: list[str] = []
+    index_status = "available"
+    family_discovery_source = "relationship_extent_index"
+
+    try:
+        predicate_projection: dict[str, int] = {
+            "_id": 0,
+            "concept_id": 1,
+        }
+        for inverse_predicate_id in INVERSE_LINK_PREDICATE_IDS:
+            predicate_projection[f"relationships.{inverse_predicate_id}"] = 1
+        predicate_documents = ConceptsRepository.find(
+            {"concept_id": {"$in": list(seeds)}},
+            projection=predicate_projection,
+        )
+        for document in predicate_documents:
+            relationships = _relationships(document)
+            for inverse_predicate_id in INVERSE_LINK_PREDICATE_IDS:
+                _append_unique(
+                    expanded,
+                    relationships.get(inverse_predicate_id),
+                    limit=resolved_limit,
+                )
+
+        family_rows, family_row_count = query_relationship_extent_index(
+            predicate_id=FAMILY_MEMBER_PREDICATE_ID,
+            target_values=expanded,
+            count_total=False,
+            projection={
+                "_id": 0,
+                "source_concept_id": 1,
+                "predicate_id": 1,
+                "target_value": 1,
+            },
+            limit=resolved_limit * 4,
+        )
+        if family_row_count < 0:
+            index_status = "unavailable"
+            family_discovery_source = "canonical_exact_predicate_query"
+            family_projection = {
+                "_id": 0,
+                "concept_id": 1,
+                f"relationships.{FAMILY_MEMBER_PREDICATE_ID}": 1,
+                f"relationships.{FOCAL_ROLE_PREDICATE_ID}": 1,
+                f"relationships.{RELATED_ROLE_PREDICATE_ID}": 1,
+                f"relationships.{REIFIED_RELATION_TYPE_PREDICATE_ID}": 1,
+            }
+            family_documents = list(
+                ConceptsRepository.find(
+                    {
+                        f"relationships.{FAMILY_MEMBER_PREDICATE_ID}": {
+                            "$in": list(expanded)
+                        }
+                    },
+                    projection=family_projection,
+                    limit=resolved_limit,
+                    max_time_ms=_CANONICAL_FAMILY_FALLBACK_MAX_TIME_MS,
+                )
+            )
+            for document in family_documents:
+                family_id = document.get("concept_id")
+                if isinstance(family_id, str):
+                    _append_unique(
+                        family_ids,
+                        [family_id],
+                        limit=resolved_limit,
+                    )
+        else:
+            family_documents = []
+            for row in family_rows:
+                family_id = row.get("source_concept_id")
+                if isinstance(family_id, str):
+                    _append_unique(family_ids, [family_id], limit=resolved_limit)
+
+        if family_ids:
+            if not family_documents:
+                family_projection = {
+                    "_id": 0,
+                    "concept_id": 1,
+                    f"relationships.{FAMILY_MEMBER_PREDICATE_ID}": 1,
+                    f"relationships.{FOCAL_ROLE_PREDICATE_ID}": 1,
+                    f"relationships.{RELATED_ROLE_PREDICATE_ID}": 1,
+                    f"relationships.{REIFIED_RELATION_TYPE_PREDICATE_ID}": 1,
+                }
+                family_documents = ConceptsRepository.find(
+                    {"concept_id": {"$in": family_ids}},
+                    projection=family_projection,
+                )
+            accessible_family_ids: list[str] = []
+            for document in family_documents:
+                family_id = document.get("concept_id")
+                if isinstance(family_id, str):
+                    _append_unique(
+                        accessible_family_ids,
+                        [family_id],
+                        limit=resolved_limit,
+                    )
+                relationships = _relationships(document)
+                _append_unique(
+                    expanded,
+                    relationships.get(FAMILY_MEMBER_PREDICATE_ID),
+                    limit=resolved_limit,
+                )
+                _append_unique(
+                    focal_role_predicate_ids,
+                    relationships.get(FOCAL_ROLE_PREDICATE_ID),
+                    limit=resolved_limit,
+                )
+                _append_unique(
+                    related_role_predicate_ids,
+                    relationships.get(RELATED_ROLE_PREDICATE_ID),
+                    limit=resolved_limit,
+                )
+                _append_unique(
+                    reified_relation_type_ids,
+                    relationships.get(REIFIED_RELATION_TYPE_PREDICATE_ID),
+                    limit=resolved_limit,
+                )
+            family_ids = accessible_family_ids
+    except Exception as exc:
+        result.update(
+            {
+                "status": "schema_read_failed",
+                "error_class": type(exc).__name__,
+                "relationship_extent_index_status": index_status,
+            }
+        )
+        return result
+
+    result.update(
+        {
+            "predicate_ids": expanded[:resolved_limit],
+            "family_concept_ids": family_ids,
+            "focal_role_predicate_ids": focal_role_predicate_ids,
+            "related_role_predicate_ids": related_role_predicate_ids,
+            "reified_relation_type_ids": reified_relation_type_ids,
+            "status": "expanded" if expanded != seeds else "unchanged",
+            "relationship_extent_index_status": index_status,
+            "family_discovery_source": family_discovery_source,
+        }
+    )
+    return result
+
+
+__all__ = [
+    "FAMILY_MEMBER_PREDICATE_ID",
+    "FOCAL_ROLE_PREDICATE_ID",
+    "INVERSE_LINK_PREDICATE_IDS",
+    "RELATED_ROLE_PREDICATE_ID",
+    "REIFIED_RELATION_TYPE_PREDICATE_ID",
+    "expand_relation_predicate_family",
+]
diff --git a/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_prompt_seed.md b/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_prompt_seed.md
index fea9c737c..abf014700 100644
--- a/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_prompt_seed.md
+++ b/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_prompt_seed.md
@@ -25,23 +25,19 @@ Tool guidance:
   `get_predicate_incidence` using `concept_id` set to the exact entity concept
   ID (for example `#V#michael_witbrock`), plus
   `include_argument_type_counts: true`, and
-  `relation_kind: "binary"` unless the user explicitly asks for incoming or
-  text relations. If subject-side results are weak or empty for authorship-style
-  turns, repeat with `argument_index: "object"` before concluding no paper-like
-  evidence.
+  `argument_index: "any"` and `relation_kind: "binary"` unless the user
+  explicitly asks for text relations. Inspect both incoming and outgoing
+  represented predicates before concluding that matching evidence is absent.
 - For both `get_predicate_incidence` and `find_relations_with_argument`, the
   anchor entity always goes in `concept_id`. `argument_index` names
   the relation slot to inspect; `subject` is outbound and `object` is inbound.
   It is not a top-level payload field. Do not use
   payload keys named `subject` or `object` for these tools.
-- For relation hits from `find_relations_with_argument`, interpret the related
-  concept by argument direction:
-  - if `argument_index` is `subject`, the related concept is in
-    `target_concept_id`;
-  - if `argument_index` is `object`, the related concept is in
-    `source_concept_id`.
-  When producing a user-facing list, emit that related concept as `#V#...`
-  (the concept ID), not just the label.
+- For relation hits from `find_relations_with_argument`, use
+  `related_concept_id`, `related_name`, `related_type_ids`, and
+  `direction_from_focal_entity`. Those fields resolve the related entity
+  consistently for both incoming and outgoing assertions. When producing a
+  user-facing list, emit `related_concept_id` as `#V#...`, not just the label.
 - Preserve the full `#V#...` concept ID exactly. Do not strip the `#V#`
   prefix or rewrite the ID into a display name.
 - Use `get_predicate_incidence` to inspect which predicates are actually used
@@ -55,7 +51,10 @@ Tool guidance:
   `find_relations_with_argument` with that same entity in `concept_id`,
   `relation_kind: "binary"`, and a narrow
   `predicate_filter` to retrieve grounded relation hits for the chosen
-  predicates. Set `argument_index` to the direction you are testing.
+  predicates. Use `argument_index: "any"` unless the request explicitly
+  restricts direction. The workflow may expand the narrowed predicate through
+  represented inverse and relation-family metadata; use every predicate in
+  that bounded expansion.
 - `predicate_filter`, `argument_index`, `relation_kind`, and relation paging
   limits are arguments for `find_relations_with_argument`, not
   `fetch_concept`. Never send those relation-query arguments to
diff --git a/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_workflow_seed_bundle.json b/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_workflow_seed_bundle.json
index 9465aed78..47928937d 100644
--- a/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_workflow_seed_bundle.json
+++ b/src/backend/workflows/repo_seed_bundles/entity_information_retrieval_workflow_seed_bundle.json
@@ -2,7 +2,7 @@
   "family_id": "entity_information_retrieval_workflow_seed_bundle",
   "managed_by": "entity_information_retrieval_workflow_vontology_service",
   "schema_version": "repo_seed_workflow_bundle.v1",
-  "seed_version": "11",
+  "seed_version": "12",
   "source_tag": "JVNAUTOSCI-2075",
   "supported_action_ids": [
     "llm.action"
@@ -84,7 +84,7 @@
               "max_tool_invocations": 5,
               "tool_argument_defaults": {
                 "get_predicate_incidence": {
-                  "argument_index": "subject",
+                  "argument_index": "any",
                   "relation_kind": "binary",
                   "include_concept_preview": false,
                   "include_argument_type_counts": true,
@@ -94,9 +94,11 @@
                   "max_relation_ids_per_group": 25
                 },
                 "find_relations_with_argument": {
-                  "argument_index": "subject",
+                  "argument_index": "any",
                   "relation_kind": "binary",
                   "limit": 20,
+                  "__expand_predicate_family_from_vontology": true,
+                  "__predicate_family_max_predicates": 12,
                   "__derive_predicate_filter_from_recent_incidence": {
                     "enabled": true,
                     "max_predicates": 2,
@@ -132,7 +134,7 @@
                 }
               },
               "context_messages_context_key": "augmented_context",
-              "response_contract_text": "Answer the user directly from grounded represented evidence. For broad self-profile, represented-self, authorship, ownership, affiliation, role, and other entity-relative relation questions, first fetch the focal entity concept, inspect predicate incidence with argument type counts, retrieve narrowed relation evidence, and check uncertainty-bearing relationship assertions for the same focal entity. When the user asks to separate established facts from likely inferences, report asserted relation/focal-concept evidence separately from uncertain or proposed relationship assertions; do not merge those categories. Once the five required evidence tools have succeeded and the relation result contains the requested related entity IDs, names, and applicable type evidence, answer immediately from those results. Do not fetch or summarise predicate concepts merely to restate a predicate label or description already present in the incidence and relation evidence. If a required evidence call fails or lacks a necessary field, preserve that typed failure so the outer represented recovery policy can choose a retry, alternate extent, or narrower answer instead of spending successful-path calls on speculative re-reading. For every predicate-filtered entity-relative request, once incidence identifies a matching predicate, the content-bearing read must use find_relations_with_argument with that exact predicate in predicate_filter; an unfiltered relation page is not evidence for a matching row that falls outside the returned page. predicate_filter, argument_index, relation_kind, and relation paging limits belong to find_relations_with_argument, not fetch_concept. Never send relation-query arguments to fetch_concept, and never answer a relation-extent request from such a fetch because it does not perform the required filtered relation read. If an earlier relation read was unfiltered or omitted the matching row, call find_relations_with_argument again with the narrowed predicate_filter before answering. For papers, inspect predicate incidence before relation paging, then filter the resulting extent by the requested type before answering. If predicate incidence for a paper request exposes #V#author_of or #V#owner_of with paper-like type counts or sample groundings, the next relation lookup must use find_relations_with_argument with a predicate_filter containing those exact predicate IDs; do not use an unfiltered relation page first. If predicate incidence exposes role-expansion summaries for reified/event/claim neighbours, use role-filler type counts to choose the follow-up extent and do not treat the reified neighbour itself as the answer entity. Do not claim that no matching results exist unless the relation-bearing and uncertainty-bearing evidence in this turn supports that negative conclusion. If grounded evidence is insufficient, say that clearly instead of guessing. When using relation hits, map the returned related concept by argument direction: if the call used `argument_index` `subject`, use `target_concept_id`; if `object`, use `source_concept_id`. Emit every relation-grounded result as a Markdown bullet with a resolved `#V#...` concept ID wherever that relation-hit concept ID is available. Never emit relation-grounded results as bare assertion syntax such as `subject --predicate--> object`. If a related concept ID is missing, do not fabricate one: omit that item from the list and state clearly that no usable `#V#` ID was returned. Use a single-item bullet if only one grounded entity is returned. Relation-grounded outputs must be bullet entries with the resolved `#V#` ID token for navigation.",
+              "response_contract_text": "Answer the user directly from grounded represented evidence. For broad self-profile, represented-self, authorship, ownership, affiliation, role, and other entity-relative relation questions, first fetch the focal entity concept, inspect predicate incidence with argument type counts in every relation direction, retrieve narrowed relation evidence, and check uncertainty-bearing relationship assertions for the same focal entity. When the user asks to separate established facts from likely inferences, report asserted relation/focal-concept evidence separately from uncertain or proposed relationship assertions; do not merge those categories. Once the five required evidence tools have succeeded and the relation result contains the requested related entity IDs, names, and applicable type evidence, answer immediately from those results. When Vontology expands an observed predicate through represented inverse or relation-family metadata, use every returned family predicate rather than reporting only the initially observed predicate. Do not fetch or summarise predicate concepts merely to restate a predicate label or description already present in the incidence and relation evidence. If a required evidence call fails or lacks a necessary field, preserve that typed failure so the outer represented recovery policy can choose a retry, alternate extent, or narrower answer instead of spending successful-path calls on speculative re-reading. For every predicate-filtered entity-relative request, once incidence or represented predicate-family expansion identifies matching predicates, the content-bearing read must use find_relations_with_argument with those exact predicates in predicate_filter; an unfiltered relation page is not evidence for a matching row that falls outside the returned page. predicate_filter, argument_index, relation_kind, and relation paging limits belong to find_relations_with_argument, not fetch_concept. Never send relation-query arguments to fetch_concept, and never answer a relation-extent request from such a fetch because it does not perform the required filtered relation read. If an earlier relation read was unfiltered or omitted the matching row, call find_relations_with_argument again with the narrowed predicate_filter before answering. For papers, inspect predicate incidence before relation paging, then filter the resulting extent by the requested type before answering. If predicate incidence for a paper request exposes #V#author_of or #V#owner_of with paper-like type counts or sample groundings, the next relation lookup must use find_relations_with_argument with a predicate_filter containing those exact predicate IDs; do not use an unfiltered relation page first. If predicate incidence exposes role-expansion summaries for reified/event/claim neighbours, use role-filler type counts to choose the follow-up extent and do not treat the reified neighbour itself as the answer entity. Do not claim that no matching results exist unless the relation-bearing and uncertainty-bearing evidence in this turn supports that negative conclusion. If grounded evidence is insufficient, say that clearly instead of guessing. When using relation hits, use each hit's related_concept_id, related_name, related_type_ids, and direction_from_focal_entity fields; these fields resolve whether the related entity is the source or target, and duplicate relations for the same neighbour are compacted. Include every unique related entity whose related_type_ids establish the requested result class, regardless of whether the relation predicate is outgoing or incoming. Emit every relation-grounded result as a Markdown bullet with a resolved `#V#...` concept ID wherever that relation-hit concept ID is available. Never emit relation-grounded results as bare assertion syntax such as `subject --predicate--> object`. If a related concept ID is missing, do not fabricate one: omit that item from the list and state clearly that no usable `#V#` ID was returned. Use a single-item bullet if only one grounded entity is returned. Relation-grounded outputs must be bullet entries with the resolved `#V#` ID token for navigation.",
               "context_fields": [
                 {
                   "context_key": "prompt",
diff --git a/tests/backend/test_concept_relation_service_extent_index.py b/tests/backend/test_concept_relation_service_extent_index.py
index cdc3cb377..f320cc160 100644
--- a/tests/backend/test_concept_relation_service_extent_index.py
+++ b/tests/backend/test_concept_relation_service_extent_index.py
@@ -203,6 +203,158 @@ def fake_filter_accessible_concept_ids(concept_ids):
     }
 
 
+def test_relation_hits_include_source_previews_for_both_directions(
+    monkeypatch,
+) -> None:
+    from src.backend.services import concept_relation_service as service
+
+    focal_id = "#V#focal"
+    monkeypatch.setattr(
+        service,
+        "_load_accessible_relation_subject_document",
+        lambda *_args, **_kwargs: {
+            "concept_id": focal_id,
+            "relationships": {"#V#direct_relation": ["#V#outgoing_related"]},
+        },
+    )
+    monkeypatch.setattr(
+        service,
+        "query_relationship_extent_index",
+        lambda **_kwargs: (
+            [
+                {
+                    "source_concept_id": "#V#incoming_related",
+                    "predicate_id": "#V#inverse_relation",
+                    "target_value": focal_id,
+                    "target_index": 0,
+                    "source_updated_at": None,
+                }
+            ],
+            1,
+        ),
+    )
+    monkeypatch.setattr(
+        service,
+        "filter_accessible_concept_ids",
+        lambda concept_ids: set(concept_ids),
+    )
+    preview_docs = {
+        focal_id: {"concept_id": focal_id, "name": "Focal", "relationships": {}},
+        "#V#outgoing_related": {
+            "concept_id": "#V#outgoing_related",
+            "name": "Outgoing Related",
+            "relationships": {"is_an_instance_of": ["#V#requested_type"]},
+        },
+        "#V#incoming_related": {
+            "concept_id": "#V#incoming_related",
+            "name": "Incoming Related",
+            "relationships": {"is_an_instance_of": ["#V#requested_type"]},
+        },
+    }
+    monkeypatch.setattr(
+        service,
+        "_load_accessible_preview_document",
+        lambda concept_id: preview_docs.get(concept_id),
+    )
+
+    payload = service.find_relations_with_argument(
+        focal_id,
+        argument_index="any",
+        predicate_filter=["#V#direct_relation", "#V#inverse_relation"],
+        relation_kind="binary",
+        include_concept_preview=True,
+        limit=10,
+    )
+
+    hits_by_source = {hit["source_concept_id"]: hit for hit in payload["hits"]}
+    assert hits_by_source[focal_id]["source_concept_preview"]["name"] == "Focal"
+    assert (
+        hits_by_source["#V#incoming_related"]["source_concept_preview"]["name"]
+        == "Incoming Related"
+    )
+    assert hits_by_source["#V#incoming_related"]["source_concept_preview"][
+        "type_ids"
+    ] == ["#V#requested_type"]
+
+
+def test_incoming_relation_previews_batch_uncached_sources(monkeypatch) -> None:
+    from src.backend.services import concept_relation_service as service
+
+    target_id = "#V#target"
+    preview_queries: list[tuple[dict[str, Any], dict[str, int]]] = []
+    monkeypatch.setattr(
+        service,
+        "_load_accessible_relation_subject_document",
+        lambda *_args, **_kwargs: None,
+    )
+    monkeypatch.setattr(
+        service,
+        "query_relationship_extent_index",
+        lambda **_kwargs: (
+            [
+                {
+                    "source_concept_id": source_id,
+                    "predicate_id": "#V#relation",
+                    "target_value": target_id,
+                    "target_index": 0,
+                    "source_updated_at": None,
+                }
+                for source_id in ("#V#source_a", "#V#source_b")
+            ],
+            2,
+        ),
+    )
+    monkeypatch.setattr(
+        service,
+        "filter_accessible_concept_ids",
+        lambda concept_ids: set(concept_ids),
+    )
+    monkeypatch.setattr(
+        service,
+        "should_enforce_access_control",
+        lambda: True,
+    )
+
+    def fake_find(query, projection):
+        preview_queries.append((query, projection))
+        return [
+            {
+                "concept_id": source_id,
+                "name": source_id.removeprefix("#V#"),
+                "relationships": {},
+            }
+            for source_id in (target_id, "#V#source_a", "#V#source_b")
+        ]
+
+    monkeypatch.setattr(
+        service.ConceptsRepository, "find", staticmethod(fake_find)
+    )
+    monkeypatch.setattr(
+        service,
+        "_load_accessible_preview_document",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("batched source previews must avoid point reads")
+        ),
+    )
+
+    payload = service.find_relations_with_argument(
+        target_id,
+        relation_kind="binary",
+        include_concept_preview=True,
+    )
+
+    assert len(preview_queries) == 1
+    assert preview_queries[0][0] == {
+        "concept_id": {
+            "$in": [target_id, "#V#source_a", "#V#source_b"]
+        }
+    }
+    assert {
+        hit["source_concept_preview"]["concept_id"]
+        for hit in payload["hits"]
+    } == {"#V#source_a", "#V#source_b"}
+
+
 def test_incoming_asserted_binary_falls_back_only_when_index_is_unavailable(
     monkeypatch,
 ) -> None:
@@ -210,15 +362,18 @@ def test_incoming_asserted_binary_falls_back_only_when_index_is_unavailable(
 
     target_id = "#V#target"
     updated_at = datetime(2026, 7, 26, 5, 6, tzinfo=UTC)
-    aggregate_pipelines: list[list[dict[str, Any]]] = []
+    exact_queries: list[
+        tuple[dict[str, Any], dict[str, Any], int | None]
+    ] = []
 
-    def fake_aggregate(pipeline):
-        aggregate_pipelines.append(pipeline)
+    def fake_find(query, projection, *, max_time_ms=None):
+        exact_queries.append((query, projection, max_time_ms))
         return [
             {
                 "concept_id": "#V#source",
-                "predicate": "#V#supervises",
-                "targets": ["#V#other", target_id],
+                "relationships": {
+                    "#V#supervises": ["#V#other", target_id],
+                },
                 "updated_at": updated_at,
             }
         ]
@@ -240,10 +395,13 @@ def fake_aggregate(pipeline):
             AssertionError("fallback rows are already access-controlled")
         ),
     )
+    monkeypatch.setattr(
+        service.ConceptsRepository, "find", staticmethod(fake_find)
+    )
     monkeypatch.setattr(
         service.ConceptsRepository,
         "aggregate",
-        staticmethod(fake_aggregate),
+        staticmethod(_fail_if_aggregated),
     )
 
     payload = service.find_relations_with_argument(
@@ -254,7 +412,18 @@ def fake_aggregate(pipeline):
         include_concept_preview=False,
     )
 
-    assert len(aggregate_pipelines) == 1
+    assert exact_queries == [
+        (
+            {"$or": [{"relationships.#V#supervises": target_id}]},
+            {
+                "_id": 0,
+                "concept_id": 1,
+                "updated_at": 1,
+                "relationships.#V#supervises": 1,
+            },
+            8_000,
+        )
+    ]
     assert payload["total_hits"] == 1
     hit = payload["hits"][0]
     assert hit["source_concept_id"] == "#V#source"
@@ -268,7 +437,7 @@ def fake_aggregate(pipeline):
     }
     assert payload["relation_query_diagnostics"]["incoming_asserted_binary"] == {
         "requested": True,
-        "path": "canonical_aggregation",
+        "path": "canonical_exact_predicate_query",
         "used_relationship_extent_index": False,
         "fallback_reason": "relationship_extent_index_unavailable",
         "index_rows_examined": 0,
@@ -277,6 +446,53 @@ def fake_aggregate(pipeline):
     }
 
 
+def test_unresolved_predicate_label_preserves_general_canonical_fallback(
+    monkeypatch,
+) -> None:
+    from src.backend.services import concept_relation_service as service
+
+    aggregate_pipelines: list[list[dict[str, Any]]] = []
+
+    monkeypatch.setattr(
+        service,
+        "_load_accessible_relation_subject_document",
+        lambda *_args, **_kwargs: None,
+    )
+    monkeypatch.setattr(
+        service,
+        "query_relationship_extent_index",
+        lambda **_kwargs: ([], -1),
+    )
+    monkeypatch.setattr(
+        service.ConceptsRepository,
+        "find",
+        staticmethod(_fail_if_aggregated),
+    )
+    monkeypatch.setattr(
+        service.ConceptsRepository,
+        "aggregate",
+        staticmethod(
+            lambda pipeline: aggregate_pipelines.append(pipeline) or []
+        ),
+    )
+
+    payload = service.find_relations_with_argument(
+        "#V#target",
+        predicate_filter=["supervises"],
+        relation_kind="binary",
+        include_concept_preview=False,
+    )
+
+    assert len(aggregate_pipelines) == 1
+    diagnostics = payload["relation_query_diagnostics"][
+        "incoming_asserted_binary"
+    ]
+    assert diagnostics["path"] == "canonical_aggregation"
+    assert diagnostics["fallback_reason"] == (
+        "relationship_extent_index_unavailable"
+    )
+
+
 def test_empty_available_extent_index_does_not_trigger_canonical_fallback(
     monkeypatch,
 ) -> None:
diff --git a/tests/backend/test_concept_relation_service_uncertainty.py b/tests/backend/test_concept_relation_service_uncertainty.py
index 55658c27a..20da4ab4b 100644
--- a/tests/backend/test_concept_relation_service_uncertainty.py
+++ b/tests/backend/test_concept_relation_service_uncertainty.py
@@ -363,6 +363,7 @@ def test_subject_relation_predicate_filter_precedes_target_access_checks(
         },
     }
     checked_concept_ids: list[str] = []
+    checked_target_batches: list[list[str]] = []
 
     monkeypatch.setattr(service, "should_enforce_access_control", lambda: True)
 
@@ -371,6 +372,14 @@ def fake_can_access_concept(concept_id: str) -> bool:
         return True
 
     monkeypatch.setattr(service, "can_access_concept", fake_can_access_concept)
+    monkeypatch.setattr(
+        service,
+        "filter_accessible_concept_ids",
+        lambda concept_ids: (
+            checked_target_batches.append(list(concept_ids))
+            or {"#V#gmail_profile"}
+        ),
+    )
     monkeypatch.setattr(
         service.ConceptsRepository,
         "find_one",
@@ -386,7 +395,8 @@ def fake_can_access_concept(concept_id: str) -> bool:
     assert loaded["relationships"] == {
         "#V#has_default_mail_profile": ["#V#gmail_profile"]
     }
-    assert checked_concept_ids == ["#V#alice", "#V#gmail_profile"]
+    assert checked_concept_ids == ["#V#alice"]
+    assert checked_target_batches == [["#V#gmail_profile"]]
 
 
 def test_get_predicate_incidence_counts_other_argument_types() -> None:
diff --git a/tests/backend/test_entity_information_retrieval_workflow_vontology_service.py b/tests/backend/test_entity_information_retrieval_workflow_vontology_service.py
index e54404dd2..c18dd879d 100644
--- a/tests/backend/test_entity_information_retrieval_workflow_vontology_service.py
+++ b/tests/backend/test_entity_information_retrieval_workflow_vontology_service.py
@@ -172,21 +172,25 @@ def test_bootstrap_materialises_entity_information_retrieval_workflow(
         response_contract
     )
     assert "Never send relation-query arguments to fetch_concept" in response_contract
+    assert "represented inverse or relation-family metadata" in response_contract
+    assert "related_concept_id, related_name, related_type_ids" in response_contract
     assert llm_policy.get("tool_argument_defaults") == {
-            "get_predicate_incidence": {
-                "argument_index": "subject",
-                "relation_kind": "binary",
-                "include_argument_type_counts": True,
-                "include_concept_preview": False,
-                "limit": 32,
-            },
+        "get_predicate_incidence": {
+            "argument_index": "any",
+            "relation_kind": "binary",
+            "include_argument_type_counts": True,
+            "include_concept_preview": False,
+            "limit": 32,
+        },
         "get_text_relations_summary": {
             "max_relation_ids_per_group": 25,
         },
         "find_relations_with_argument": {
-            "argument_index": "subject",
+            "argument_index": "any",
             "relation_kind": "binary",
             "limit": 20,
+            "__expand_predicate_family_from_vontology": True,
+            "__predicate_family_max_predicates": 12,
             "__derive_predicate_filter_from_recent_incidence": {
                 "enabled": True,
                 "max_predicates": 2,
@@ -287,6 +291,9 @@ def test_entity_information_retrieval_prompt_support_seeds_content_from_repo_ass
     assert "first ontology-native" in prompt_text_lower
     assert "anchor entity always goes in `concept_id`" in prompt_text_lower
     assert "payload keys named `subject` or `object`" in prompt_text_lower
+    assert '`argument_index: "any"`' in prompt_text_lower
+    assert "represented inverse and relation-family metadata" in prompt_text_lower
+    assert "`related_concept_id`" in prompt_text_lower
     assert "preserve the full `#v#" in prompt_text_lower
     assert "must include a `predicate_filter`" in prompt_text_lower
     assert "before any broader relation paging" in prompt_text_lower
diff --git a/tests/backend/test_internal_mcp_orchestrator_context_limits.py b/tests/backend/test_internal_mcp_orchestrator_context_limits.py
index 74befe52e..1119b0847 100644
--- a/tests/backend/test_internal_mcp_orchestrator_context_limits.py
+++ b/tests/backend/test_internal_mcp_orchestrator_context_limits.py
@@ -655,6 +655,87 @@ def test_format_tool_result_shapes_find_relations_payload_with_target_type_ids()
     ]
 
 
+def test_format_tool_result_projects_unique_related_entities_in_both_directions():
+    orchestrator = InternalMCPChatOrchestrator(
+        gateway=cast(Any, _StubGateway()),
+        max_tool_invocations=1,
+        max_tool_result_chars=5_000,
+        max_tool_result_field_chars=1_500,
+        max_context_chars=80_000,
+    )
+
+    encoded = orchestrator._format_tool_result(
+        "find_relations_with_argument",
+        {
+            "concept_id": "#V#focal_entity",
+            "total_hits": 3,
+            "hits": [
+                {
+                    "source_concept_id": "#V#focal_entity",
+                    "predicate_concept_id": "#V#outgoing_relation",
+                    "relation_kind": "binary",
+                    "argument_indexes": [1],
+                    "target_value": "#V#related_one",
+                    "target_concept_preview": {
+                        "concept_id": "#V#related_one",
+                        "name": "Related One",
+                        "type_ids": ["#V#requested_type"],
+                    },
+                },
+                {
+                    "source_concept_id": "#V#related_two",
+                    "predicate_concept_id": "#V#incoming_relation",
+                    "relation_kind": "binary",
+                    "argument_indexes": [2],
+                    "target_value": "#V#focal_entity",
+                    "source_concept_preview": {
+                        "concept_id": "#V#related_two",
+                        "name": "Related Two",
+                        "type_ids": ["#V#requested_type"],
+                    },
+                    "target_concept_preview": {
+                        "concept_id": "#V#focal_entity",
+                        "name": "Focal Entity",
+                    },
+                },
+                {
+                    "source_concept_id": "#V#focal_entity",
+                    "predicate_concept_id": "#V#duplicate_relation",
+                    "relation_kind": "binary",
+                    "argument_indexes": [1],
+                    "target_value": "#V#related_one",
+                    "target_concept_preview": {
+                        "concept_id": "#V#related_one",
+                        "name": "Related One",
+                        "type_ids": ["#V#requested_type"],
+                    },
+                },
+            ],
+        },
+        1.0,
+        "ok",
+    )
+
+    payload = json.loads(encoded)["payload"]
+    assert payload["shown_hit_count"] == 2
+    assert [hit["related_concept_id"] for hit in payload["hits"]] == [
+        "#V#related_one",
+        "#V#related_two",
+    ]
+    assert payload["hits"][0]["direction_from_focal_entity"] == "outgoing"
+    assert payload["hits"][0]["related_name"] == "Related One"
+    assert payload["hits"][0]["related_type_ids"] == ["#V#requested_type"]
+    assert payload["hits"][0]["related_predicate_concept_ids"] == [
+        "#V#outgoing_relation",
+        "#V#duplicate_relation",
+    ]
+    assert payload["hits"][0]["directions_from_focal_entity"] == ["outgoing"]
+    assert payload["hits"][1]["direction_from_focal_entity"] == "incoming"
+    assert payload["hits"][1]["related_name"] == "Related Two"
+    assert payload["hits"][1]["related_type_ids"] == ["#V#requested_type"]
+    assert payload["compacted_duplicate_hit_count"] == 1
+
+
 def test_turn_scoped_tool_payload_support_applies_workflow_tool_argument_defaults():
     orchestrator = InternalMCPChatOrchestrator(
         gateway=cast(Any, _StubGateway()),
@@ -704,6 +785,79 @@ def test_turn_scoped_tool_payload_support_applies_workflow_tool_argument_default
     assert payload_with_explicit_values["limit"] == 3
 
 
+def test_turn_scoped_tool_payload_support_expands_represented_predicate_family(
+    monkeypatch,
+):
+    from src.backend.services import predicate_family_vontology_service as service
+
+    orchestrator = InternalMCPChatOrchestrator(
+        gateway=cast(Any, _StubGateway()),
+        max_tool_invocations=1,
+        max_context_chars=80_000,
+    )
+    monkeypatch.setattr(
+        service,
+        "expand_relation_predicate_family",
+        lambda predicate_ids, *, max_predicates: {
+            "schema_version": "relation_predicate_family_expansion.v1",
+            "seed_predicate_ids": list(predicate_ids),
+            "predicate_ids": [
+                "#V#direct_relation",
+                "#V#inverse_relation",
+                "#V#role_relation",
+            ],
+            "family_concept_ids": ["#V#represented_relation_family"],
+            "status": "expanded",
+            "relationship_extent_index_status": "available",
+            "family_discovery_source": "relationship_extent_index",
+        },
+    )
+    payload = {
+        "concept_id": "#V#focal_entity",
+        "predicate_filter": ["#V#direct_relation"],
+    }
+    data: dict[str, Any] = {
+        "tool_argument_defaults": {
+            "find_relations_with_argument": {
+                "__expand_predicate_family_from_vontology": True,
+                "__predicate_family_max_predicates": 12,
+            }
+        }
+    }
+
+    orchestrator._apply_turn_scoped_tool_payload_support(
+        tool_name="find_relations_with_argument",
+        payload=payload,
+        data=data,
+    )
+
+    assert payload["predicate_filter"] == [
+        "#V#direct_relation",
+        "#V#inverse_relation",
+        "#V#role_relation",
+    ]
+    assert data["relation_predicate_family_context"]["family_concept_ids"] == [
+        "#V#represented_relation_family"
+    ]
+    assert data["tool_payload_support_events"] == [
+        {
+            "tool": "find_relations_with_argument",
+            "field": "predicate_filter",
+            "source": "vontology_predicate_schema",
+            "seed_predicate_ids": ["#V#direct_relation"],
+            "predicate_ids": [
+                "#V#direct_relation",
+                "#V#inverse_relation",
+                "#V#role_relation",
+            ],
+            "family_concept_ids": ["#V#represented_relation_family"],
+            "status": "expanded",
+            "relationship_extent_index_status": "available",
+            "family_discovery_source": "relationship_extent_index",
+        }
+    ]
+
+
 def test_turn_scoped_tool_payload_support_uses_represented_predicate_follow_up_profile():
     orchestrator = InternalMCPChatOrchestrator(
         gateway=cast(Any, _StubGateway()),
diff --git a/tests/backend/test_predicate_family_vontology_service.py b/tests/backend/test_predicate_family_vontology_service.py
new file mode 100644
index 000000000..a206292f9
--- /dev/null
+++ b/tests/backend/test_predicate_family_vontology_service.py
@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+from typing import Any
+
+from src.backend.services import predicate_family_vontology_service as service
+
+
+def test_expand_relation_predicate_family_uses_inverse_and_indexed_family_schema(
+    monkeypatch,
+) -> None:
+    queries: list[dict[str, Any]] = []
+
+    def fake_find(query, projection=None):
+        queries.append({"query": query, "projection": projection})
+        concept_ids = query.get("concept_id", {}).get("$in", [])
+        if concept_ids == ["#V#direct_relation"]:
+            return [
+                {
+                    "concept_id": "#V#direct_relation",
+                    "relationships": {
+                        "#V#predicate_is_inverse_of_predicate": ["#V#inverse_relation"]
+                    },
+                }
+            ]
+        assert concept_ids == ["#V#represented_relation_family"]
+        return [
+            {
+                "concept_id": "#V#represented_relation_family",
+                "relationships": {
+                    "#V#has_member_predicate": [
+                        "#V#direct_relation",
+                        "#V#inverse_relation",
+                        "#V#focal_role",
+                        "#V#related_role",
+                    ],
+                    "#V#has_focal_role_predicate": ["#V#focal_role"],
+                    "#V#has_related_role_predicate": ["#V#related_role"],
+                    "#V#has_reified_relation_type": ["#V#relation_event"],
+                },
+            }
+        ]
+
+    index_queries: list[dict[str, Any]] = []
+
+    def fake_query_relationship_extent_index(**kwargs):
+        captured = dict(kwargs)
+        captured["target_values"] = list(kwargs.get("target_values") or [])
+        index_queries.append(captured)
+        return (
+            [
+                {
+                    "source_concept_id": "#V#represented_relation_family",
+                    "predicate_id": "#V#has_member_predicate",
+                    "target_value": "#V#inverse_relation",
+                }
+            ],
+            1,
+        )
+
+    monkeypatch.setattr(service.ConceptsRepository, "find", staticmethod(fake_find))
+    monkeypatch.setattr(
+        service,
+        "query_relationship_extent_index",
+        fake_query_relationship_extent_index,
+    )
+
+    result = service.expand_relation_predicate_family(["#V#direct_relation"])
+
+    assert result["status"] == "expanded"
+    assert result["predicate_ids"] == [
+        "#V#direct_relation",
+        "#V#inverse_relation",
+        "#V#focal_role",
+        "#V#related_role",
+    ]
+    assert result["family_concept_ids"] == ["#V#represented_relation_family"]
+    assert result["focal_role_predicate_ids"] == ["#V#focal_role"]
+    assert result["related_role_predicate_ids"] == ["#V#related_role"]
+    assert result["reified_relation_type_ids"] == ["#V#relation_event"]
+    assert result["relationship_extent_index_status"] == "available"
+    assert result["family_discovery_source"] == "relationship_extent_index"
+    assert index_queries == [
+        {
+            "predicate_id": "#V#has_member_predicate",
+            "target_values": ["#V#direct_relation", "#V#inverse_relation"],
+            "count_total": False,
+            "projection": {
+                "_id": 0,
+                "source_concept_id": 1,
+                "predicate_id": 1,
+                "target_value": 1,
+            },
+            "limit": 128,
+        }
+    ]
+    assert len(queries) == 2
+
+
+def test_expand_relation_predicate_family_uses_exact_fallback_when_index_unavailable(
+    monkeypatch,
+) -> None:
+    queries: list[dict[str, Any]] = []
+
+    def fake_find(
+        query,
+        projection=None,
+        *,
+        limit=0,
+        max_time_ms=None,
+    ):
+        queries.append(
+            {
+                "query": query,
+                "projection": projection,
+                "limit": limit,
+                "max_time_ms": max_time_ms,
+            }
+        )
+        if "concept_id" in query:
+            return [
+                {
+                    "concept_id": "#V#direct_relation",
+                    "relationships": {
+                        "#V#inverse_predicates": ["#V#inverse_relation"]
+                    },
+                }
+            ]
+        return [
+            {
+                "concept_id": "#V#represented_relation_family",
+                "relationships": {
+                    "#V#has_member_predicate": [
+                        "#V#direct_relation",
+                        "#V#inverse_relation",
+                        "#V#role_relation",
+                    ],
+                    "#V#has_focal_role_predicate": ["#V#role_relation"],
+                },
+            }
+        ]
+
+    monkeypatch.setattr(
+        service.ConceptsRepository,
+        "find",
+        staticmethod(fake_find),
+    )
+    monkeypatch.setattr(
+        service,
+        "query_relationship_extent_index",
+        lambda **_kwargs: ([], -1),
+    )
+
+    result = service.expand_relation_predicate_family(["#V#direct_relation"])
+
+    assert result["status"] == "expanded"
+    assert result["predicate_ids"] == [
+        "#V#direct_relation",
+        "#V#inverse_relation",
+        "#V#role_relation",
+    ]
+    assert result["family_concept_ids"] == [
+        "#V#represented_relation_family"
+    ]
+    assert result["relationship_extent_index_status"] == "unavailable"
+    assert result["family_discovery_source"] == (
+        "canonical_exact_predicate_query"
+    )
+    assert queries[1]["query"] == {
+        "relationships.#V#has_member_predicate": {
+            "$in": ["#V#direct_relation", "#V#inverse_relation"]
+        }
+    }
+    assert queries[1]["limit"] == 32
+    assert queries[1]["max_time_ms"] == 8_000
+
+
+def test_expand_relation_predicate_family_fails_soft(monkeypatch) -> None:
+    monkeypatch.setattr(
+        service.ConceptsRepository,
+        "find",
+        staticmethod(
+            lambda *_args, **_kwargs: (_ for _ in ()).throw(
+                RuntimeError("database unavailable")
+            )
+        ),
+    )
+
+    result = service.expand_relation_predicate_family(["#V#direct_relation"])
+
+    assert result["predicate_ids"] == ["#V#direct_relation"]
+    assert result["status"] == "schema_read_failed"
+    assert result["error_class"] == "RuntimeError"