feat(atlas-handoff): graph-pivot results through unified SearchService

- SearchService.graph_pivot_results(): wraps linked_neighbors() output
  in SearchResult model with match_reasons (graph_pivot, relation:X,
  direction:X, atlas:scope), scores (hop-distance based), and relation
  meta string for UI surfaces
- AtlasWayfinderScreen: propagates atlas_source_scope to LinkedItemsPanel
  when opening entries or recent items from the wayfinder
- LinkedItemsPanel: adds #linked-atlas-context label (◈ Atlas context:
  scope_path) that refreshes via property setter; atlas scope is
  preserved across linked-panel pivots; _update_atlas_context_label()
  helper keeps display in sync

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
thePR0M3TH3ANandClaude Sonnet 4.6 committed 2026-03-11 07:34:48 -04:00
1 parent 2ab73c50a0
commit b4e84a4327
3 files changed
+170 -1

No files matched your search

+120
View File
@@ -1815,6 +1815,126 @@ class SearchService:
limit=limit,
)
def graph_pivot_results(
self,
pivot_entry_id: int,
*,
pivot_scope: str | None = None,
direction: str = "both",
include_archived: bool = True,
limit: int = 50,
) -> list[dict[str, Any]]:
"""Return neighbors of ``pivot_entry_id`` as :class:`SearchResult`-compatible dicts.
Wraps graph-traversal results in the unified SearchResult model so that
atlas-pivot navigation produces results with proper ``match_reasons``,
``score``, and relation context — identical in shape to
:meth:`search` output and ready for any UI surface that consumes it.
Scoring:
- Direct (hop-1) outgoing links score highest (~1.0).
- Incoming links get a small direction penalty (−0.1).
- Each additional hop beyond 1 reduces the score by 0.3.
"""
with self._lock:
scope_path = self._scope_path()
target_id = int(pivot_entry_id or 0)
if target_id <= 0:
return []
neighbors = self.linked_neighbors(
target_id,
direction=direction,
include_archived=include_archived,
limit=limit,
)
em = getattr(self._manager, "entry_manager", None)
results: list[dict[str, Any]] = []
for neighbor in neighbors:
neighbor_id = int(neighbor.get("entry_id", 0) or 0)
if neighbor_id <= 0:
continue
# Fetch full entry for excerpt / meta
entry: dict[str, Any] = {}
if em is not None:
try:
entry = em.retrieve_entry(neighbor_id) or {}
except Exception:
pass
kind = str(neighbor.get("kind", "")).strip()
relation = str(neighbor.get("relation", "")).strip()
direction_val = str(neighbor.get("direction", "")).strip()
hop = int(neighbor.get("hop", 1) or 1)
# Score: hop-1 outgoing ≈ 1.0, incoming −0.1, each extra hop −0.3
hop_penalty = (hop - 1) * 0.3
direction_penalty = 0.1 if direction_val == "incoming" else 0.0
total_score = round(max(0.05, 1.0 - hop_penalty - direction_penalty), 6)
match_reasons: list[str] = ["graph_pivot"]
if relation:
match_reasons.append(f"relation:{relation}")
if direction_val:
match_reasons.append(f"direction:{direction_val}")
if pivot_scope:
match_reasons.append(f"atlas:{pivot_scope}")
# Relation context visible as meta string
arrow = "->" if direction_val == "outgoing" else "<-"
meta_parts: list[str] = []
if relation:
meta_parts.append(f"{arrow}[{relation}]")
if pivot_scope:
meta_parts.append(f"atlas:{pivot_scope}")
meta_str = " ".join(meta_parts) or self._meta(entry, kind)
modified_ts = int(entry.get("modified_ts", 0) or 0)
result = SearchResult(
entry_id=neighbor_id,
label=str(neighbor.get("label", "")).strip(),
kind=kind,
scope_path=scope_path,
archived=bool(neighbor.get("archived", False)),
score=total_score,
score_breakdown={
"lexical": 0.0,
"semantic": 0.0,
"structural": total_score,
"recency": 0.0,
},
match_reasons=sorted(set(match_reasons)),
excerpt=(
""
if kind in self._SECRET_EXCERPT_KINDS
else self._safe_excerpt(entry, kind)
),
linked_hits=[
{
"target_id": target_id,
"relation": relation,
}
],
tags=list(neighbor.get("tags", [])),
modified_ts=modified_ts,
meta=meta_str,
).model_dump()
results.append(result)
results.sort(
key=lambda r: (
-float(r.get("score", 0.0)),
str(r.get("label", "")).lower(),
int(r.get("entry_id", 0)),
)
)
return results[: max(1, int(limit))]
class AtlasService:
"""Thread-safe wrapper around canonical atlas/index0 read operations."""
+13
View File
@@ -114,6 +114,17 @@ class AtlasWayfinderScreen(Screen):
classes="maintenance-footer",
)
def _propagate_atlas_context(self, scope_path: str) -> None:
"""Set atlas_source_scope on the inspector LinkedItemsPanel if present."""
if not scope_path:
return
try:
from ..widgets.inspector import LinkedItemsPanel # noqa: PLC0415
panel = self.app.query_one(LinkedItemsPanel)
panel.atlas_source_scope = scope_path
except Exception:
pass
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id or ""
if button_id == "atlas-filter-all":
@@ -133,6 +144,7 @@ class AtlasWayfinderScreen(Screen):
entry_id = int(button_id.removeprefix("atlas-open-entry-"))
except ValueError:
return
self._propagate_atlas_context(str(self.payload.get("scope_path", "")))
self.app.selected_entry_id = entry_id
self.app.pop_screen()
return
@@ -141,5 +153,6 @@ class AtlasWayfinderScreen(Screen):
entry_id = int(button_id.removeprefix("atlas-open-recent-"))
except ValueError:
return
self._propagate_atlas_context(str(self.payload.get("scope_path", "")))
self.app.selected_entry_id = entry_id
self.app.pop_screen()
+37 -1
View File
@@ -750,6 +750,11 @@ class LinkedItemsPanel(Vertical):
background: #1a1a2e;
color: #aaaaff;
}
#linked-atlas-context {
color: #7777cc;
margin-bottom: 1;
height: auto;
}
.linked-nav-hint {
color: #555555;
margin-top: 1;
@@ -766,10 +771,21 @@ class LinkedItemsPanel(Vertical):
self._focused_idx: int = -1
# Set by the app when the user arrived here from an atlas wayfinder view.
# Format: scope_path string or None
self.atlas_source_scope: str | None = None
self._atlas_source_scope: str | None = None
@property
def atlas_source_scope(self) -> str | None:
return self._atlas_source_scope
@atlas_source_scope.setter
def atlas_source_scope(self, value: str | None) -> None:
self._atlas_source_scope = value
if self.is_mounted:
self._update_atlas_context_label()
def compose(self) -> ComposeResult:
yield Label("Linked Items", id="linked-items-title")
yield Label("", id="linked-atlas-context")
yield Label("No linked items.", id="linked-items-summary", classes="linked-summary")
yield Label("", id="linked-filter-bar", classes="linked-filter-bar")
with Vertical(id="linked-items-list"):
@@ -864,12 +880,25 @@ class LinkedItemsPanel(Vertical):
)
)
def _update_atlas_context_label(self) -> None:
"""Refresh the atlas context label based on current atlas_source_scope."""
try:
ctx_label = self.query_one("#linked-atlas-context", Label)
if self.atlas_source_scope:
ctx_label.update(f"[dim]◈ Atlas context: {self.atlas_source_scope}[/dim]")
else:
ctx_label.update("")
except Exception:
pass
def update_entry(self, entry_id: int | None) -> None:
summary_label = self.query_one("#linked-items-summary", Label)
filter_bar = self.query_one("#linked-filter-bar", Label)
list_container = self.query_one("#linked-items-list", Vertical)
nav_hint = self.query_one("#linked-nav-hint", Label)
self._update_atlas_context_label()
if entry_id is None:
summary_label.update("No linked items.")
filter_bar.update("")
@@ -1008,5 +1037,12 @@ class LinkedItemsPanel(Vertical):
except ValueError:
self.app.notify("Invalid linked item target", severity="error")
return
# Preserve atlas context across linked-panel pivots so the scope
# label remains visible and the back button stays available.
current_scope = self.atlas_source_scope
self.app.selected_entry_id = entry_id
if current_scope:
# Re-apply after entry selection (selected_entry_id may trigger
# update_entry which clears the panel but keeps atlas_source_scope).
self.atlas_source_scope = current_scope
self.app.notify(f"Opened linked entry #{entry_id}")