def mount_semantic_create_route(
router: Any, *, service: Any, resolve: Any, verified_resolver: bool,
) -> None:
"""Mount only when a semantic-create service is explicitly supplied."""
if service is None:
return
from fastapi import Header, Request
@router.post(
"/memories/semantic", status_code=201, response_model=SemanticCreateResponse,
)
async def create_semantic(
body: SemanticCreateRequest,
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
caller = (
await resolve(x_tenant_id, authorization, request=request, principal_only=True)
if verified_resolver else None
)
scope = caller.scope if caller is not None else await resolve(
x_tenant_id, authorization, request=request,
)
row = await service.create(
memory_scope(scope), label=body.label, content=body.content,
properties=body.properties, importance=body.importance,
metadata=body.metadata,
)
logger = get_audit_logger()
if logger is not None:
from symfonic.agent.fastapi.identity import request_user_id
user_id = caller.principal_id if caller is not None else request_user_id(request)
entry = AuditLogEntry(
tenant_id=str(scope.tenant_id),
user_id=str(user_id) if user_id else None,
action="memory.create.semantic", resource_type="memory_node",
resource_id=row["id"], metadata={"label": row["label"]},
ip_address=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
)
try:
await logger.record(entry)
except Exception: # noqa: BLE001 - legacy audit is best effort
import logging
logging.getLogger(__name__).warning(
"Audit logger record() raised — audit event lost (action=%s)",
entry.action,
)
return row