symfonic.agent.fastapi.memory_create_helpers

memory_create_helpers

Shared helpers for the memory-create handlers.

Extracted so memory_create_handlers.py can stay under the 300-LOC cap with room for the five per-layer handler bodies.

emit_create_audit async

emit_create_audit(
    request: Request,
    scope: FrameworkTenantScope,
    *,
    layer: MemoryLayer,
    resource_id: str,
    metadata: dict[str, Any] | None = None,
) -> None

Await the audit record directly so the entry is committed before the handler returns its response.

Using emit_audit_event (fire-and-forget via loop.create_task) caused a CI-only race: TestClient's per-request event loop tears down before the scheduled task executes, so drain() always saw an empty list. Awaiting directly eliminates that window at negligible cost (the in-memory logger is a lock + list append). Audit failures are swallowed so a sick audit store can never 500 a user request.

Source code in src/symfonic/agent/fastapi/memory_create_helpers.py

async def emit_create_audit(
    request: Request,
    scope: FrameworkTenantScope,
    *,
    layer: MemoryLayer,
    resource_id: str,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Await the audit record directly so the entry is committed before
    the handler returns its response.

    Using ``emit_audit_event`` (fire-and-forget via ``loop.create_task``)
    caused a CI-only race: TestClient's per-request event loop tears down
    before the scheduled task executes, so ``drain()`` always saw an empty
    list.  Awaiting directly eliminates that window at negligible cost
    (the in-memory logger is a lock + list append).  Audit failures are
    swallowed so a sick audit store can never 500 a user request.
    """
    logger_ = get_audit_logger()
    if logger_ is None:
        return
    user_id = getattr(request.state, "user_id", None)
    ip = request.client.host if request.client else None
    ua = request.headers.get("user-agent")
    entry = AuditLogEntry(
        tenant_id=str(scope.tenant_id),
        user_id=str(user_id) if user_id else None,
        action=f"memory.create.{layer.value}",
        resource_type="memory_node",
        resource_id=resource_id,
        metadata=metadata or {},
        ip_address=ip,
        user_agent=ua,
    )
    try:
        await logger_.record(entry)
    except Exception:  # noqa: BLE001
        import logging as _logging
        _logging.getLogger(__name__).warning(
            "Audit logger record() raised — audit event lost (action=%s)",
            entry.action,
        )

require_layer

require_layer(
    agent: SymfonicAgent, layer: MemoryLayer
) -> Any

Return the layer store or raise 404 if the layer is disabled.

Source code in src/symfonic/agent/fastapi/memory_create_helpers.py

def require_layer(agent: SymfonicAgent, layer: MemoryLayer) -> Any:
    """Return the layer store or raise 404 if the layer is disabled."""
    store = agent._orchestrator.get_layer(layer)
    if store is None:
        raise HTTPException(
            status_code=404,
            detail=f"{layer.value} layer not available",
        )
    return store

scope_error

scope_error(exc: Exception) -> HTTPException

Translate SecurityScopeError / ScopeValidationError → HTTPException.

Source code in src/symfonic/agent/fastapi/memory_create_helpers.py

def scope_error(exc: Exception) -> HTTPException:
    """Translate SecurityScopeError / ScopeValidationError → HTTPException."""
    if isinstance(exc, SecurityScopeError):
        return HTTPException(status_code=403, detail=str(exc))
    if isinstance(exc, ScopeValidationError):
        return HTTPException(status_code=400, detail=str(exc))
    return HTTPException(status_code=500, detail=str(exc))