def create_tenant_privacy_router(agent: SymfonicAgent) -> APIRouter:
"""Create the tenant-privacy sub-router (export + erase)."""
router = APIRouter(tags=["tenant-privacy"])
@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).
* ``complete`` and ``failed_layers`` — whether every layer was
readable and, for a partial export, which layers were unavailable.
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, failed_layers, unsupported_layers = (
await collect_nodes_for_export(agent, memory_scope)
)
try:
edges = await collect_edges_for_export(agent, memory_scope)
except ExportLayerReadError as exc:
edges = []
failed_layers[exc.layer] = "read failed"
except ExportLayerUnsupportedError as exc:
edges = []
unsupported_layers[exc.layer] = "not exported"
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,
"complete": not failed_layers,
"failed_layers": failed_layers,
"unsupported_layers": unsupported_layers,
}
audit_metadata: dict[str, Any] = {
"node_count": total_nodes,
"edge_count": len(edges),
"schema_version": SCHEMA_VERSION,
"complete": not failed_layers,
"failed_layers": failed_layers,
"unsupported_layers": unsupported_layers,
}
_audit(request, scope, action="export_data", resource_type="tenant",
metadata=audit_metadata)
return payload
@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] = {}
failed_layers: dict[str, str] = {}
# 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
unsupported_layers: dict[str, str] = {}
if graph is None:
unsupported_layers["edges"] = "not erasable"
else:
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:
failed_layers["edges"] = "delete failed"
logger.warning("Erase: failed to delete edge %s", e.id)
except Exception:
failed_layers["edges"] = "read failed"
logger.warning("Erase: list_edges failed", exc_info=True)
counts["edges"] = edges_deleted
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:
failed_layers[ml.value] = "clear failed"
logger.warning("Erase: %s.clear() failed", ml,
exc_info=True)
else:
unsupported_layers[ml.value] = "not erasable"
continue
try:
# include_retracted=True: GDPR Art.17 erasure must delete
# soft-retracted nodes too -- they still hold the user's data.
nodes = await layer_graph.query_nodes(
memory_scope, layer=ml, include_retracted=True,
)
except Exception:
failed_layers[ml.value] = "read failed"
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:
failed_layers[ml.value] = "delete failed"
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:
failed_layers[ml.value] = "clear failed"
logger.debug("working.clear() failed", exc_info=True)
# SEC-PRIV-5 — sweep every store that registered with the framework's
# tenant-erasure path. Governed stores added after this router was
# written (the T2.3.7 recording store is the first) reach erasure
# through this registry rather than by editing the loop above; the
# import is local so the legacy module's import graph is unchanged.
from symfonic.services.shadow.ports import PRIVACY_DELETION_PARTICIPANTS
# Like every other step here, the sweep is guarded: the memory layers
# above are already erased, so a participant raising must not turn a
# mostly-successful purge into a 500 with no completion audit record.
failed_participants: dict[str, str] = {}
try:
sweep = await PRIVACY_DELETION_PARTICIPANTS.sweep(str(scope.tenant_id))
except Exception:
failed_layers["participants"] = "sweep failed"
logger.warning("Erase: privacy-deletion participant sweep failed",
exc_info=True)
else:
counts.update(sweep.counts)
failed_participants = dict(sweep.failed)
metadata: dict[str, Any] = {"stage": "completed", "counts": counts}
if failed_layers:
metadata["failed_layers"] = failed_layers
if failed_participants:
metadata["failed_participants"] = sorted(failed_participants)
metadata["participant_errors"] = failed_participants
if unsupported_layers:
metadata["unsupported_layers"] = unsupported_layers
_audit(request, scope, action="erase_all", resource_type="tenant",
metadata=metadata)
body: dict[str, Any] = {
"tenant_id": str(scope.tenant_id),
"erased_at": datetime.now(UTC).isoformat(),
"counts": counts,
"complete": not failed_layers and not failed_participants,
"failed_layers": failed_layers,
"unsupported_layers": unsupported_layers,
}
if failed_participants:
# A half-succeeded erasure must not read as a clean one.
body["failed_participants"] = sorted(failed_participants)
return body
return router