Turning a memory into a stored row, and scoring one against a cue.
Lifted out of the store so that module is about the ports -- retrieve,
write, flush, forget -- and this one is about the row underneath them. They
shared a file until the store grew a second recall route, which is the point
at which two concepts sharing one file stops being a convenience.
lexical_score is the cue-overlap half of recall, and naming it that way
is the change: it used to be the score. A vector route scores by meaning
and this one by words, the store blends them, and a reader can now see
exactly what each contributes.
graph_candidates is the read those two serve: every stored row this scope
may see, minus the ones no read should. That filter is worth its own place --
one of its two clauses was missing for as long as the adapter existed, and a
merged-away memory kept being recalled because of it.
graph_candidates
async
graph_candidates(graph: Any, query: MemoryQuery, visible: Any, *, unavailable: list[str] | None = None) -> tuple[RetrievedMemory, ...]
Every row the graph route offers for query, scored by words.
Two kinds of row are dropped before anything else looks at them, and both
are the store's single read-side choke point rather than a caller's
responsibility:
- pending -- staged and not yet published. The write/flush split
promises a staged memory is not retrievable, and this is where that
promise is kept.
- retracted -- superseded by a consolidation phase. The legacy store
drops these in
query_nodes(include_retracted=False); this adapter
reads through the backend protocol, which has no such argument, so for
as long as it existed a memory phase 13 merged away was still recalled
and the merge was invisible to every consumer that matters.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| async def graph_candidates(
graph: Any, query: MemoryQuery, visible: Any, *, unavailable: list[str] | None = None
) -> tuple[RetrievedMemory, ...]:
"""Every row the graph route offers for ``query``, scored by words.
Two kinds of row are dropped before anything else looks at them, and both
are the store's single read-side choke point rather than a caller's
responsibility:
* **pending** -- staged and not yet published. The write/flush split
promises a staged memory is not retrievable, and this is where that
promise is kept.
* **retracted** -- superseded by a consolidation phase. The legacy store
drops these in ``query_nodes(include_retracted=False)``; this adapter
reads through the backend protocol, which has no such argument, so for
as long as it existed a memory phase 13 merged away was still recalled
and the merge was invisible to every consumer that matters.
"""
selector = getattr(graph, "select_candidates", None)
if callable(selector):
from symfonic.memory.candidates import CandidateRequest
page = await selector(tenant_scope(query.scope), CandidateRequest(
cue=query.cue, layers=tuple(layer.value for layer in query.layers),
session_id=query.session_id, limit=query.candidate_limit,
))
nodes, incomplete = page.nodes, not page.complete
else:
nodes = await graph.query_nodes(
tenant_scope(query.scope), {}, limit=query.candidate_limit,
)
incomplete = len(nodes) >= query.candidate_limit
if incomplete and unavailable is not None:
unavailable.append("graph_scan_incomplete")
nodes = nodes[:query.candidate_limit]
candidates: list[RetrievedMemory] = []
for node in nodes:
if is_pending(node.properties) or is_retracted(node.properties):
continue
record = record_from_legacy_node(node_payload(node))
if record.layer not in query.layers:
continue
distance = visible(record)
if distance < 0:
# The backend's own prefix filter already excluded these; the
# re-check is the same posture the bridge takes toward an
# adapter -- the component that might be wrong is the one below.
continue
candidates.append(
RetrievedMemory(
record=record,
score=lexical_score(record, query.cue, distance),
scope_distance=distance,
)
)
return tuple(candidates)
|
lexical_score
lexical_score(record: MemoryRecord, cue: str, distance: int) -> float
The reference adapter's formula, restated over the public helpers.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| def lexical_score(record: MemoryRecord, cue: str, distance: int) -> float:
"""The reference adapter's formula, restated over the public helpers."""
overlap = cue_overlap(cue, record.text)
return record.salience * (_FLOOR + (1.0 - _FLOOR) * overlap) * (0.5**distance)
|
node_payload
node_payload(node: MemoryNode) -> dict[str, Any]
The mapping compat.record_from_legacy_node reads, off a real node.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| def node_payload(node: MemoryNode) -> dict[str, Any]:
"""The mapping ``compat.record_from_legacy_node`` reads, off a real node."""
return {
"id": str(node.id),
"layer": node.layer.value,
"tenant_id": node.tenant_id,
"label": node.label,
"properties": dict(node.properties),
"importance": node.importance,
}
|
node_scope
node_scope(node: MemoryNode) -> MemoryScope
The scope a stored row actually sits at.
A pre-v8.0 row carries no scope_path and dual-reads as tenant-global,
which is the same conservative backfill every shipped backend applies.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| def node_scope(node: MemoryNode) -> MemoryScope:
"""The scope a stored row actually sits at.
A pre-v8.0 row carries no ``scope_path`` and dual-reads as tenant-global,
which is the same conservative backfill every shipped backend applies.
"""
stored = node.properties.get(SCOPE_PATH_KEY)
if isinstance(stored, str) and stored:
return scope_from_legacy_path(stored)
return MemoryScope(node.tenant_id)
|
published_vector_candidates
async
published_vector_candidates(graph: Any, query: MemoryQuery, candidates: tuple[RetrievedMemory, ...], visible: Any) -> tuple[RetrievedMemory, ...]
Validate bounded hit ids against graph authority, beyond the lexical page.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| async def published_vector_candidates(
graph: Any, query: MemoryQuery, candidates: tuple[RetrievedMemory, ...], visible: Any
) -> tuple[RetrievedMemory, ...]:
"""Validate bounded hit ids against graph authority, beyond the lexical page."""
verified: list[RetrievedMemory] = []
trusted_scope = tenant_scope(query.scope)
for candidate in candidates:
node = await graph.get_node(trusted_scope, NodeId(candidate.record.record_id))
if node is None or is_pending(node.properties) or is_retracted(node.properties):
continue
authoritative = record_from_legacy_node(node_payload(node))
distance = visible(authoritative)
if distance >= 0 and authoritative.layer in query.layers:
verified.append(RetrievedMemory(
record=authoritative, score=candidate.score, scope_distance=distance,
))
return tuple(verified)
|
tenant_scope
tenant_scope(scope: MemoryScope) -> TenantScope
The legacy scope value for a capability scope. Kinds from compat.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| def tenant_scope(scope: MemoryScope) -> TenantScope:
"""The legacy scope value for a capability scope. Kinds from ``compat``."""
levels = [
ScopeLevel(kind=kind, id=segment)
for kind, segment in zip(LEGACY_KINDS, scope.segments, strict=False)
]
return TenantScope.from_path(levels)
|