Skip to content

symfonic.agent.fastapi.tenant_privacy_export

tenant_privacy_export

Export-side helpers for the tenant-privacy router.

Split from :mod:tenant_privacy_router to keep each file below the 300-LOC cap. These helpers report when a source cannot be read.

ExportLayerReadError 

ExportLayerReadError(layer: str)

Bases: RuntimeError

A named memory layer could not be included in an export.

Source code in symfonic/agent/fastapi/tenant_privacy_export.py 
24
25
26
def __init__(self, layer: str) -> None:
    self.layer = layer
    super().__init__(f"export could not read layer: {layer}")

ExportLayerUnsupportedError 

ExportLayerUnsupportedError(layer: str)

Bases: RuntimeError

A named memory source is structurally unavailable for export.

Source code in symfonic/agent/fastapi/tenant_privacy_export.py 
32
33
34
def __init__(self, layer: str) -> None:
    self.layer = layer
    super().__init__(f"export does not support layer: {layer}")

collect_edges_for_export async 

collect_edges_for_export(agent: SymfonicAgent, memory_scope: Any) -> list[dict[str, Any]]

Return every edge owned by the tenant.

The GraphMemoryStore uniquely identifies edges (they live across all layers conceptually), so we hit the semantic layer's graph which is the canonical handle.

Source code in symfonic/agent/fastapi/tenant_privacy_export.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
async def collect_edges_for_export(
    agent: SymfonicAgent,
    memory_scope: Any,
) -> list[dict[str, Any]]:
    """Return every edge owned by the tenant.

    The GraphMemoryStore uniquely identifies edges (they live across
    all layers conceptually), so we hit the semantic layer's graph
    which is the canonical handle.
    """
    semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
    if semantic is None:
        raise ExportLayerUnsupportedError("edges")
    graph = getattr(semantic, "_graph", None)
    if graph is None:
        raise ExportLayerUnsupportedError("edges")
    try:
        edges_raw = await graph.list_edges(memory_scope, limit=100_000)
    except Exception:
        logger.debug("Export: list_edges failed", exc_info=True)
        raise ExportLayerReadError("edges") from None
    return [
        {
            "id": str(e.id),
            "source": str(e.source),
            "target": str(e.target),
            "relationship": e.relationship,
            "weight": e.weight,
            "properties": getattr(e, "properties", {}),
        }
        for e in edges_raw
    ]

collect_nodes_for_export async 

collect_nodes_for_export(agent: SymfonicAgent, memory_scope: Any) -> tuple[dict[str, list[dict[str, Any]]], dict[str, str], dict[str, str]]

Return tenant nodes plus unreadable and unsupported layers.

Layers that are not registered are silently skipped (empty list entries stay out of the payload rather than implying data exists). Layer-specific failures are surfaced with non-sensitive layer names while readable layers remain available in the export.

Source code in symfonic/agent/fastapi/tenant_privacy_export.py 
 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
async def collect_nodes_for_export(
    agent: SymfonicAgent,
    memory_scope: Any,
) -> tuple[
    dict[str, list[dict[str, Any]]], dict[str, str], dict[str, str],
]:
    """Return tenant nodes plus unreadable and unsupported layers.

    Layers that are not registered are silently skipped (empty list
    entries stay out of the payload rather than implying data exists).
    Layer-specific failures are surfaced with non-sensitive layer names while
    readable layers remain available in the export.
    """
    grouped: dict[str, list[dict[str, Any]]] = {}
    failed: dict[str, str] = {}
    unsupported: dict[str, str] = {}
    readable: list[tuple[MemoryLayer, Any]] = []
    for ml in MemoryLayer:
        store = agent._orchestrator.get_layer(ml)
        if store is None:
            continue
        graph = getattr(store, "_graph", None)
        if graph is None:
            # Working layer stores the graph handle under ``_graph_store``.
            graph = getattr(store, "_graph_store", None)
            if graph is None:
                unsupported[ml.value] = "not exported"
                continue
        readable.append((ml, graph))
    results = await asyncio.gather(*(
        _query_nodes_for_export(graph, memory_scope, ml)
        for ml, graph in readable
    ), return_exceptions=True)
    for (ml, _graph), nodes in zip(readable, results, strict=True):
        if isinstance(nodes, ExportLayerReadError):
            failed[nodes.layer] = "read failed"
            continue
        grouped[ml.value] = [
            {
                "id": str(n.id),
                "label": str(n.label),
                "layer": n.layer.value,
                "properties": n.properties,
                "importance": n.importance,
                "access_count": n.access_count,
                "created_at": (
                    str(n.created_at) if n.created_at else None
                ),
                "updated_at": (
                    str(n.updated_at) if n.updated_at else None
                ),
            }
            for n in nodes
        ]
    return grouped, failed, unsupported