"""Tool catalogue for the assistant agent. Each tool is built as a closure capturing `user` so the LLM can never pass a `user_id` argument and impersonate someone. Every data-fetching tool re-applies the platform's existing permission rules — `Document.objects.for_user(user)`, `has_role(user, ...)`, `is_admin(user)` — so the agent cannot return more than the user could see in the regular UI. Tools that surface a clickable item include a ``url`` field generated by ``django.urls.reverse``; the LLM is instructed to render those as markdown links rather than inventing URLs. """ from __future__ import annotations import logging from typing import Any from django.db.models import Q from django.urls import NoReverseMatch, reverse from django.utils import timezone from langchain_core.tools import tool from pydantic import BaseModel, Field from apps.core.dashboard.role_groups import ( ADMIN_ROLES, ALUMNI_OFFICER_ROLES, ALUMNI_ROLES, ENTREPRENEUR_ROLES, GRANTS_MANAGEMENT_ROLES, INSTRUCTOR_ROLES, LEARNER_ROLES, MEL_ROLES, RESEARCHER_ROLES, REVIEWER_ROLES, RIMS_APPLICANT_ROLES, SME_ADMIN_ROLES, has_role, is_admin, role_of, ) from .prompts import PLATFORM_HELP, PLATFORM_HELP_TOPICS logger = logging.getLogger(__name__) # ── Pydantic argument schemas ───────────────────────────────────────────── class _NoArgs(BaseModel): pass class _SearchArgs(BaseModel): query: str = Field(..., min_length=2, max_length=500, description="Natural-language search over published research documents.") k: int = Field(5, ge=1, le=10, description="Maximum results to return.") class _DocumentDetailArgs(BaseModel): public_document_id: str = Field(..., min_length=4, max_length=64, description="The human-friendly Document ID (e.g. REPO-2026-D5F1EE).") class _LimitArgs(BaseModel): limit: int = Field(5, ge=1, le=10) class _PlatformHelpArgs(BaseModel): topic: str = Field(..., description=( "One of: " + ", ".join(PLATFORM_HELP_TOPICS) + "." )) class _RepoFilterArgs(BaseModel): document_type: str | None = Field( None, description="thesis | journal_article | conference_paper | policy_brief | " "technical_report | project_report | dataset | innovation_document | publication | other") collection: str | None = Field(None, description="Collection slug (see list_repository_collections).") country: str | None = Field(None, description="Country full name, e.g. Uganda.") region: str | None = Field(None, description="Region slug, e.g. east-africa, southern-africa.") subject: str | None = Field(None, description="Subject/keyword slug, e.g. agricultural-research.") conference: str | None = Field(None, description="Exact conference name.") publisher: str | None = Field(None, description="Exact publisher / institution name.") year_from: int | None = Field(None, ge=1900, le=2100) year_to: int | None = Field(None, ge=1900, le=2100) limit: int = Field(8, ge=1, le=20) class _AuthorArgs(BaseModel): name: str = Field(..., min_length=2, max_length=120, description="Author surname or full name, e.g. 'Egeru' or 'Mwatu'.") limit: int = Field(8, ge=1, le=20) # ── Helpers ─────────────────────────────────────────────────────────────── def _safe_reverse(name: str, **kwargs) -> str: """Return a reversed URL or empty string. We never want a missing route to crash a tool — the bot just omits the link in that case.""" try: if "args" in kwargs: return reverse(name, args=kwargs["args"]) if "kwargs" in kwargs: return reverse(name, kwargs=kwargs["kwargs"]) return reverse(name) except NoReverseMatch: return "" # ── Tool factory ────────────────────────────────────────────────────────── def build_tools(user) -> list: """Return a list of LangChain tools bound to this user. Tools are reconstructed per-request so the closure captures the live user. Some tools are role-gated and simply not registered for users without the right role — the LLM never sees them in its tool catalogue. """ tools: list = [] # ============================================================ # Universal tools # ============================================================ # ---- whoami ---------------------------------------------------------- @tool("whoami", args_schema=_NoArgs) def _whoami() -> dict: """Return basic identity for the current user — name, role, group flags. Useful for personalising answers.""" return { "name": (user.get_full_name() or user.email), "email": user.email, "role": role_of(user) or "(no role)", "is_admin": is_admin(user), } tools.append(_whoami) # ---- platform_help (static) ----------------------------------------- @tool("platform_help", args_schema=_PlatformHelpArgs) def _platform_help(topic: str) -> dict: """Return curated help text for a platform topic. Use this for 'how do I…?' questions about the platform itself.""" topic_norm = topic.strip().lower().replace(" ", "_") if topic_norm not in PLATFORM_HELP: return { "error": "unknown_topic", "available_topics": list(PLATFORM_HELP_TOPICS), } return {"topic": topic_norm, "body": PLATFORM_HELP[topic_norm]} tools.append(_platform_help) # ---- list_my_unread_notifications ----------------------------------- @tool("list_my_unread_notifications", args_schema=_LimitArgs) def _list_my_unread_notifications(limit: int = 5) -> dict: """List the user's unread notifications with title, verb, time, and the in-app URL to open each. Returns total unread count plus up to `limit` most recent items.""" try: from apps.core.notifications.models import Notification except Exception: # noqa: BLE001 return {"error": "notifications_unavailable", "count": 0, "items": []} base = Notification.objects.filter(recipient=user, is_read=False) items = list( base.order_by("-created_at") .values("id", "verb", "message", "action_url", "created_at")[:limit] ) return { "count": base.count(), "items": [ { "id": n["id"], "verb": n["verb"], "title": (n["message"] or "")[:120], "created_at": n["created_at"].isoformat() if n["created_at"] else None, "url": n["action_url"] or _safe_reverse("core:notification_inbox"), } for n in items ], } tools.append(_list_my_unread_notifications) # ============================================================ # Repository # ============================================================ # ---- search_repository_documents (semantic → FTS → title fallback) -- @tool("search_repository_documents", args_schema=_SearchArgs) def _search_repository_documents(query: str, k: int = 5) -> list[dict]: """Search published research documents in the Repository the current user is allowed to see. Returns up to k items with public_document_id, title, snippet, url, and a relevance score. Prefer this for any "find me papers about X" question; cite the results.""" from apps.repository.documents.models import Document def _brief(d, snippet, score) -> dict: return { "public_document_id": d.public_document_id, "document_uid": str(d.document_uid), "title": d.title, "snippet": (snippet or "")[:280], "url": _safe_reverse("repo_documents:document_detail", kwargs={"uid": d.document_uid}), "score": score, } # 1) Semantic search over pgvector embeddings (best, once indexed). hits = [] try: from apps.repository.analytics.ai_insights import get_pgvector_store hits = get_pgvector_store().similarity_search_with_score(query, k=max(k * 3, 9)) except Exception: # noqa: BLE001 — store may be unindexed / offline logger.exception("pgvector similarity search unavailable") if hits: ids = [h[0].metadata.get("document_id") for h in hits if h[0].metadata.get("document_id")] visible = { d.id: d for d in Document.objects.for_user(user) .filter(id__in=ids) .only("id", "public_document_id", "document_uid", "title") } out, seen = [], set() for doc, score in hits: did = doc.metadata.get("document_id") if did in visible and did not in seen and len(out) < k: seen.add(did) out.append(_brief(visible[did], doc.page_content, round(float(score), 4))) if out: return out # 2) Postgres full-text search over the enriched tsvector (title + # abstract + keywords + authors + geography). Works without embeddings. try: from apps.repository.search.services import search_documents fts = list(search_documents(query=query, user=user)[:k]) except Exception: # noqa: BLE001 logger.exception("FTS search_documents failed") fts = [] if fts: return [_brief(d, d.abstract, None) for d in fts] # 3) Last resort: title substring (matches even before vectors are built). fallback = ( Document.objects.for_user(user) .filter(title__icontains=query) .order_by("-published_at", "-created_at")[:k] ) return [_brief(d, d.abstract, None) for d in fallback] tools.append(_search_repository_documents) # ---- get_document_details ------------------------------------------- @tool("get_document_details", args_schema=_DocumentDetailArgs) def _get_document_details(public_document_id: str) -> dict: """Get full metadata for a Repository document by its public ID. Only returns documents the current user is allowed to see.""" from apps.repository.documents.models import Document try: d = Document.objects.for_user(user).get( public_document_id=public_document_id.strip() ) except Document.DoesNotExist: return {"error": "not_found_or_not_accessible"} return { "public_document_id": d.public_document_id, "document_uid": str(d.document_uid), "title": d.title, "abstract": (d.abstract or "")[:1500], "document_type": d.get_document_type_display(), "status": d.get_status_display(), "year": getattr(d, "year", None), "collection": d.collection.name if d.collection_id else None, "url": _safe_reverse("repo_documents:document_detail", kwargs={"uid": d.document_uid}), } tools.append(_get_document_details) # ---- list_repository_collections (catalogue) ------------------------ @tool("list_repository_collections", args_schema=_NoArgs) def _list_repository_collections() -> list[dict]: """List Repository collections the user can browse. Returns name, slug, description, and document count. Admins additionally see internal collections.""" from apps.repository.documents.models import Collection from django.db.models import Count qs = Collection.objects.annotate(documents_count=Count("documents")) if not is_admin(user): qs = qs.exclude(visibility=Collection.Visibility.INTERNAL) out = [] for c in qs.order_by("name")[:20]: out.append({ "name": c.name, "slug": c.slug, "documents_count": c.documents_count, "visibility": c.get_visibility_display(), "url": _safe_reverse( "repo_documents:collection_detail", kwargs={"slug": c.slug} ), }) return out tools.append(_list_repository_collections) # ---- list_my_repository_submissions (any authenticated user) -------- # Filter `submitted_by=user` is inherently safe — non-submitters get []. # No role gate so repo managers, MEL officers, alumni etc. can also see # their own submissions. @tool("list_my_repository_submissions", args_schema=_LimitArgs) def _list_my_repository_submissions(limit: int = 6) -> list[dict]: """List the documents the current user has submitted to the Repository, including drafts, ordered by most recently updated. Use this to answer 'what have I submitted?' / 'where is my draft?' type questions. Returns an empty list if the user has not submitted.""" from apps.repository.documents.models import Document qs = ( Document.objects.filter(submitted_by=user) .select_related("collection") .order_by("-updated_at")[:limit] ) return [ { "public_document_id": d.public_document_id, "title": d.title, "status": d.get_status_display(), "document_type": d.get_document_type_display(), "collection": d.collection.name if d.collection_id else None, "updated_at": d.updated_at.isoformat() if d.updated_at else None, "url": _safe_reverse( "repo_documents:document_detail", kwargs={"uid": d.document_uid}, ), } for d in qs ] tools.append(_list_my_repository_submissions) # ---- filter_repository_documents (structured catalogue query) ------- @tool("filter_repository_documents", args_schema=_RepoFilterArgs) def _filter_repository_documents( document_type: str | None = None, collection: str | None = None, country: str | None = None, region: str | None = None, subject: str | None = None, conference: str | None = None, publisher: str | None = None, year_from: int | None = None, year_to: int | None = None, limit: int = 8, ) -> list[dict]: """List published Repository documents matching structured filters (any combination of document_type, collection, country, region, subject, conference, publisher, year range). Use for catalogue questions like 'theses from Uganda in 2020' or 'conference papers on maize'. Cite the results.""" from apps.repository.search.services import search_documents filters = {key: val for key, val in { "document_type": document_type, "collection": collection, "country": country, "region": region, "subject": subject, "conference": conference, "publisher": publisher, "year_from": year_from, "year_to": year_to, }.items() if val not in (None, "")} qs = search_documents(filters=filters, sort="recent", user=user).select_related("collection")[:limit] return [ { "public_document_id": d.public_document_id, "document_uid": str(d.document_uid), "title": d.title, "year": d.year, "document_type": d.get_document_type_display(), "collection": d.collection.name if d.collection_id else None, "url": _safe_reverse("repo_documents:document_detail", kwargs={"uid": d.document_uid}), } for d in qs ] tools.append(_filter_repository_documents) # ---- repository_statistics (corpus analytics) ---------------------- @tool("repository_statistics", args_schema=_NoArgs) def _repository_statistics() -> dict: """High-level statistics about the published Repository corpus: totals and top breakdowns by collection, region, country, subject, conference, and author. Use for 'how many documents', 'top subjects', 'most published authors', or 'publications by year' questions.""" from apps.repository.analytics.services import get_corpus_analytics from apps.repository.documents.models import Document c = get_corpus_analytics() return { "total_published": Document.objects.filter(status=Document.Status.PUBLISHED).count(), "total_authors": c["total_authors"], "collections": c["total_collections"], "countries_covered": c["countries_covered"], "year_range": (f"{c['year_min']}-{c['year_max']}" if c.get("year_min") else None), "busiest_year": c.get("year_peak"), "by_collection": c["by_collection"][:8], "by_region": c["by_region"], "top_countries": c["top_countries"], "top_subjects": [{"label": s["label"], "count": s["total"]} for s in c["top_subjects"][:8]], "top_conferences": c["top_conferences"][:5], "most_published_authors": [ { "name": f"{a['last_name']}, {a['first_name']}".strip(", "), "documents": a["total"], "url": _safe_reverse("repo_analytics:author_impact", kwargs={"pk": a["id"]}), } for a in c["top_authors"][:8] ], } tools.append(_repository_statistics) # ---- find_papers_by_author ----------------------------------------- @tool("find_papers_by_author", args_schema=_AuthorArgs) def _find_papers_by_author(name: str, limit: int = 8) -> dict: """Find a Repository author by name and list their published documents that this user can see, with a link to each paper and to the author's profile. Cite the papers.""" from django.db.models import Count from apps.repository.documents.models import Author, Document q = name.strip() author = ( Author.objects.filter(Q(last_name__icontains=q) | Q(first_name__icontains=q)) .annotate(n=Count("documents", filter=Q(documents__status="published"), distinct=True)) .filter(n__gt=0) .order_by("-n") .first() ) if author is None: return {"error": "no_author_found", "query": q} docs = ( Document.objects.for_user(user) .filter(authors=author) .order_by("-published_at", "-created_at")[:limit] ) return { "author": author.full_name, "author_url": _safe_reverse("repo_analytics:author_impact", kwargs={"pk": author.pk}), "papers": [ { "public_document_id": d.public_document_id, "document_uid": str(d.document_uid), "title": d.title, "year": d.year, "url": _safe_reverse("repo_documents:document_detail", kwargs={"uid": d.document_uid}), } for d in docs ], } tools.append(_find_papers_by_author) # ---- get_document_citation ----------------------------------------- @tool("get_document_citation", args_schema=_DocumentDetailArgs) def _get_document_citation(public_document_id: str) -> dict: """Build a formatted reference string + link for a Repository document by its public ID (authors, year, title, journal/conference, publisher). Only for documents the current user can see.""" from apps.repository.documents.models import Document try: d = ( Document.objects.for_user(user) .prefetch_related("authors") .get(public_document_id=public_document_id.strip()) ) except Document.DoesNotExist: return {"error": "not_found_or_not_accessible"} authors = list(d.authors.all()) who = "; ".join(a.full_name for a in authors) if authors else (d.publisher or "RUFORUM") parts = [who] if d.year: parts.append(f"({d.year}).") parts.append(f"{d.title}.") if d.journal_title: jp = d.journal_title if d.volume: jp += f", {d.volume}" + (f"({d.issue})" if d.issue else "") if d.pages: jp += f": {d.pages}" parts.append(jp + ".") elif d.conference_name: parts.append(d.conference_name + ".") if d.publisher and not d.journal_title: parts.append(d.publisher + ".") return { "public_document_id": d.public_document_id, "citation": " ".join(p for p in parts if p), "url": _safe_reverse("repo_documents:document_detail", kwargs={"uid": d.document_uid}), } tools.append(_get_document_citation) # ============================================================ # RIMS — grants # ============================================================ # ---- list_open_grant_calls (public catalogue) ----------------------- @tool("list_open_grant_calls", args_schema=_LimitArgs) def _list_open_grant_calls(limit: int = 5) -> list[dict]: """List currently published grant/scholarship/fellowship calls whose deadline has not passed. The catalogue is public — no role gating.""" try: from apps.rims.grants.models import GrantCall except Exception: # noqa: BLE001 return [] now = timezone.now() qs = ( GrantCall.objects.filter(status=GrantCall.Status.PUBLISHED, closes_at__gte=now) .order_by("closes_at")[:limit] ) return [ { "title": c.title, "slug": c.slug, "call_type": c.get_call_type_display(), "closes_at": c.closes_at.isoformat() if c.closes_at else None, "url": _safe_reverse("rims_grants:call_detail", kwargs={"slug": c.slug}), } for c in qs ] tools.append(_list_open_grant_calls) # ---- count_my_open_applications ------------------------------------- if has_role(user, RIMS_APPLICANT_ROLES) or is_admin(user): @tool("count_my_open_applications", args_schema=_NoArgs) def _count_my_open_applications() -> dict: """Count the current user's grant applications that are still in an open state (draft, submitted, under review, or shortlisted).""" try: from apps.rims.grants.models import Application except Exception: # noqa: BLE001 return {"error": "rims_unavailable"} open_states = [ Application.Status.DRAFT, Application.Status.SUBMITTED, Application.Status.UNDER_REVIEW, Application.Status.SHORTLISTED, ] n = Application.objects.filter( applicant=user, status__in=open_states ).count() return {"open_applications": n} tools.append(_count_my_open_applications) # ---- list_my_applications ------------------------------------------- if has_role(user, RIMS_APPLICANT_ROLES) or is_admin(user): @tool("list_my_applications", args_schema=_LimitArgs) def _list_my_applications(limit: int = 5) -> list[dict]: """List the current user's grant/scholarship applications with their call title, status, and submission timestamp. Excludes withdrawn / rejected / expired drafts.""" try: from apps.rims.grants.models import Application except Exception: # noqa: BLE001 return [] terminal = [ Application.Status.WITHDRAWN, Application.Status.WITHDRAWN_NO_RESPONSE, Application.Status.REJECTED, Application.Status.REJECTED_INELIGIBLE, Application.Status.EXPIRED_DRAFT, ] qs = ( Application.objects.filter(applicant=user) .exclude(status__in=terminal) .select_related("call") .order_by("-id")[:limit] ) return [ { "id": a.pk, "call_title": a.call.title, "call_slug": a.call.slug, "status": a.get_status_display(), "submitted_at": a.submitted_at.isoformat() if a.submitted_at else None, "url": _safe_reverse("rims_grants:application_detail", args=[a.pk]), } for a in qs ] tools.append(_list_my_applications) # ---- list_my_pending_reviews ---------------------------------------- if has_role(user, REVIEWER_ROLES) or is_admin(user): @tool("list_my_pending_reviews", args_schema=_LimitArgs) def _list_my_pending_reviews(limit: int = 5) -> list[dict]: """List grant-application reviews assigned to the current user that have not yet been submitted. For blind-review calls the applicant identity is hidden (anonymised code only).""" try: from apps.rims.grants.models import Review except Exception: # noqa: BLE001 return [] qs = ( Review.objects.filter(reviewer=user, submitted_at__isnull=True) .select_related("application", "application__call") .order_by("due_at")[:limit] ) out = [] for r in qs: app = r.application call = app.call ref = ( app.anonymized_code or f"App #{app.pk}" if call.blind_review else f"App #{app.pk}" ) out.append({ "review_id": r.pk, "application_ref": ref, "call_title": call.title, "due_at": r.due_at.isoformat() if r.due_at else None, "url": _safe_reverse("rims_grants:review_submit", args=[r.pk]), }) return out tools.append(_list_my_pending_reviews) # REP course tools (list_active_courses / list_my_courses) were removed # with the REP LMS. Learning lives in Moodle now; a Moodle-backed catalogue # tool can be added later. # ============================================================ # Alumni # ============================================================ # ---- list_upcoming_alumni_events ------------------------------------ if (has_role(user, ALUMNI_ROLES) or has_role(user, ALUMNI_OFFICER_ROLES) or is_admin(user)): @tool("list_upcoming_alumni_events", args_schema=_LimitArgs) def _list_upcoming_alumni_events(limit: int = 5) -> list[dict]: """List upcoming alumni events the current user can attend. Includes public + alumni-only events; group-only events are excluded for safety (they require explicit group membership).""" try: from apps.alumni.engagement.models import AlumniEvent, EventVisibility except Exception: # noqa: BLE001 return [] allowed = [EventVisibility.PUBLIC, EventVisibility.ALUMNI_ONLY] qs = ( AlumniEvent.objects.filter(start_at__gte=timezone.now()) .filter(visibility__in=allowed) .order_by("start_at")[:limit] ) return [ { "title": e.title, "slug": e.slug, "start_at": e.start_at.isoformat() if e.start_at else None, "location": e.location or "", "url": _safe_reverse( "alumni_engagement:event_detail", kwargs={"slug": e.slug} ), } for e in qs ] tools.append(_list_upcoming_alumni_events) # ---- list_my_mentorships -------------------------------------------- if (has_role(user, ALUMNI_ROLES) or has_role(user, ALUMNI_OFFICER_ROLES) or is_admin(user)): @tool("list_my_mentorships", args_schema=_LimitArgs) def _list_my_mentorships(limit: int = 5) -> list[dict]: """List the user's mentorship pairings — as mentor (their alumni profile is the mentor) or as mentee. Returns counterpart name, status, topic, and started/ended timestamps.""" try: from apps.alumni.engagement.models import MentorshipPairing except Exception: # noqa: BLE001 return [] qs = ( MentorshipPairing.objects.filter( Q(mentor__user=user) | Q(mentee=user) ) .select_related("mentor__user", "mentee") .order_by("-created_at")[:limit] ) out: list[dict] = [] for p in qs: if p.mentor.user_id == user.pk: role = "mentor" counterpart = p.mentee.get_full_name() or p.mentee.email else: role = "mentee" counterpart = ( p.mentor.user.get_full_name() or p.mentor.user.email ) out.append({ "role": role, "counterpart": counterpart, "status": p.get_status_display(), "topic": p.topic or "", "started_at": p.started_at.isoformat() if p.started_at else None, "url": _safe_reverse("alumni_engagement:mentorship_list"), }) return out tools.append(_list_my_mentorships) # ============================================================ # SME-Hub # ============================================================ # ---- list_my_smehub_businesses -------------------------------------- if has_role(user, ENTREPRENEUR_ROLES) or is_admin(user): @tool("list_my_smehub_businesses", args_schema=_LimitArgs) def _list_my_smehub_businesses(limit: int = 5) -> list[dict]: """List the businesses owned by the current entrepreneur. Returns name, sector, stage, country, and verification status.""" try: from apps.smehub.onboarding.models import Business except Exception: # noqa: BLE001 return [] qs = ( Business.objects.filter(entrepreneur__user=user) .order_by("-created_at")[:limit] ) return [ { "business_id": b.business_id, "name": b.name, "sector": b.sector, "stage": b.get_business_stage_display(), "country": b.country or "", "verification_status": b.get_verification_status_display(), "url": _safe_reverse( "smehub_onboarding:business_detail", kwargs={"business_id": b.pk}, ), } for b in qs ] tools.append(_list_my_smehub_businesses) # ---- list_my_funding_applications ----------------------------------- if has_role(user, ENTREPRENEUR_ROLES) or is_admin(user): @tool("list_my_funding_applications", args_schema=_LimitArgs) def _list_my_funding_applications(limit: int = 5) -> list[dict]: """List the current entrepreneur's funding-call applications. Returns the call title, business name, requested amount, and current status.""" try: from apps.smehub.investment.models import FundingApplication except Exception: # noqa: BLE001 return [] qs = ( FundingApplication.objects.filter(entrepreneur__user=user) .select_related("funding_call", "business") .order_by("-id")[:limit] ) out = [] for a in qs: out.append({ "application_id": a.application_id, "call_title": a.funding_call.title, "business": a.business.name, "amount_requested": ( str(a.funding_request_amount) if a.funding_request_amount is not None else None ), "status": a.get_status_display(), "url": _safe_reverse( "smehub_investment:funding_application_detail", kwargs={"pk": a.pk}, ), }) return out tools.append(_list_my_funding_applications) # ---- count_pending_verifications (admin) ---------------------------- if has_role(user, SME_ADMIN_ROLES) or is_admin(user): @tool("count_pending_verifications", args_schema=_NoArgs) def _count_pending_verifications() -> dict: """Count SME-Hub records awaiting admin verification, broken down by category (entrepreneurs, businesses, investors). Use for the admin's daily ops digest.""" try: from apps.smehub.investment.models import Investor from apps.smehub.onboarding.models import ( Business, EntrepreneurProfile, InvestorProfile, MentorProfile, ) except Exception: # noqa: BLE001 return {"error": "smehub_unavailable"} counts = { "entrepreneurs": EntrepreneurProfile.objects.filter( verification_status=EntrepreneurProfile.Status.PENDING_VERIFICATION ).count(), "businesses": Business.objects.filter( verification_status=Business.Status.PENDING_ADMIN_REVIEW ).count(), } try: # Investor uses Status.PENDING ("pending"), not the shared base. counts["investors"] = Investor.objects.filter( verification_status=Investor.Status.PENDING ).count() except Exception: # noqa: BLE001 pass for label, model in (("mentor_profiles", MentorProfile), ("investor_profiles", InvestorProfile)): try: counts[label] = model.objects.filter( verification_status="pending_verification" ).count() except Exception: # noqa: BLE001 pass counts["total"] = sum(v for v in counts.values() if isinstance(v, int)) counts["url"] = _safe_reverse("smehub_onboarding:business_queue") return counts tools.append(_count_pending_verifications) # ============================================================ # MEL # ============================================================ # ---- list_at_risk_activities (officer / management) ----------------- if (has_role(user, MEL_ROLES) or has_role(user, GRANTS_MANAGEMENT_ROLES) or has_role(user, ADMIN_ROLES) or is_admin(user)): @tool("list_at_risk_activities", args_schema=_LimitArgs) def _list_at_risk_activities(limit: int = 5) -> list[dict]: """List MEL tracked activities that are off-track: scheduled_end in the past and status != completed, or deviation_flag=True. Returns name, scheduled_end, status, and responsible person.""" try: from apps.mel.tracking.models import Activity, ActivityStatus except Exception: # noqa: BLE001 return [] today = timezone.now().date() qs = ( Activity.objects.filter( Q(scheduled_end__lt=today) | Q(deviation_flag=True) ) .exclude(status=ActivityStatus.COMPLETED) .select_related("responsible", "logframe_row") .order_by("scheduled_end")[:limit] ) out = [] for a in qs: resp = "" if a.responsible_id: resp = (a.responsible.get_full_name() or a.responsible.email or "") out.append({ "name": a.name, "scheduled_end": a.scheduled_end.isoformat() if a.scheduled_end else None, "status": a.get_status_display(), "deviation_flag": a.deviation_flag, "responsible": resp, }) return out tools.append(_list_at_risk_activities) return tools # ── Tool-result helpers (used by view + citations) ──────────────────────── def extract_surfaced_object_ids(tool_name: str, observation: Any) -> list[str]: """Pull out citable identifiers (public_document_id) from a tool result. Note: only repository-document tools feed the citations chip system. Other tools surface their results as inline markdown links via the ``url`` field — no chip hydration needed. """ out: list[str] = [] if tool_name in {"search_repository_documents", "get_document_details", "list_my_repository_submissions", "filter_repository_documents", "find_papers_by_author", "get_document_citation"}: if isinstance(observation, dict): pid = observation.get("public_document_id") if pid: out.append(pid) # find_papers_by_author nests citable docs under "papers". for item in observation.get("papers", []) or []: if isinstance(item, dict) and item.get("public_document_id"): out.append(item["public_document_id"]) elif isinstance(observation, list): for item in observation: if isinstance(item, dict) and item.get("public_document_id"): out.append(item["public_document_id"]) return out