Skip to content

symfonic.memory.retrieval.discovery

discovery

DiscoveryService -- scans the graph for emergent memory block labels.

Instead of matching entries against a hardcoded keyword dictionary, this service queries the graph store for actual distinct labels and merges them with domain-required labels to produce a unified status view.

Used by the Memory Status API to surface per-tenant memory health.

DiscoveryService 

DiscoveryService(orchestrator: object)

Scans the graph store for distinct labels with emergent discovery.

Instead of matching against hardcoded keyword categories, this service asks the graph store what labels actually exist and merges them with any domain-required labels that may not exist yet.

Source code in src/symfonic/memory/retrieval/discovery.py 
46
47
def __init__(self, orchestrator: object) -> None:
    self._orchestrator = orchestrator

scan_status async 

scan_status(
    scope: TenantScope,
    required_labels: list[str] | None = None,
) -> list[MemoryBlockInfo]

Scan the semantic layer for distinct graph labels.

Parameters:

Name Type Description Default
scope TenantScope

Tenant scope for isolation.

 required
required_labels list[str] | None

Labels the domain requires. Missing ones are returned as empty/pending with is_required=True.

 None

Returns:

Type Description
list[MemoryBlockInfo]

List of MemoryBlockInfo -- one per discovered label, plus
list[MemoryBlockInfo]

any missing required labels appended as pending.
Source code in src/symfonic/memory/retrieval/discovery.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
async def scan_status(
    self,
    scope: TenantScope,
    required_labels: list[str] | None = None,
) -> list[MemoryBlockInfo]:
    """Scan the semantic layer for distinct graph labels.

    Args:
        scope: Tenant scope for isolation.
        required_labels: Labels the domain requires. Missing ones
            are returned as empty/pending with ``is_required=True``.

    Returns:
        List of MemoryBlockInfo -- one per discovered label, plus
        any missing required labels appended as pending.
    """
    semantic = self._orchestrator.get_layer(MemoryLayer.SEMANTIC)  # type: ignore[union-attr]

    if semantic is None:
        # No semantic layer -- return required labels as pending stubs.
        return [
            MemoryBlockInfo(label=label, is_required=True)
            for label in (required_labels or [])
        ]

    # Query the graph store for actual distinct labels.
    db_labels = await semantic._graph.distinct_labels(scope)

    req_list = required_labels or []
    results: list[MemoryBlockInfo] = []
    matched_required: set[str] = set()

    def _matches_required(label: str) -> str | None:
        """Check if label matches a required label (exact or prefix)."""
        for req in req_list:
            if label == req or label.startswith(f"{req}:") or label.startswith(f"{req} "):
                return req
        return None

    for entry in db_labels:
        label = entry["label"]
        req_match = _matches_required(label)
        if req_match:
            matched_required.add(req_match)
        results.append(
            MemoryBlockInfo(
                label=label,
                exists=True,
                entry_count=entry["count"],
                last_updated=entry["last_updated"],
                importance=5.0,
                layer=MemoryLayer.SEMANTIC,
                is_required=req_match is not None,
            ),
        )

    # Append any required labels not yet matched as pending stubs.
    for req in req_list:
        if req not in matched_required:
            results.append(MemoryBlockInfo(label=req, is_required=True))

    return results

MemoryBlockInfo

Bases: BaseModel

Status of a named memory block for a tenant.