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)
admission = query.read_admission
if admission is not None and (
not callable(selector)
or not getattr(graph, "supports_read_admission", False)
):
raise MemoryContractError("graph backend does not support read admission")
if callable(selector):
from symfonic.memory.candidates import CandidateRequest
allowed = None if admission is None else tuple(
_adapter_scope(scope_from_path(path)).scope_path
for path in admission.allowed_scope_paths
)
page = await selector(_adapter_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,
allowed_scope_paths=allowed,
))
nodes, incomplete = page.nodes, not page.complete
else:
nodes = await graph.query_nodes(
_adapter_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)
or not _is_recallable_procedure(node)
):
continue
record = record_from_legacy_node(node_payload(node))
if admission is not None and record.scope.path not in admission.allowed_scope_paths:
raise MemoryContractError(
"graph backend returned a row outside read admission"
)
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,
selected_publication=published_recall_ref(
record.record_id, node_scope(node), node.properties,
tenant_id=node.tenant_id,
),
)
)
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,
"updated_at": node.updated_at,
}
|
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_storage_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 = _adapter_scope(query.scope)
for candidate in candidates:
admission = query.read_admission
if admission is not None:
if candidate.record.scope.path not in admission.allowed_scope_paths:
raise MemoryContractError("vector backend returned a row outside read admission")
owner_scope = _adapter_scope(candidate.record.scope)
else:
owner_scope = trusted_scope
node = await graph.get_node(owner_scope, NodeId(candidate.record.record_id))
if (
node is None
or is_pending(node.properties)
or is_retracted(node.properties)
or not _is_recallable_procedure(node)
):
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,
selected_publication=published_recall_ref(
authoritative.record_id, node_scope(node), node.properties,
tenant_id=node.tenant_id,
),
))
return tuple(verified)
|
tenant_scope
tenant_scope(scope: MemoryScope) -> TenantScope
Shared direct services use positional storage until individually migrated.
Source code in src/symfonic/capabilities/memory/graph_rows.py
| def tenant_scope(scope: MemoryScope) -> TenantScope:
"""Shared direct services use positional storage until individually migrated."""
if scope.identity_version:
raise MemoryContractError("native scope requires an admitted graph adapter")
levels = [
ScopeLevel(kind=kind, id=segment)
for kind, segment in zip(LEGACY_KINDS, scope.segments, strict=False)
]
return TenantScope.from_path(levels)
|