def mount_procedure_toggle_route(
router: Any, *, procedures: Any, resolve: Any, scope_resolver: Any = None,
) -> None:
"""Mount the bodyless procedure toggle when its admin port is configured."""
from fastapi import Header, HTTPException, Request
@router.post("/procedures/{node_id}/toggle")
async def toggle_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]:
if scope_resolver is None:
from symfonic.agent.fastapi.identity import request_user_id
scope = await resolve(x_tenant_id, authorization, request=request)
user_id = request_user_id(request)
else:
verified = await verified_principal(
scope_resolver, x_tenant_id, authorization, request=request,
)
scope = verified.scope
user_id = str(verified.principal_id)
if len(node_id) > 256:
raise HTTPException(400, "procedure ID exceeds audit resource limit")
audit_entry = AuditLogEntry(
tenant_id=str(scope.tenant_id), user_id=user_id,
action="update_procedure", resource_type="procedure", resource_id=node_id,
metadata={"fields": ["active"]},
ip_address=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
)
v0_scope = memory_scope(scope) # outside the catch-all: 501 stays typed
try:
row = await procedures.toggle_procedure(v0_scope, node_id)
except Exception as unavailable: # noqa: BLE001 - stored data stays private
logger.error("Procedure toggle failed")
raise HTTPException(500, "Internal server error") from unavailable
if row is None:
raise HTTPException(404, f"Node {node_id} not found")
try:
emit_audit_event(audit_entry.model_copy(update={
"metadata": {"fields": ["active"], "new_active": row["properties"].get("active")},
}))
except Exception: # noqa: BLE001 - a committed toggle must not invite a retry
logger.exception("Procedure toggle audit dispatch failed")
return row