"""RBAC-scoped data lookups exposed to the assistant as tool calls. Every function takes `user` first and applies the same scoping rule the real ViewSets use (`user_has_full_scope`): full access for superusers/managers, otherwise limited to the user's own projects/tasks. The model never sees unscoped data — scoping happens here, not in the prompt. """ from django.db.models import Q from apps.core.permissions import user_has_full_scope from apps.projects.models import Project, ProjectTeamMember from apps.tasks.models import Task TOOL_SCHEMAS = [ { "type": "function", "function": { "name": "list_projects", "description": "List projects visible to the current user, optionally filtered by status/ownership.", "parameters": { "type": "object", "properties": { "status": { "type": "string", "enum": ["planning", "active", "on_hold", "completed", "archived"], "description": "Filter by project status. Omit to list all visible projects.", }, "mine": { "type": "boolean", "description": "Set true when the user asks about their own/my projects — filters to projects where the current user is lead or a team member.", }, }, }, }, }, { "type": "function", "function": { "name": "get_project_detail", "description": "Get full detail for one project (team, funding, schedule, disseminations) by name or id.", "parameters": { "type": "object", "properties": { "project": {"type": "string", "description": "Project id (UUID) or exact/partial name."}, }, "required": ["project"], }, }, }, { "type": "function", "function": { "name": "list_tasks", "description": "List tasks visible to the current user, optionally filtered by status/project/assignee.", "parameters": { "type": "object", "properties": { "status": { "type": "string", "enum": ["todo", "in_progress", "review", "done", "cancelled", "archived"], }, "project": {"type": "string", "description": "Project id (UUID) or name to filter by."}, "mine": { "type": "boolean", "description": "Set true when the user asks about their own/my tasks — filters to tasks assigned to the current user.", }, }, }, }, }, ] def _visible_projects(user, organization): qs = Project.objects.filter(organization=organization).select_related("lead__user") if user_has_full_scope(user, "project", organization): return qs return qs.filter( Q(lead__user=user) | Q(team_members__user=user) ).distinct() def _visible_tasks(user, organization): qs = Task.objects.filter(organization=organization).select_related("assignee", "project", "created_by") if user_has_full_scope(user, "tasks", organization): return qs member_projects = ProjectTeamMember.objects.filter(user=user).values("project_id") return qs.filter( Q(created_by=user) | Q(assignee=user) | Q(project_id__in=member_projects) | Q(project__lead__user=user) ).distinct() def list_projects(user, organization, status=None, mine=False): qs = _visible_projects(user, organization) if mine: qs = qs.filter(Q(lead__user=user) | Q(team_members__user=user)).distinct() if status: qs = qs.filter(status=status) return [ { "id": str(p.id), "name": p.name, "status": p.status, "lead": p.lead.user.get_full_name() or p.lead.user.email if p.lead and p.lead.user_id else None, "start_date": str(p.start_date) if p.start_date else None, "end_date": str(p.end_date) if p.end_date else None, } for p in qs[:50] ] def get_project_detail(user, organization, project): qs = _visible_projects(user, organization) p = qs.filter(id=project).first() if _looks_like_uuid(project) else None if not p: p = qs.filter(name__icontains=project).first() if not p: return {"error": "Project not found or not visible to you."} team = list(p.team_members.select_related("user")[:20]) funds = list(p.funds.select_related("donor")[:20]) return { "id": str(p.id), "name": p.name, "status": p.status, "start_date": str(p.start_date) if p.start_date else None, "end_date": str(p.end_date) if p.end_date else None, "tags": p.tags, "team": [ {"name": tm.user.get_full_name() or tm.user.email, "role": tm.role} for tm in team ], "funding": [ { "donor": f.donor.name if f.donor_id else f.source, "amount": str(f.amount), "currency": f.currency, "status": f.status, } for f in funds ], } def list_tasks(user, organization, status=None, project=None, mine=False): qs = _visible_tasks(user, organization) if mine: qs = qs.filter(assignee=user) if status: qs = qs.exclude(status="archived") if status != "archived" else qs qs = qs.filter(status=status) elif not status: qs = qs.exclude(status="archived") if project: if _looks_like_uuid(project): qs = qs.filter(project_id=project) else: qs = qs.filter(project__name__icontains=project) return [ { "id": str(t.id), "title": t.title, "status": t.status, "priority": t.priority, "assignee": (t.assignee.get_full_name() or t.assignee.email) if t.assignee_id else None, "project": t.project.name if t.project_id else None, "due_date": str(t.due_date) if t.due_date else None, } for t in qs[:50] ] def _looks_like_uuid(value: str) -> bool: import uuid try: uuid.UUID(str(value)) return True except (ValueError, AttributeError): return False DISPATCH = { "list_projects": list_projects, "get_project_detail": get_project_detail, "list_tasks": list_tasks, } # Args allowed per tool, with their enum whitelist (None = free text, capped below). ALLOWED_ARGS = { "list_projects": { "status": {"planning", "active", "on_hold", "completed", "archived"}, "mine": None, }, "get_project_detail": {"project": None}, "list_tasks": { "status": {"todo", "in_progress", "review", "done", "cancelled", "archived"}, "project": None, "mine": None, }, } MAX_ARG_LEN = 200 def call_tool(user, organization, name, arguments: dict): fn = DISPATCH.get(name) if not fn: return {"error": f"Unknown tool: {name}"} allowed = ALLOWED_ARGS[name] if not isinstance(arguments, dict): return {"error": "Invalid tool arguments."} clean = {} for key, value in arguments.items(): if key not in allowed: continue # drop unknown args instead of erroring — model may hallucinate extra keys enum = allowed[key] if key == "mine": clean[key] = bool(value) continue if not isinstance(value, str): continue value = value[:MAX_ARG_LEN] if enum is not None and value not in enum: continue clean[key] = value return fn(user, organization, **clean)