def mount_data_routes(
router: Any, *, memory: Any, graph: Any, resolve: Any, edge_delete: Any = None,
principal: Any = None, memory_audit: Any = None, privacy: Any = None,
procedures: Any = None, scope_resolver: Any = None, working: Any = None,
) -> None:
"""Add the store-backed routes to ``router``.
Args:
router: the ``APIRouter`` the agent routes are already on.
memory: a ``MemoryAdminService``, or ``None``.
graph: a ``GraphAdminService``, or ``None``.
resolve: the transport's authenticated-scope resolver. Passed in
rather than rebuilt, so these routes cannot end up authorising
differently from the ones next to them.
"""
from fastapi import Header, HTTPException, Request, Response
if edge_delete is not None:
from symfonic.platform.transport_edge_delete import mount_edge_delete_route
mount_edge_delete_route(router, service=edge_delete, principal=principal,
sink=memory_audit)
if procedures is not None:
from symfonic.platform.transport_procedure_toggle import mount_procedure_toggle_route
from symfonic.platform.transport_procedure_update import mount_procedure_update_route
mount_procedure_toggle_route(
router, procedures=procedures, resolve=resolve, scope_resolver=scope_resolver,
)
mount_procedure_update_route(
router, procedures=procedures, resolve=resolve, scope_resolver=scope_resolver,
)
from symfonic.platform.transport_procedure_delete import mount_procedure_delete_route
mount_procedure_delete_route(router, procedures=procedures, principal=principal,
sink=memory_audit)
@router.get("/procedures/drafts")
async def list_draft_procedures(
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
scope = await resolve(x_tenant_id, authorization, request=request)
v0_scope = memory_scope(scope) # outside the catch-all: 501 stays typed
try:
return await procedures.list_draft_procedures(v0_scope)
except Exception as unavailable: # noqa: BLE001 - do not expose stored data
logger.error("Procedure draft inventory failed")
raise HTTPException(500, "Internal server error") from unavailable
@router.get("/procedures/{node_id}")
async def get_procedure(
node_id: str,
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
scope = await resolve(x_tenant_id, authorization, request=request)
v0_scope = memory_scope(scope) # outside the catch-all: 501 stays typed
try:
row = await procedures.get_procedure(v0_scope, node_id)
except Exception as unavailable: # noqa: BLE001 - do not expose stored data
logger.error("Procedure detail failed")
raise HTTPException(500, "Internal server error") from unavailable
if row is None:
raise HTTPException(404, f"Node {node_id} not found")
return row
@router.get("/procedures")
async def list_procedures(
request: Request,
status: str | None = None,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
scope = await resolve(x_tenant_id, authorization, request=request)
v0_scope = memory_scope(scope) # outside the catch-all: 501 stays typed
try:
return await procedures.list_procedures(v0_scope, status=status)
except Exception as unavailable: # noqa: BLE001 - do not expose stored data
logger.error("Procedure inventory failed")
raise HTTPException(500, "Internal server error") from unavailable
if memory is not None:
@router.get("/memories")
async def list_memories(
response: Response,
request: Request,
layer: str | None = None,
limit: int = 200,
cursor: str | None = None,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
"""A published inventory page in stable record-ID order.
Reads through the memory admin service, so it sees exactly what a
turn would recall -- the same store, the same scope, the same
published-only rule. A browser built on a second query path is a
browser that can disagree with the agent.
"""
scope, guard = await working(x_tenant_id, authorization, request=request)
try:
page = await memory.inventory_page(
record_scope(scope), layer=memory_layer(layer), limit=limit, cursor=cursor,
)
except NotImplementedError as unsupported:
raise HTTPException(status_code=501, detail=str(unsupported)) from unsupported
except ValueError as unknown:
raise HTTPException(status_code=400, detail=str(unknown)) from unknown
if not all([await guard.allows(record, "list") for record in page.records]):
# Do not expose a keyset cursor encoding a denied record ID.
raise HTTPException(503, "owner-restricted memory page unavailable")
response.headers["X-Memory-Has-More"] = str(page.next_cursor is not None).lower()
if page.next_cursor is not None:
response.headers["X-Memory-Next-Cursor"] = page.next_cursor
response.headers["Access-Control-Expose-Headers"] = (
"X-Memory-Has-More, X-Memory-Next-Cursor"
)
return [record_body(record) for record in page.records]
if graph is not None:
from symfonic.capabilities.memory.graph_admin import GraphNodeNotFound
from symfonic.capabilities.memory.graph_owner_reads import OwnerReadUnavailable
from symfonic.platform.transport_graph_paths import mount_graph_path_route
if callable(getattr(graph, "shortest_path", None)):
mount_graph_path_route(router, graph=graph, resolve=resolve, working=working)
if callable(getattr(graph, "clusters", None)):
@router.get("/graph/clusters")
async def list_graph_clusters(
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
"""Complete semantic label groups for the authenticated scope."""
scope = await resolve(x_tenant_id, authorization, request=request)
v0_scope = memory_scope(scope) # outside the catch-all: 501 stays typed
try:
return await graph.clusters(v0_scope)
except HTTPException:
raise
except Exception as unavailable: # noqa: BLE001 - sanitize port failures
# Port exceptions can contain credentials or memory data.
# Record the failed operation without payloads or traceback.
logger.error("Graph cluster projection failed")
raise HTTPException(500, "Internal server error") from unavailable
@router.get("/graph/nodes")
async def list_graph_nodes(
request: Request,
limit: int = 200,
offset: int = 0,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
"""The nodes paired with the graph relationship endpoint."""
scope, guard = await working(x_tenant_id, authorization, request=request)
if limit < 1 or limit > 500 or offset < 0:
raise HTTPException(
status_code=400,
detail="limit must be 1..500 and offset must be non-negative",
)
try:
return await graph.nodes(
memory_scope(scope), limit=limit, offset=offset, admit=guard.admission("graph"),
)
except OwnerReadUnavailable as unavailable:
raise HTTPException(503, "owner-restricted graph page unavailable") from unavailable
@router.get("/edges")
async def list_edges(
request: Request,
limit: int = 200,
offset: int = 0,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
"""The relationships this tenant can see."""
scope, guard = await working(x_tenant_id, authorization, request=request)
if limit < 1 or limit > 500 or offset < 0:
raise HTTPException(
status_code=400,
detail="limit must be 1..500 and offset must be non-negative",
)
try:
return await graph.edges(
memory_scope(scope), limit=limit, offset=offset, admit=guard.admission("graph"),
)
except OwnerReadUnavailable as unavailable:
raise HTTPException(503, "owner-restricted edge page unavailable") from unavailable
@router.get("/graph/nodes/{node_id}/neighborhood")
async def neighborhood(
node_id: str,
request: Request,
depth: int = 2,
min_weight: float = 0,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
"""One node's neighbours, out to ``depth`` hops."""
scope, guard = await working(x_tenant_id, authorization, request=request)
v0_scope = memory_scope(scope) # outside the catch-all: 501 stays typed
try:
return await graph.neighborhood(
v0_scope, node_id, depth=depth, min_weight=min_weight,
admit=guard.admission("graph"),
)
except GraphNodeNotFound as missing:
raise HTTPException(404, f"Node {node_id} not found") from missing
except HTTPException:
raise
except Exception as unavailable: # noqa: BLE001 - sanitize port failures
logger.error("Graph neighborhood query failed")
raise HTTPException(500, "Internal server error") from unavailable
if graph is not None and privacy is None:
@router.get("/tenants/me/export")
async def export_tenant_data(
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
"""GDPR Article 20 -- everything this tenant owns, as one document.
A partial export still answers the request and says so in
``complete``: a portability response that silently omitted a layer
would be a failure nobody could see.
"""
scope, guard = await working(x_tenant_id, authorization, request=request)
return await graph.export(memory_scope(scope), admit=guard.admission("export"))
if memory is not None and privacy is None:
@router.delete("/tenants/me/data")
async def erase_tenant_data(
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
"""GDPR Article 17 -- erase this tenant and everything beneath it.
``forget`` walks the subtree, which is the direction erasure has to
run: a request that deleted a tenant and left its sessions behind
would report success and leave the data.
"""
await resolve(x_tenant_id, authorization, request=request)
# Tenant membership is not authority over private descendant chats.
# The generic store port cannot prove a complete authorized subtree.
raise HTTPException(503, "erasure requires the explicit privacy service")