def create_tenant_privacy_router(agent: SymfonicAgent) -> APIRouter:
"""Create the tenant-privacy sub-router (export + erase)."""
router = APIRouter(tags=["tenant-privacy"])
# ------------------------------------------------------------------
# GET /tenants/me/export
# ------------------------------------------------------------------
@router.get("/tenants/me/export")
async def export_tenant_data(
request: Request,
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""GDPR Article 20 — Right to Data Portability.
Returns a JSON document containing every piece of data owned by
the caller's tenant:
* ``memories`` — nodes grouped by layer (semantic, episodic,
procedural, prospective, working).
* ``edges`` — all relationships between semantic nodes.
* ``schema_version`` — payload format version (bump on breaking
changes so clients can migrate).
Scaffolded apps append conversation history + token-usage
history to this shape (see ``app/admin/router.py.j2``) via a
post-processing hook. The library payload is the minimum
portable representation.
"""
memory_scope = scope.to_memory_scope()
memories = await collect_nodes_for_export(agent, memory_scope)
edges = await collect_edges_for_export(agent, memory_scope)
total_nodes = sum(len(v) for v in memories.values())
payload: dict[str, Any] = {
"tenant_id": str(scope.tenant_id),
"exported_at": datetime.now(UTC).isoformat(),
"schema_version": SCHEMA_VERSION,
"memories": memories,
"edges": edges,
}
_audit(
request, scope,
action="export_data",
resource_type="tenant",
metadata={
"node_count": total_nodes,
"edge_count": len(edges),
"schema_version": SCHEMA_VERSION,
},
)
return payload
# ------------------------------------------------------------------
# DELETE /tenants/me/data
# ------------------------------------------------------------------
@router.delete("/tenants/me/data")
async def erase_tenant_data(
request: Request,
confirmation: str = Query( # noqa: B008
...,
description=(
"Must equal 'DELETE-<tenant_id>' exactly to confirm "
"destructive intent."
),
),
scope: FrameworkTenantScope = Depends(get_tenant_scope), # noqa: B008
) -> dict[str, Any]:
"""GDPR Article 17 — Right to Erasure.
Purges every piece of data owned by the caller's tenant. To
guard against accidental fires (e.g. a mis-wired admin UI,
overzealous CI), the caller MUST pass
``?confirmation=DELETE-<tenant_id>`` where ``<tenant_id>`` is
exactly the tenant id resolved from the ``X-Tenant-ID`` header.
Returns counts of deleted resources. The audit event is
emitted BEFORE the deletion loop so a failure mid-purge still
leaves a trail of intent.
"""
expected = f"DELETE-{scope.tenant_id}"
if confirmation != expected:
raise HTTPException(
status_code=400,
detail=(
"Missing or incorrect confirmation. Pass "
"?confirmation=DELETE-<tenant_id> to proceed."
),
)
memory_scope = scope.to_memory_scope()
counts: dict[str, int] = {}
# Emit the intent event before we start deleting so a partial
# failure still leaves an "erase was attempted" audit record.
_audit(
request, scope,
action="erase_all",
resource_type="tenant",
metadata={"stage": "initiated", "confirmation": expected},
)
# Delete edges first so cascade failures don't leave dangling.
semantic = agent._orchestrator.get_layer(MemoryLayer.SEMANTIC)
graph = getattr(semantic, "_graph", None) if semantic else None
edges_deleted = 0
if graph is not None:
try:
edges = await graph.list_edges(memory_scope, limit=100_000)
for e in edges:
try:
await graph.delete_edge(memory_scope, EdgeId(str(e.id)))
edges_deleted += 1
except Exception:
logger.warning(
"Erase: failed to delete edge %s", e.id,
)
except Exception:
logger.warning("Erase: list_edges failed", exc_info=True)
counts["edges"] = edges_deleted
# Delete nodes per layer.
for ml in MemoryLayer:
store = agent._orchestrator.get_layer(ml)
if store is None:
continue
layer_graph = getattr(store, "_graph", None)
if layer_graph is None:
layer_graph = getattr(store, "_graph_store", None)
if layer_graph is None:
# Working layer w/o graph persistence — clear buffers.
clear = getattr(store, "clear", None)
if callable(clear):
try:
await clear(memory_scope)
except Exception:
logger.warning(
"Erase: %s.clear() failed", ml, exc_info=True,
)
continue
try:
# include_retracted=True: GDPR Art.17 erasure must delete
# soft-retracted nodes too -- they still hold the user's data
# and are hidden from the default read path by design.
nodes = await layer_graph.query_nodes(
memory_scope, layer=ml, include_retracted=True,
)
except Exception:
logger.warning(
"Erase: query_nodes(%s) failed", ml, exc_info=True,
)
continue
deleted = 0
for n in nodes:
try:
await layer_graph.delete_node(
memory_scope, NodeId(str(n.id)),
)
deleted += 1
except Exception:
logger.warning(
"Erase: delete_node %s failed", n.id,
)
counts[ml.value] = deleted
# Drain working-layer buffer too even when graph-mirrored.
if ml == MemoryLayer.WORKING:
clear = getattr(store, "clear", None)
if callable(clear):
try:
await clear(memory_scope)
except Exception:
logger.debug(
"working.clear() failed", exc_info=True,
)
_audit(
request, scope,
action="erase_all",
resource_type="tenant",
metadata={"stage": "completed", "counts": counts},
)
return {
"tenant_id": str(scope.tenant_id),
"erased_at": datetime.now(UTC).isoformat(),
"counts": counts,
}
return router