From 0bc618186bc736968047f7538b20de40af34771a Mon Sep 17 00:00:00 2001 From: Paul Lizer Date: Wed, 7 Oct 2026 23:19:59 -0400 Subject: [PATCH 1/3] Improve the Control Center dashboard and add Chat with this dashboard Dashboard: - Move the dashboard aggregation into functions_control_center_dashboard.py and make the summary and insights routes thin wrappers with split activity/token caches. - Show names instead of IDs in token and activity rankings, with deleted entities labeled. - Reorganize the V2 dashboard into Directory, Sign-ins, Conversations and documents, Token usage and Most active, with plain-language definitions for DAU, WAU and MAU. - Put the token-only filters inside Token usage, next to the figures they change. - Split sign-ins from conversations, drill into the clicked stacked segment, keep token filters on drill-through, and show links only for sections the viewer can open. Chat with this dashboard: - Add a read-only, global-only Control Center action that answers only for Control Center dashboard viewers and refuses runs without a signed-in session. - Gate action discovery and manifest resolution on the caller's roles. - Add GET /api/v2/control-center/dashboard/chat-readiness and a requirements checklist that links administrators to each missing setting; when ready, open a new orchestrated chat with the prompt ready to send. - Retire the broken /api/admin/control-center/activity-trends/chat endpoint and the classic page's unreachable chat modal. Tests and docs for the dashboard, action, readiness and UI; version 0.261.294. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- application/single_app/config.py | 2 +- .../single_app/functions_action_catalog.py | 53 +- .../single_app/functions_authentication.py | 13 + .../functions_control_center_dashboard.py | 1427 +++++++++++++++++ ...functions_control_center_dashboard_chat.py | 210 +++ .../single_app/functions_governance.py | 12 + .../functions_orchestration_actions.py | 2 + .../functions_orchestration_context.py | 10 +- .../functions_orchestration_execution.py | 1 + .../functions_orchestration_plan_editing.py | 1 + .../single_app/json_schema_validation.py | 2 + .../route_backend_control_center.py | 968 ++--------- .../single_app/route_backend_orchestration.py | 3 + .../single_app/route_backend_plugins.py | 14 + .../control_center_plugin.py | 402 +++++ .../static/js/admin/admin_governance.js | 2 + .../single_app/static/js/control-center.js | 118 -- .../static/js/plugin_modal_stepper.js | 5 +- .../static/js/workspace/view-utils.js | 1 + .../schemas/control_center.definition.json | 7 + .../single_app/templates/control_center.html | 82 - application/v2_ui/src/App.tsx | 14 +- .../v2_ui/src/components/chat/Composer.tsx | 36 + .../controlCenter/DashboardChat.tsx | 186 +++ .../controlCenter/DashboardSection.tsx | 1091 ++++++++----- .../v2_ui/src/lib/composerDraftHandoff.ts | 59 + application/v2_ui/src/lib/governance.ts | 1 + .../v2_ui/src/lib/workspaceActionRegistry.ts | 4 + .../src/pages/workspace/ActionEditorPage.tsx | 11 + docs/_data/app_surface.yml | 6 +- docs/admin/agents-actions.md | 8 + docs/admin/orchestration.md | 6 + .../features/CONTROL_CENTER_ACTION.md | 126 ++ .../explanation/features/V2_CONTROL_CENTER.md | 18 +- ..._CONTROL_CENTER_DASHBOARD_USABILITY_FIX.md | 77 + docs/guides/v2-control-center.md | 53 +- docs/reference/actions/control-center.md | 64 + docs/reference/actions/index.md | 1 + docs/reference/logging-tags.md | 1 + .../test_route_blueprint_policy_inventory.py | 1 + .../test_control_center_token_filters.py | 23 +- .../test_orchestration_action_catalog.py | 76 +- functional_tests/test_v2_api_security.py | 1 + .../test_v2_control_center_action.py | 680 ++++++++ ...ntrol_center_cosmos_query_compatibility.py | 26 +- .../test_v2_control_center_dashboard.py | 625 +++++--- .../test_workspace_authoring_backend.py | 4 +- ui_tests/test_v2_control_center_dashboard.py | 444 +++-- 48 files changed, 5146 insertions(+), 1831 deletions(-) create mode 100644 application/single_app/functions_control_center_dashboard.py create mode 100644 application/single_app/functions_control_center_dashboard_chat.py create mode 100644 application/single_app/semantic_kernel_plugins/control_center_plugin.py create mode 100644 application/single_app/static/json/schemas/control_center.definition.json create mode 100644 application/v2_ui/src/components/controlCenter/DashboardChat.tsx create mode 100644 application/v2_ui/src/lib/composerDraftHandoff.ts create mode 100644 docs/explanation/features/CONTROL_CENTER_ACTION.md create mode 100644 docs/explanation/fixes/V2_CONTROL_CENTER_DASHBOARD_USABILITY_FIX.md create mode 100644 docs/reference/actions/control-center.md create mode 100644 functional_tests/test_v2_control_center_action.py diff --git a/application/single_app/config.py b/application/single_app/config.py index 4206520c9..05464869a 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -101,7 +101,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.261.293" +VERSION = "0.261.294" IS_DEVELOPMENT = is_development_env_enabled() # Opt-out for deployments where App Service Easy Auth is active but the platform diff --git a/application/single_app/functions_action_catalog.py b/application/single_app/functions_action_catalog.py index ee58d269a..e8b84614e 100644 --- a/application/single_app/functions_action_catalog.py +++ b/application/single_app/functions_action_catalog.py @@ -16,6 +16,7 @@ from functions_action_manifest import bind_action_origin, is_retired_mcp_stdio from functions_agent_delegation import AGENT_PLUGIN_TYPE, _identifier +from functions_control_center_dashboard import CONTROL_CENTER_ACTION_TYPE _SCOPES = frozenset({"personal", "group", "global"}) @@ -158,6 +159,42 @@ def _container(scope_type): return getattr(import_module("config"), f"cosmos_{scope_type}_actions_container") +def _is_control_center_action(action): + action_type = action.get("type") if isinstance(action, dict) else None + return isinstance(action_type, str) and action_type.strip().lower() == CONTROL_CENTER_ACTION_TYPE + + +def _request_roles(): + """The signed-in request's app roles, or none outside a request. + + Flask is imported here, with the rest of this module's runtime dependencies, so the + planner projection stays importable without the application. + """ + from flask import has_request_context, session + + if not has_request_context(): + return [] + user = session.get("user") or {} + roles = user.get("roles") if isinstance(user, dict) else None + return list(roles) if isinstance(roles, (list, tuple, set, frozenset)) else [] + + +def _control_center_access(settings, user_roles=None): + """Whether the caller may read Control Center insights; no identity means no access. + + ``user_roles`` should come from the authenticated identity a caller captured. Without it + the current request's session is read, and outside a request access is refused. + """ + roles = _request_roles() if user_roles is None else user_roles + if not isinstance(roles, (list, tuple, set, frozenset)) or not roles: + return False + authentication = import_module("functions_authentication") + capabilities = authentication.get_control_center_capabilities( + {"roles": [str(role) for role in roles]}, settings, + ) + return bool(capabilities.get("can_view_dashboard")) + + def _stored_actions(scope_type, scope_id): if scope_type == "global": return _container(scope_type).query_items( @@ -256,13 +293,15 @@ def _descriptor(action, scope_type, scope_id, scope_label): } -def build_accessible_action_catalog(user_id, *, settings=None, user_groups=None): +def build_accessible_action_catalog(user_id, *, settings=None, user_groups=None, user_roles=None): """Return safe metadata for enabled, governed actions belonging to this actor. ``user_id`` must be captured from authenticated server context, never from plan arguments. ``user_groups`` may contain group IDs or group records and only narrows current membership. Global actions retain their own scope and are available in global mode, or when global/workspace merging is enabled. + ``user_roles`` are the caller's authenticated app roles; a Control Center action is + listed only for a caller who may view the Control Center dashboard. """ user_id = _require_actor(user_id) settings = _resolve_settings(settings) @@ -281,6 +320,7 @@ def build_accessible_action_catalog(user_id, *, settings=None, user_groups=None) group_id = group["id"] scopes.append(("group", group_id, group.get("name") or "Group")) + control_center_allowed = None catalog = {} for scope_type, scope_id, scope_label in scopes: actions = [ @@ -293,6 +333,11 @@ def build_accessible_action_catalog(user_id, *, settings=None, user_groups=None) feature = "governance_user_actions" if scope_type == "personal" else "governance_group_actions" actions = governance.filter_actions_by_action_type_access(user_id, actions, feature, scope_type) for action in actions: + if _is_control_center_action(action): + if control_center_allowed is None: + control_center_allowed = _control_center_access(settings, user_roles) + if scope_type != "global" or not control_center_allowed: + continue descriptor = _descriptor(action, scope_type, scope_id, scope_label) catalog.setdefault(descriptor["action_ref"], descriptor) return list(catalog.values()) @@ -325,7 +370,7 @@ def build_action_planner_projection(actions): return projection -def resolve_action_manifest(user_id, action_ref, *, settings=None, user_groups=None): +def resolve_action_manifest(user_id, action_ref, *, settings=None, user_groups=None, user_roles=None): """Reauthorize an exact stored ID and return its secret-reference manifest. No lookup by name, plugin initialization, or workspace-identity hydration is @@ -366,6 +411,10 @@ def resolve_action_manifest(user_id, action_ref, *, settings=None, user_groups=N raise LookupError(_UNAVAILABLE) if not _eligible_record(action, scope_type, scope_id): raise PermissionError(_UNAVAILABLE) + if _is_control_center_action(action) and ( + scope_type != "global" or not _control_center_access(settings, user_roles) + ): + raise PermissionError(_UNAVAILABLE) governance = import_module("functions_governance") try: diff --git a/application/single_app/functions_authentication.py b/application/single_app/functions_authentication.py index 1d466e4a4..295ad877d 100644 --- a/application/single_app/functions_authentication.py +++ b/application/single_app/functions_authentication.py @@ -1000,6 +1000,19 @@ def get_control_center_capabilities(user=None, settings=None): } +def get_request_control_center_capabilities(settings=None): + """Return the signed-in request's Control Center capabilities, failing closed outside one. + + Agent and orchestration runs re-enter a request context that carries the authenticated + user and roles, so tools invoked from chat are answered here too. A scheduled run has no + signed-in session, so it receives no Control Center access. + """ + if not has_request_context(): + return get_control_center_capabilities({}, {}) + user = session.get('user') or {} + return get_control_center_capabilities(user, get_settings() if settings is None else settings) + + def control_center_required(access_level='admin'): """ Unified Control Center access control decorator. diff --git a/application/single_app/functions_control_center_dashboard.py b/application/single_app/functions_control_center_dashboard.py new file mode 100644 index 000000000..54331bba2 --- /dev/null +++ b/application/single_app/functions_control_center_dashboard.py @@ -0,0 +1,1427 @@ +# functions_control_center_dashboard.py +"""Aggregate V2 Control Center dashboard metrics from recorded activity. + +The dashboard routes and the Control Center action both read through these functions, so a +figure on the dashboard and the same figure reported in chat come from the same queries. +Storage handles arrive in a ``DashboardStores`` bundle and logging through a callback, so this +module never imports the Azure bootstrap and tests can drive it with fake containers. + +The azure-cosmos Python SDK cannot run a cross-partition GROUP BY or COUNT over DISTINCT +values. Grouped figures are therefore totalled here from narrow streamed projections, and +distinct values are counted after ``SELECT DISTINCT VALUE``. +""" + +import json +import logging +import threading +import time +from collections import Counter, defaultdict +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Tuple + + +CONTROL_CENTER_DASHBOARD_CACHE_TTL_SECONDS = 90 +CONTROL_CENTER_DASHBOARD_CACHE_MAX_ENTRIES = 128 +CONTROL_CENTER_DASHBOARD_NAME_CACHE_MAX_ENTRIES = 1024 +CONTROL_CENTER_DASHBOARD_RANKING_LIMIT = 10 +CONTROL_CENTER_DASHBOARD_MAX_RANGE_DAYS = 366 +CONTROL_CENTER_DASHBOARD_PRESET_DAYS = (7, 30, 90) +CONTROL_CENTER_DASHBOARD_RANKED_FIELDS = ( + ("users", "user_id"), + ("groups", "group_id"), + ("public_workspaces", "public_workspace_id"), +) +DASHBOARD_INVALID_RANGE_ERROR = ( + "Invalid dashboard date range. Use 7, 30, or 90 days or a valid custom range of up to 366 days." +) +DASHBOARD_TOKEN_TYPES = ("chat", "embedding", "web_search") +DASHBOARD_WORKSPACE_TYPES = ("personal", "group", "public") +DASHBOARD_WEEKDAYS = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday") +DASHBOARD_DAILY_ACTIVITY_FIELDS = ( + "sign_ins", + "conversations_created", + "uploads_personal", + "uploads_group", + "uploads_public", +) +DASHBOARD_ENTITY_KINDS = ("users", "groups", "public_workspaces") +DASHBOARD_ENTITY_MISSING_NAMES = { + "users": "Unknown user", + "groups": "Deleted group", + "public_workspaces": "Deleted public workspace", +} +DASHBOARD_ENTITY_UNNAMED = { + "users": "Unnamed user", + "groups": "Unnamed group", + "public_workspaces": "Unnamed public workspace", +} +DASHBOARD_ENTITY_SEARCH_MIN_LENGTH = 2 +DASHBOARD_ENTITY_SEARCH_MAX_LENGTH = 100 +DASHBOARD_ENTITY_SEARCH_MAX_RESULTS = 10 +DASHBOARD_REPORT_MAX_ROWS = 50 +DASHBOARD_TOKEN_GROUPINGS = ("day", "model", "token_type", "user", "group", "public_workspace") +DASHBOARD_LOGIN_HEATMAP_CONVENTION = "Monday=0 through Sunday=6" + +# The read-only Control Center action type. Declared here, with no runtime dependencies, so +# validation, governance and action discovery can name it without loading Semantic Kernel. +CONTROL_CENTER_ACTION_TYPE = "control_center" +CONTROL_CENTER_ACTION_DEFAULT_ENDPOINT = "control_center://internal" +# What chat orchestration reads when it decides whether this action can answer a question. +CONTROL_CENTER_ACTION_DEFAULT_DESCRIPTION = ( + "Read-only SimpleChat usage insights from the Control Center dashboard: sign-ins, daily, " + "weekly and monthly active users, conversations, document uploads, token usage by day, model " + "and usage type, the most active users, groups and public workspaces, and sign-in times. " + "Available only to Control Center administrators and dashboard readers." +) + +# A record counts toward a period when either recorded time falls inside it. Older records +# carry only created_at, newer ones carry timestamp, and many carry both. +RECORDED_IN_WINDOW = """( + (c.timestamp >= @start_date AND c.timestamp <= @end_date) + OR (c.created_at >= @start_date AND c.created_at <= @end_date) + )""" + +_control_center_dashboard_cache = {} +_control_center_entity_name_cache = {} +_cache_lock = threading.Lock() + + +@dataclass(frozen=True) +class DashboardStores: + """The containers the dashboard reads, supplied by the caller that owns configuration.""" + + activity_logs: Any + user_settings: Any = None + groups: Any = None + public_workspaces: Any = None + user_documents: Any = None + group_documents: Any = None + public_documents: Any = None + approvals: Any = None + + +@dataclass(frozen=True) +class DashboardPeriod: + """A range of whole UTC calendar days and the equally long range just before it.""" + + start: datetime + end: datetime + previous_start: datetime + previous_end: datetime + days: int + + @property + def start_day(self) -> str: + return self.start.date().isoformat() + + @property + def end_day(self) -> str: + return self.end.date().isoformat() + + def cache_key(self) -> Tuple[str, str]: + return (self.start.isoformat(), self.end.isoformat()) + + def describe(self) -> Dict[str, Any]: + return { + "start_date": self.start_day, + "end_date": self.end_day, + "days": self.days, + "timezone": "UTC", + } + + def describe_previous(self) -> Dict[str, Any]: + return { + "start_date": self.previous_start.date().isoformat(), + "end_date": self.previous_end.date().isoformat(), + "days": self.days, + "timezone": "UTC", + } + + def dates(self) -> List[str]: + first_day = self.start.date() + return [(first_day + timedelta(days=offset)).isoformat() for offset in range(self.days)] + + +# -------------------------------------------------------------------------------------- +# Periods +# -------------------------------------------------------------------------------------- + +def build_dashboard_period(start_day: date, end_day: date) -> DashboardPeriod: + """Return the UTC range covering whole calendar days, with its comparison range.""" + if end_day < start_day: + raise ValueError("end_date must be on or after start_date.") + days = (end_day - start_day).days + 1 + if days > CONTROL_CENTER_DASHBOARD_MAX_RANGE_DAYS: + raise ValueError(f"Ranges may not exceed {CONTROL_CENTER_DASHBOARD_MAX_RANGE_DAYS} days.") + start = datetime.combine(start_day, datetime.min.time()) + end = datetime.combine(end_day, datetime.max.time()) + previous_end = start - timedelta(microseconds=1) + previous_start = previous_end - timedelta(days=days) + timedelta(microseconds=1) + return DashboardPeriod(start, end, previous_start, previous_end, days) + + +def _parse_day(value: Any) -> date: + try: + return datetime.strptime(str(value).strip(), "%Y-%m-%d").date() + except (TypeError, ValueError) as ex: + raise ValueError("Dates must use YYYY-MM-DD format.") from ex + + +def parse_dashboard_period(args: Mapping[str, Any], *, today: Optional[date] = None) -> DashboardPeriod: + """Resolve the dashboard's 7, 30 or 90 day preset, or a custom range of up to 366 days.""" + custom_start = args.get("start_date") + custom_end = args.get("end_date") + if custom_start or custom_end: + if not custom_start or not custom_end: + raise ValueError("Both start_date and end_date are required.") + return build_dashboard_period(_parse_day(custom_start), _parse_day(custom_end)) + try: + days = int(args.get("days", 30)) + except (TypeError, ValueError) as ex: + raise ValueError("days must be 7, 30, or 90.") from ex + if days not in CONTROL_CENTER_DASHBOARD_PRESET_DAYS: + raise ValueError("days must be 7, 30, or 90.") + end_day = today or datetime.utcnow().date() + return build_dashboard_period(end_day - timedelta(days=days - 1), end_day) + + +def resolve_requested_period( + start_date: Any = "", + end_date: Any = "", + days: Any = None, + *, + today: Optional[date] = None, +) -> DashboardPeriod: + """Resolve explicit dates, or a number of whole days ending today (UTC).""" + start_text = str(start_date or "").strip() + end_text = str(end_date or "").strip() + if start_text or end_text: + if not start_text or not end_text: + raise ValueError("Provide both start_date and end_date, or neither.") + return build_dashboard_period(_parse_day(start_text), _parse_day(end_text)) + if days is None or (isinstance(days, str) and not days.strip()): + days = 30 + if isinstance(days, bool): + raise ValueError(f"days must be a whole number from 1 to {CONTROL_CENTER_DASHBOARD_MAX_RANGE_DAYS}.") + try: + day_count = int(str(days).strip()) + except (TypeError, ValueError) as ex: + raise ValueError( + f"days must be a whole number from 1 to {CONTROL_CENTER_DASHBOARD_MAX_RANGE_DAYS}." + ) from ex + if not 1 <= day_count <= CONTROL_CENTER_DASHBOARD_MAX_RANGE_DAYS: + raise ValueError(f"days must be a whole number from 1 to {CONTROL_CENTER_DASHBOARD_MAX_RANGE_DAYS}.") + end_day = today or datetime.utcnow().date() + return build_dashboard_period(end_day - timedelta(days=day_count - 1), end_day) + + +# -------------------------------------------------------------------------------------- +# Cache +# -------------------------------------------------------------------------------------- + +def _cache_get(cache, key, force_refresh=False, now=None): + if force_refresh: + return None + current_time = time.monotonic() if now is None else now + with _cache_lock: + entry = cache.get(key) + if not entry: + return None + expires_at, payload = entry + if expires_at <= current_time: + cache.pop(key, None) + return None + return payload + + +def _cache_set(cache, key, payload, max_entries, now=None): + current_time = time.monotonic() if now is None else now + with _cache_lock: + expired_keys = [ + cache_key + for cache_key, (expires_at, _) in cache.items() + if expires_at <= current_time + ] + for cache_key in expired_keys: + cache.pop(cache_key, None) + if key not in cache and len(cache) >= max_entries: + cache.pop(next(iter(cache))) + cache[key] = (current_time + CONTROL_CENTER_DASHBOARD_CACHE_TTL_SECONDS, payload) + + +def dashboard_cache_get(key, force_refresh=False, now=None): + """Return an unexpired dashboard cache entry, if one is available.""" + return _cache_get(_control_center_dashboard_cache, key, force_refresh=force_refresh, now=now) + + +def dashboard_cache_set(key, payload, now=None): + """Store a short-lived dashboard result and bound cache growth.""" + _cache_set( + _control_center_dashboard_cache, + key, + payload, + CONTROL_CENTER_DASHBOARD_CACHE_MAX_ENTRIES, + now=now, + ) + + +def _filters_cache_key(filters: Optional[Mapping[str, Any]]) -> str: + return json.dumps( + {key: value for key, value in (filters or {}).items() if value}, + sort_keys=True, + ) + + +def _utc_now_text() -> str: + return datetime.utcnow().isoformat() + "Z" + + +# -------------------------------------------------------------------------------------- +# Metric formatting +# -------------------------------------------------------------------------------------- + +def dashboard_metric(current_value, previous_value=None): + """Format a current-period value with a comparable prior-period delta.""" + current_value = int(current_value or 0) + if previous_value is None: + return {"value": current_value, "delta": None, "percent_change": None} + previous_value = int(previous_value or 0) + delta = current_value - previous_value + percent_change = ( + round((delta / previous_value) * 100, 1) + if previous_value + else (0.0 if delta == 0 else None) + ) + return { + "value": current_value, + "delta": delta, + "previous": previous_value, + "percent_change": percent_change, + } + + +def dashboard_status_counts(total, grouped_statuses, unknown_status="active"): + """Normalize stored status counts while preserving each workspace type's default.""" + counts = { + "active": 0, + "locked": 0, + "upload_disabled": 0, + "inactive": 0, + } + for row in grouped_statuses: + raw_status = row.get("status") + status = str(raw_status or "active").strip().lower() + count = int(row.get("count") or 0) + if status in {"locked", "upload_disabled", "inactive"}: + counts[status] += count + elif status != "active" and raw_status and unknown_status == "inactive": + counts["inactive"] += count + counts["active"] = max( + int(total or 0) - counts["locked"] - counts["upload_disabled"] - counts["inactive"], + 0, + ) + return counts + + +# -------------------------------------------------------------------------------------- +# Token filters +# -------------------------------------------------------------------------------------- + +def normalize_token_filter_value(value): + """Normalize optional token filter values from query params or request JSON.""" + if value is None: + return None + + normalized = str(value).strip() + if not normalized or normalized.lower() == "all": + return None + + return normalized + + +def extract_token_filters(source): + """Extract supported token filters from a query string or JSON payload.""" + if not source: + return {} + + token_source = source + nested_filters = source.get("token_filters") if hasattr(source, "get") else None + if isinstance(nested_filters, dict): + token_source = nested_filters + + token_filters = { + "user_id": normalize_token_filter_value(token_source.get("user_id")), + "workspace_type": normalize_token_filter_value(token_source.get("workspace_type")), + "group_id": normalize_token_filter_value(token_source.get("group_id")), + "public_workspace_id": normalize_token_filter_value(token_source.get("public_workspace_id")), + "model": normalize_token_filter_value(token_source.get("model")), + "token_type": normalize_token_filter_value(token_source.get("token_type")), + } + + if token_filters["workspace_type"] not in {None, "personal", "group", "public"}: + token_filters["workspace_type"] = None + + if token_filters["token_type"] not in {None, "embedding", "chat", "web_search"}: + token_filters["token_type"] = None + + if token_filters["group_id"] and not token_filters["workspace_type"]: + token_filters["workspace_type"] = "group" + + if token_filters["public_workspace_id"] and not token_filters["workspace_type"]: + token_filters["workspace_type"] = "public" + + return token_filters + + +def append_token_usage_filters(query_conditions, parameters, token_filters): + """Append optional token usage filters to the Cosmos query state.""" + if not token_filters: + return + + user_id = token_filters.get("user_id") + if user_id: + query_conditions.append("c.user_id = @token_user_id") + parameters.append({"name": "@token_user_id", "value": user_id}) + + workspace_type = token_filters.get("workspace_type") + if workspace_type: + query_conditions.append("c.workspace_type = @token_workspace_type") + parameters.append({"name": "@token_workspace_type", "value": workspace_type}) + + group_id = token_filters.get("group_id") + if group_id: + query_conditions.append("c.workspace_context.group_id = @token_group_id") + parameters.append({"name": "@token_group_id", "value": group_id}) + + public_workspace_id = token_filters.get("public_workspace_id") + if public_workspace_id: + query_conditions.append("c.workspace_context.public_workspace_id = @token_public_workspace_id") + parameters.append({"name": "@token_public_workspace_id", "value": public_workspace_id}) + + model = token_filters.get("model") + if model: + query_conditions.append("c.usage.model = @token_model") + parameters.append({"name": "@token_model", "value": model}) + + token_type = token_filters.get("token_type") + if token_type: + query_conditions.append("c.token_type = @token_type") + parameters.append({"name": "@token_type", "value": token_type}) + + +def build_token_usage_query_context(start_date, end_date, token_filters=None): + """Build shared WHERE-clause state for token usage queries.""" + parameters = [ + {"name": "@start_date", "value": start_date.isoformat()}, + {"name": "@end_date", "value": end_date.isoformat()} + ] + query_conditions = [ + "c.activity_type = 'token_usage'", + "((c.timestamp >= @start_date AND c.timestamp <= @end_date) OR (c.created_at >= @start_date AND c.created_at <= @end_date))" + ] + + append_token_usage_filters(query_conditions, parameters, token_filters or {}) + return " AND ".join(query_conditions), parameters + + +# -------------------------------------------------------------------------------------- +# Counts +# -------------------------------------------------------------------------------------- + +def _period_parameters(start: datetime, end: datetime) -> List[Dict[str, Any]]: + return [ + {"name": "@start_date", "value": start.isoformat()}, + {"name": "@end_date", "value": end.isoformat()}, + ] + + +def query_count(container, query, parameters=None): + """Execute a cross-partition aggregate count.""" + rows = list(container.query_items( + query=query, + parameters=parameters or [], + enable_cross_partition_query=True, + )) + return int(rows[0] or 0) if rows else 0 + + +def count_active_users(container, start_date, end_date): + """Count distinct users with recorded login activity in a UTC interval. + + The Python Cosmos SDK cannot run COUNT over a DISTINCT subquery across partitions, + so the distinct user IDs are read with SELECT DISTINCT VALUE and counted here. + """ + user_ids = container.query_items( + query=""" + SELECT DISTINCT VALUE c.user_id FROM c + WHERE c.activity_type = 'user_login' + AND IS_DEFINED(c.user_id) + AND c.timestamp >= @start_date + AND c.timestamp <= @end_date + """, + parameters=_period_parameters(start_date, end_date), + enable_cross_partition_query=True, + ) + return len(set(user_ids)) + + +def count_activity(container, activity_type, start_date, end_date): + """Count recorded events of one activity type in a UTC interval.""" + return query_count( + container, + f""" + SELECT VALUE COUNT(1) FROM c + WHERE c.activity_type = @activity_type + AND {RECORDED_IN_WINDOW} + """, + [{"name": "@activity_type", "value": activity_type}] + _period_parameters(start_date, end_date), + ) + + +def count_document_uploads(container, start_date, end_date): + """Count document-creation activity by the recorded workspace type. + + Cross-partition GROUP BY is unavailable to the Python Cosmos SDK, so group and + public uploads are counted directly. Every other recorded or missing type is + personal, so personal is the remainder of the period total. + """ + window = f""" + c.activity_type = 'document_creation' + AND {RECORDED_IN_WINDOW} + """ + parameters = _period_parameters(start_date, end_date) + total = query_count( + container, + f"SELECT VALUE COUNT(1) FROM c WHERE {window}", + parameters, + ) + counts = { + workspace_type: query_count( + container, + f"SELECT VALUE COUNT(1) FROM c WHERE {window} AND c.workspace_type = @workspace_type", + parameters + [{"name": "@workspace_type", "value": workspace_type}], + ) + for workspace_type in ("group", "public") + } + return {"personal": max(total - counts["group"] - counts["public"], 0), **counts} + + +def status_rows(container): + """Tally stored status values for dashboard_status_counts without a GROUP BY query.""" + statuses = Counter( + status if status is None or isinstance(status, str) else str(status) + for status in container.query_items( + query="SELECT VALUE c.status FROM c WHERE IS_DEFINED(c.status)", + enable_cross_partition_query=True, + ) + ) + return [{"status": status, "count": count} for status, count in statuses.items()] + + +def token_total(container, start_date, end_date, token_filters): + """Aggregate token usage without returning individual activity records.""" + where_clause, parameters = build_token_usage_query_context( + start_date, + end_date, + token_filters=token_filters, + ) + rows = list(container.query_items( + query=f"SELECT VALUE SUM(c.usage.total_tokens) FROM c WHERE {where_clause}", + parameters=parameters, + enable_cross_partition_query=True, + )) + return int(rows[0] or 0) if rows else 0 + + +def count_document_failures(container, start_date, end_date): + """Count stored document processing states that explicitly report failure.""" + return query_count( + container, + """ + SELECT VALUE COUNT(1) FROM c + WHERE IS_STRING(c.status) + AND (CONTAINS(LOWER(c.status), 'failed') OR CONTAINS(LOWER(c.status), 'error')) + AND ( + (c.upload_date >= @start_date AND c.upload_date <= @end_date) + OR (c.created_at >= @start_date AND c.created_at <= @end_date) + ) + """, + _period_parameters(start_date, end_date), + ) + + +def count_all_document_failures(stores: DashboardStores, start_date, end_date): + """Count failed processing records across personal, group, and public workspaces.""" + return sum( + count_document_failures(container, start_date, end_date) + for container in (stores.user_documents, stores.group_documents, stores.public_documents) + if container is not None + ) + + +def count_pending_approvals(container): + """Count approval requests that are still waiting for a decision.""" + return query_count( + container, + """ + SELECT VALUE COUNT(1) FROM c + WHERE c.status = 'pending' + """, + ) + + +# -------------------------------------------------------------------------------------- +# Streamed aggregation +# -------------------------------------------------------------------------------------- + +def token_value(value): + """Return the numeric token count Cosmos SUM would add; other values count as zero.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return 0 + return value + + +def rank_totals(totals, value_key, limit=CONTROL_CENTER_DASHBOARD_RANKING_LIMIT): + """Return the largest totals, with ties ordered by ID so the ranking is stable.""" + ranked = sorted(totals.items(), key=lambda item: (-item[1], item[0])) + return [ + {"id": entity_id, value_key: int(total)} + for entity_id, total in ranked[:limit] + ] + + +def _row_moment(row: Mapping[str, Any]) -> Optional[str]: + """The recorded time of an activity row as an ISO string, preferring ``timestamp``.""" + for field in ("timestamp", "created_at"): + value = row.get(field) + if isinstance(value, str) and value: + return value + return None + + +def _row_day(row: Mapping[str, Any]) -> Optional[str]: + moment = _row_moment(row) + if moment is None or len(moment) < 10: + return None + day = moment[:10] + try: + datetime.strptime(day, "%Y-%m-%d") + except ValueError: + return None + return day + + +def _row_weekday_hour(row: Mapping[str, Any]) -> Optional[Tuple[int, int]]: + moment = _row_moment(row) + if moment is None: + return None + try: + weekday = datetime.strptime(moment[:10], "%Y-%m-%d").weekday() + hour = int(moment[11:13]) + except ValueError: + return None + if 0 <= hour <= 23: + return weekday, hour + return None + + +def _upload_field(workspace_type: Any) -> str: + if workspace_type == "group": + return "uploads_group" + if workspace_type == "public": + return "uploads_public" + return "uploads_personal" + + +def _workspace_kind(workspace_type: Any) -> str: + if workspace_type in {"group", "public"}: + return workspace_type + return "personal" + + +def aggregate_token_usage(container, period: DashboardPeriod, token_filters=None) -> Dict[str, Any]: + """Total the filtered token records by day, model, usage type and consumer. + + The Python Cosmos SDK cannot run cross-partition GROUP BY, so one narrow projection of + the filtered token records is streamed and totalled here. + """ + where_clause, parameters = build_token_usage_query_context( + period.start, + period.end, + token_filters=token_filters, + ) + by_day_model = defaultdict(int) + by_day_type = defaultdict(int) + by_model = defaultdict(int) + by_type = defaultdict(int) + consumers = {key: defaultdict(int) for key, _ in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS} + total = 0 + rows = container.query_items( + query=f""" + SELECT c.timestamp, + c.created_at, + c.token_type AS token_type, + c.usage.model AS model, + c.usage.total_tokens AS tokens, + c.user_id AS user_id, + c.workspace_context.group_id AS group_id, + c.workspace_context.public_workspace_id AS public_workspace_id + FROM c + WHERE {where_clause} + """, + parameters=parameters, + enable_cross_partition_query=True, + ) + for row in rows: + tokens = token_value(row.get("tokens")) + total += tokens + day = _row_day(row) + if "model" in row: + model = str(row.get("model") or "Unknown model") + by_model[model] += tokens + if day: + by_day_model[(day, model)] += tokens + token_type = row.get("token_type") + if token_type in DASHBOARD_TOKEN_TYPES: + by_type[token_type] += tokens + if day: + by_day_type[(day, token_type)] += tokens + for key, field in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS: + entity_id = row.get(field) + if isinstance(entity_id, str) and entity_id: + consumers[key][entity_id] += tokens + + daily_by_type = {day: dict.fromkeys(DASHBOARD_TOKEN_TYPES, 0) for day in period.dates()} + for (day, token_type), tokens in by_day_type.items(): + if day in daily_by_type: + daily_by_type[day][token_type] += int(tokens) + return { + "total": int(total), + "token_usage_by_model": [ + {"date": day, "model": model, "tokens": int(tokens)} + for (day, model), tokens in sorted(by_day_model.items()) + ], + "token_usage_by_type": [ + {"date": day, **types} for day, types in daily_by_type.items() + ], + "by_model": {model: int(tokens) for model, tokens in by_model.items()}, + "by_type": {token_type: int(by_type.get(token_type, 0)) for token_type in DASHBOARD_TOKEN_TYPES}, + "consumers": {key: dict(values) for key, values in consumers.items()}, + } + + +def _daily_activity_template(period: DashboardPeriod) -> Dict[str, Dict[str, int]]: + return {day: dict.fromkeys(DASHBOARD_DAILY_ACTIVITY_FIELDS, 0) for day in period.dates()} + + +def _count_daily_activity(daily, row) -> None: + counts = daily.get(_row_day(row)) + if counts is None: + return + activity_type = row.get("activity_type") + if activity_type == "user_login": + counts["sign_ins"] += 1 + elif activity_type == "conversation_creation": + counts["conversations_created"] += 1 + elif activity_type == "document_creation": + counts[_upload_field(row.get("workspace_type"))] += 1 + + +def _activity_rows(container, period: DashboardPeriod, user_id: Optional[str] = None): + conditions = [RECORDED_IN_WINDOW] + parameters = _period_parameters(period.start, period.end) + if user_id: + conditions.append("c.user_id = @activity_user_id") + parameters.append({"name": "@activity_user_id", "value": user_id}) + where_clause = "\n AND ".join(conditions) + return container.query_items( + query=f""" + SELECT c.timestamp, + c.created_at, + c.activity_type, + c.workspace_type, + c.user_id AS user_id, + c.workspace_context.group_id AS group_id, + c.workspace_context.public_workspace_id AS public_workspace_id + FROM c + WHERE {where_clause} + """, + parameters=parameters, + enable_cross_partition_query=True, + ) + + +def aggregate_activity(container, period: DashboardPeriod) -> Dict[str, Any]: + """Count recorded activity by day, actor and sign-in hour from one streamed projection. + + Each heatmap cell totals every sign-in in the period for one UTC weekday and hour. + """ + daily = _daily_activity_template(period) + actors = {key: defaultdict(int) for key, _ in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS} + logins = Counter() + for row in _activity_rows(container, period): + for key, field in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS: + entity_id = row.get(field) + if isinstance(entity_id, str) and entity_id: + actors[key][entity_id] += 1 + _count_daily_activity(daily, row) + if row.get("activity_type") == "user_login": + weekday_hour = _row_weekday_hour(row) + if weekday_hour is not None: + logins[weekday_hour] += 1 + return { + "daily_activity": [{"date": day, **counts} for day, counts in daily.items()], + "actors": {key: dict(values) for key, values in actors.items()}, + "login_cells": [ + {"weekday": weekday, "hour": hour, "count": count} + for (weekday, hour), count in sorted(logins.items()) + ], + } + + +def _matches_workspace_filters(row: Mapping[str, Any], filters: Mapping[str, Any]) -> bool: + workspace_type = filters.get("workspace_type") + if workspace_type and _workspace_kind(row.get("workspace_type")) != workspace_type: + return False + group_id = filters.get("group_id") + if group_id and row.get("group_id") != group_id: + return False + public_workspace_id = filters.get("public_workspace_id") + if public_workspace_id and row.get("public_workspace_id") != public_workspace_id: + return False + return True + + +def filtered_daily_activity(container, period: DashboardPeriod, filters: Mapping[str, Any]) -> List[Dict[str, Any]]: + """Daily activity narrowed to one user and/or workspace. + + Sign-ins are not recorded against a workspace, so the workspace filters narrow + conversations and uploads only; sign-ins honor the user filter alone. + """ + daily = _daily_activity_template(period) + for row in _activity_rows(container, period, user_id=filters.get("user_id")): + if row.get("activity_type") != "user_login" and not _matches_workspace_filters(row, filters): + continue + _count_daily_activity(daily, row) + return [{"date": day, **counts} for day, counts in daily.items()] + + +# -------------------------------------------------------------------------------------- +# Names +# -------------------------------------------------------------------------------------- + +def _clean_text(value: Any, limit: int = 200) -> str: + if not isinstance(value, str): + return "" + return " ".join(value.split())[:limit] + + +def _entity_container(stores: DashboardStores, kind: str): + return { + "users": stores.user_settings, + "groups": stores.groups, + "public_workspaces": stores.public_workspaces, + }.get(kind) + + +def _describe_entity(kind: str, document: Any) -> Dict[str, Any]: + if not isinstance(document, dict): + return {"name": DASHBOARD_ENTITY_MISSING_NAMES[kind], "detail": "", "found": False} + if kind == "users": + display_name = _clean_text(document.get("display_name")) + email = _clean_text(document.get("email")) + return { + "name": display_name or email or DASHBOARD_ENTITY_UNNAMED[kind], + "detail": email if display_name else "", + "found": True, + } + return { + "name": _clean_text(document.get("name")) or DASHBOARD_ENTITY_UNNAMED[kind], + "detail": "", + "found": True, + } + + +def resolve_entity_names( + stores: DashboardStores, + ids_by_kind: Mapping[str, Iterable[str]], + *, + log: Optional[Callable[..., Any]] = None, + now=None, +) -> Dict[str, Dict[str, Dict[str, Any]]]: + """Look up display names for ranked IDs, caching each answer as briefly as the dashboard. + + A user, group or workspace that cannot be read is reported with a fallback name rather + than failing the dashboard: rankings are built from activity that can outlive the record. + """ + resolved = {kind: {} for kind in DASHBOARD_ENTITY_KINDS} + for kind, entity_ids in ids_by_kind.items(): + if kind not in DASHBOARD_ENTITY_KINDS: + continue + container = _entity_container(stores, kind) + for entity_id in dict.fromkeys(entity_ids or ()): + if not isinstance(entity_id, str) or not entity_id: + continue + cache_key = (kind, entity_id) + described = _cache_get(_control_center_entity_name_cache, cache_key, now=now) + if described is None: + document = None + if container is not None: + try: + document = container.read_item(item=entity_id, partition_key=entity_id) + except Exception as ex: + if log is not None: + log( + "[CONTROL_CENTER] A ranked dashboard entity could not be resolved.", + extra={"entity_kind": kind, "error_type": type(ex).__name__}, + level=logging.DEBUG, + debug_only=True, + ) + described = _describe_entity(kind, document) + _cache_set( + _control_center_entity_name_cache, + cache_key, + described, + CONTROL_CENTER_DASHBOARD_NAME_CACHE_MAX_ENTRIES, + now=now, + ) + resolved[kind][entity_id] = described + return resolved + + +def name_ranked_rows(rows: Iterable[Mapping[str, Any]], names: Mapping[str, Mapping[str, Any]]) -> List[Dict[str, Any]]: + """Return ranked rows with each entity's name, detail and whether it still exists.""" + named = [] + for row in rows: + described = names.get(row.get("id")) or {} + named.append({ + **row, + "name": described.get("name") or "", + "detail": described.get("detail") or "", + "found": bool(described.get("found")), + }) + return named + + +def _name_rankings(stores, rankings, *, log=None, now=None): + names = resolve_entity_names( + stores, + {kind: [row["id"] for row in rows] for kind, rows in rankings.items()}, + log=log, + now=now, + ) + return {kind: name_ranked_rows(rows, names.get(kind, {})) for kind, rows in rankings.items()} + + +# -------------------------------------------------------------------------------------- +# Summary +# -------------------------------------------------------------------------------------- + +def _summary_base(stores: DashboardStores, period: DashboardPeriod, *, log=None) -> Dict[str, Any]: + total_users = query_count(stores.user_settings, "SELECT VALUE COUNT(1) FROM c") + blocked_users = query_count( + stores.user_settings, + """ + SELECT VALUE COUNT(1) FROM c + WHERE c.settings.access.status = 'deny' + """, + ) + + groups_total = query_count(stores.groups, "SELECT VALUE COUNT(1) FROM c") + groups_by_status = dashboard_status_counts(groups_total, status_rows(stores.groups)) + + workspaces_total = query_count(stores.public_workspaces, "SELECT VALUE COUNT(1) FROM c") + workspaces_by_status = dashboard_status_counts( + workspaces_total, + status_rows(stores.public_workspaces), + unknown_status="inactive", + ) + + activity = stores.activity_logs + end_day = period.end.date() + active_users = count_active_users(activity, period.start, period.end) + previous_active_users = count_active_users(activity, period.previous_start, period.previous_end) + current_dau = count_active_users(activity, datetime.combine(end_day, datetime.min.time()), period.end) + current_wau = count_active_users( + activity, + datetime.combine(end_day - timedelta(days=6), datetime.min.time()), + period.end, + ) + current_mau = count_active_users( + activity, + datetime.combine(end_day - timedelta(days=29), datetime.min.time()), + period.end, + ) + + conversations = count_activity(activity, "conversation_creation", period.start, period.end) + previous_conversations = count_activity( + activity, + "conversation_creation", + period.previous_start, + period.previous_end, + ) + document_uploads = count_document_uploads(activity, period.start, period.end) + previous_document_uploads = count_document_uploads(activity, period.previous_start, period.previous_end) + + try: + failure_metric = { + **dashboard_metric( + count_all_document_failures(stores, period.start, period.end), + count_all_document_failures(stores, period.previous_start, period.previous_end), + ), + "available": True, + } + except Exception as ex: + if log is not None: + log( + "[CONTROL_CENTER] Document failure metrics are unavailable.", + extra={"error_type": type(ex).__name__}, + level=logging.WARNING, + ) + failure_metric = { + "value": None, + "delta": None, + "percent_change": None, + "available": False, + } + + pending_approvals = None + if stores.approvals is not None: + try: + pending_approvals = count_pending_approvals(stores.approvals) + except Exception as ex: + if log is not None: + log( + "[CONTROL_CENTER] Pending approval count is unavailable.", + extra={"error_type": type(ex).__name__}, + level=logging.WARNING, + ) + + return { + "computed_at": _utc_now_text(), + "users": { + "total": dashboard_metric(total_users), + "active": dashboard_metric(active_users, previous_active_users), + "dau": dashboard_metric(current_dau), + "wau": dashboard_metric(current_wau), + "mau": dashboard_metric(current_mau), + "blocked": dashboard_metric(blocked_users), + }, + "groups": { + "total": dashboard_metric(groups_total), + "by_status": {status: dashboard_metric(count) for status, count in groups_by_status.items()}, + }, + "public_workspaces": { + "total": dashboard_metric(workspaces_total), + "by_status": {status: dashboard_metric(count) for status, count in workspaces_by_status.items()}, + }, + "conversations": dashboard_metric(conversations, previous_conversations), + "document_uploads": { + "total": dashboard_metric(sum(document_uploads.values()), sum(previous_document_uploads.values())), + "by_workspace_type": { + workspace_type: dashboard_metric(count, previous_document_uploads.get(workspace_type, 0)) + for workspace_type, count in document_uploads.items() + }, + }, + "document_processing_failures": failure_metric, + "pending_approvals": ( + dashboard_metric(pending_approvals) if pending_approvals is not None else None + ), + } + + +def _summary_tokens(stores: DashboardStores, period: DashboardPeriod, token_filters) -> Dict[str, Any]: + return { + "computed_at": _utc_now_text(), + "tokens": dashboard_metric( + token_total(stores.activity_logs, period.start, period.end, token_filters), + token_total(stores.activity_logs, period.previous_start, period.previous_end, token_filters), + ), + } + + +def _cached_half(key, build, *, force_refresh=False, now=None): + cached = dashboard_cache_get(key, force_refresh=force_refresh, now=now) + if cached is not None: + return cached, True + payload = build() + dashboard_cache_set(key, payload, now=now) + return payload, False + + +def get_dashboard_summary( + stores: DashboardStores, + period: DashboardPeriod, + token_filters=None, + *, + force_refresh: bool = False, + log: Optional[Callable[..., Any]] = None, + now=None, +) -> Tuple[Dict[str, Any], bool]: + """Return the dashboard's KPI metrics and whether every part came from the cache. + + The token figures are cached separately from the rest, so changing a token filter + recalculates only the token totals. + """ + base, base_cached = _cached_half( + ("summary-base", *period.cache_key()), + lambda: _summary_base(stores, period, log=log), + force_refresh=force_refresh, + now=now, + ) + tokens, tokens_cached = _cached_half( + ("summary-tokens", *period.cache_key(), _filters_cache_key(token_filters)), + lambda: _summary_tokens(stores, period, token_filters), + force_refresh=force_refresh, + now=now, + ) + payload = { + "period": period.describe(), + "refreshed_at": min(base["computed_at"], tokens["computed_at"]), + "users": base["users"], + "groups": base["groups"], + "public_workspaces": base["public_workspaces"], + "conversations": base["conversations"], + "document_uploads": base["document_uploads"], + "document_processing_failures": base["document_processing_failures"], + "tokens": tokens["tokens"], + "pending_approvals": base["pending_approvals"], + "status_history_available": False, + } + return payload, base_cached and tokens_cached + + +# -------------------------------------------------------------------------------------- +# Insights +# -------------------------------------------------------------------------------------- + +def _activity_half(stores, period, *, force_refresh=False, log=None, now=None): + def build(): + aggregate = aggregate_activity(stores.activity_logs, period) + rankings = { + key: rank_totals(aggregate["actors"][key], "activity_count") + for key, _ in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS + } + return { + "computed_at": _utc_now_text(), + "daily_activity": aggregate["daily_activity"], + "actors": aggregate["actors"], + "top_activity": _name_rankings(stores, rankings, log=log, now=now), + "login_cells": aggregate["login_cells"], + } + + return _cached_half( + ("insights-activity", *period.cache_key()), + build, + force_refresh=force_refresh, + now=now, + ) + + +def _token_half(stores, period, token_filters, *, force_refresh=False, log=None, now=None): + def build(): + aggregate = aggregate_token_usage(stores.activity_logs, period, token_filters) + rankings = { + key: rank_totals(aggregate["consumers"][key], "tokens") + for key, _ in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS + } + return { + "computed_at": _utc_now_text(), + "aggregate": aggregate, + "top_tokens": _name_rankings(stores, rankings, log=log, now=now), + } + + return _cached_half( + ("insights-tokens", *period.cache_key(), _filters_cache_key(token_filters)), + build, + force_refresh=force_refresh, + now=now, + ) + + +def get_dashboard_insights( + stores: DashboardStores, + period: DashboardPeriod, + token_filters=None, + *, + force_refresh: bool = False, + log: Optional[Callable[..., Any]] = None, + now=None, +) -> Tuple[Dict[str, Any], bool]: + """Return the dashboard's chart data and rankings, and whether all of it was cached. + + Activity (sign-ins, conversations, uploads, rankings and the sign-in heatmap) is cached + by period. Token data is cached by period and token filters, so changing a token filter + does not re-read the rest of the period's activity. + """ + activity, activity_cached = _activity_half( + stores, period, force_refresh=force_refresh, log=log, now=now, + ) + tokens, tokens_cached = _token_half( + stores, period, token_filters, force_refresh=force_refresh, log=log, now=now, + ) + payload = { + "period": period.describe(), + "refreshed_at": min(activity["computed_at"], tokens["computed_at"]), + "daily_activity": activity["daily_activity"], + "token_usage_by_type": tokens["aggregate"]["token_usage_by_type"], + "token_usage_by_model": tokens["aggregate"]["token_usage_by_model"], + "top_tokens": tokens["top_tokens"], + "top_activity": activity["top_activity"], + "login_heatmap": { + "weekday_convention": DASHBOARD_LOGIN_HEATMAP_CONVENTION, + "timezone": "UTC", + "cells": activity["login_cells"], + }, + } + return payload, activity_cached and tokens_cached + + +# -------------------------------------------------------------------------------------- +# Reports for the Control Center action +# -------------------------------------------------------------------------------------- + +def summary_report_rows(summary: Mapping[str, Any]) -> List[Dict[str, Any]]: + """Flatten a dashboard summary into one row per figure, with what each figure counts.""" + period = summary.get("period") or {} + end_day = period.get("end_date") or "the end date" + users = summary.get("users") or {} + uploads = summary.get("document_uploads") or {} + rows = [] + + def add(metric, figure, definition, *, comparable=True): + figure = figure or {} + row = {"metric": metric, "value": figure.get("value"), "definition": definition} + if comparable: + row.update({ + "previous_period_value": figure.get("previous"), + "change": figure.get("delta"), + "percent_change": figure.get("percent_change"), + }) + rows.append(row) + + add("Total users", users.get("total"), "Accounts with a SimpleChat profile today.", comparable=False) + add("Blocked users", users.get("blocked"), "Accounts currently denied access.", comparable=False) + add( + "Signed-in users", + users.get("active"), + "People who signed in at least once in the period; each person counts once.", + ) + add( + "Daily active users", + users.get("dau"), + f"People who signed in on {end_day} (UTC).", + comparable=False, + ) + add( + "Weekly active users", + users.get("wau"), + f"People who signed in during the 7 days ending {end_day} (UTC).", + comparable=False, + ) + add( + "Monthly active users", + users.get("mau"), + f"People who signed in during the 30 days ending {end_day} (UTC).", + comparable=False, + ) + add("Conversations created", summary.get("conversations"), "New conversations started in the period.") + add("Document uploads", uploads.get("total"), "Documents added to any workspace in the period.") + for workspace_type, figure in (uploads.get("by_workspace_type") or {}).items(): + add( + f"Document uploads ({workspace_type})", + figure, + f"Documents added to {workspace_type} workspaces in the period.", + ) + failures = summary.get("document_processing_failures") or {} + if failures.get("available", True): + add( + "Processing failures", + failures, + "Documents uploaded in the period whose processing reports a failure.", + ) + add("Tokens used", summary.get("tokens"), "Model tokens recorded in the period.") + for label, section in (("Groups", summary.get("groups")), ("Public workspaces", summary.get("public_workspaces"))): + section = section or {} + add(label, section.get("total"), f"{label} that exist today.", comparable=False) + for status, figure in (section.get("by_status") or {}).items(): + add( + f"{label} ({status.replace('_', ' ')})", + figure, + f"{label} whose current status is {status.replace('_', ' ')}.", + comparable=False, + ) + if summary.get("pending_approvals") is not None: + add( + "Pending approvals", + summary.get("pending_approvals"), + "Approval requests waiting for a decision now.", + comparable=False, + ) + return rows + + +def daily_activity_report( + stores: DashboardStores, + period: DashboardPeriod, + filters: Optional[Mapping[str, Any]] = None, + *, + force_refresh: bool = False, + log: Optional[Callable[..., Any]] = None, + now=None, +) -> List[Dict[str, Any]]: + """Daily sign-ins, conversations and uploads, optionally narrowed to a user or workspace.""" + filters = {key: value for key, value in (filters or {}).items() if value} + if not filters: + activity, _ = _activity_half(stores, period, force_refresh=force_refresh, log=log, now=now) + rows = activity["daily_activity"] + else: + rows, _ = _cached_half( + ("report-daily-activity", *period.cache_key(), _filters_cache_key(filters)), + lambda: filtered_daily_activity(stores.activity_logs, period, filters), + force_refresh=force_refresh, + now=now, + ) + # Copies, so a caller that reshapes a row cannot change the cached answer. + return [dict(row) for row in rows] + + +_TOKEN_ENTITY_GROUPINGS = { + "user": "users", + "group": "groups", + "public_workspace": "public_workspaces", +} + + +def token_usage_report( + stores: DashboardStores, + period: DashboardPeriod, + token_filters=None, + *, + group_by: str = "day", + limit: int = CONTROL_CENTER_DASHBOARD_RANKING_LIMIT, + force_refresh: bool = False, + log: Optional[Callable[..., Any]] = None, + now=None, +) -> Dict[str, Any]: + """Token usage totalled one way: by day, model, usage type, user, group or workspace.""" + if group_by not in DASHBOARD_TOKEN_GROUPINGS: + raise ValueError(f"group_by must be one of: {', '.join(DASHBOARD_TOKEN_GROUPINGS)}.") + limit = max(1, min(int(limit), DASHBOARD_REPORT_MAX_ROWS)) + tokens, _ = _token_half( + stores, period, token_filters, force_refresh=force_refresh, log=log, now=now, + ) + aggregate = tokens["aggregate"] + if group_by == "day": + rows = [ + { + "date": row["date"], + "total_tokens": sum(int(row.get(token_type) or 0) for token_type in DASHBOARD_TOKEN_TYPES), + **{token_type: int(row.get(token_type) or 0) for token_type in DASHBOARD_TOKEN_TYPES}, + } + for row in aggregate["token_usage_by_type"] + ] + elif group_by == "model": + rows = [ + {"model": model, "tokens": int(total)} + for model, total in sorted(aggregate["by_model"].items(), key=lambda item: (-item[1], item[0]))[:limit] + ] + elif group_by == "token_type": + rows = [ + {"token_type": token_type, "tokens": int(aggregate["by_type"].get(token_type, 0))} + for token_type in DASHBOARD_TOKEN_TYPES + ] + else: + kind = _TOKEN_ENTITY_GROUPINGS[group_by] + ranked = rank_totals(aggregate["consumers"][kind], "tokens", limit=limit) + names = resolve_entity_names(stores, {kind: [row["id"] for row in ranked]}, log=log, now=now) + rows = name_ranked_rows(ranked, names.get(kind, {})) + return {"total_tokens": int(aggregate["total"]), "rows": rows} + + +def top_activity_report( + stores: DashboardStores, + period: DashboardPeriod, + *, + entity: str = "users", + limit: int = CONTROL_CENTER_DASHBOARD_RANKING_LIMIT, + force_refresh: bool = False, + log: Optional[Callable[..., Any]] = None, + now=None, +) -> List[Dict[str, Any]]: + """The users, groups or public workspaces with the most recorded actions in a period.""" + if entity not in DASHBOARD_ENTITY_KINDS: + raise ValueError(f"entity must be one of: {', '.join(DASHBOARD_ENTITY_KINDS)}.") + limit = max(1, min(int(limit), DASHBOARD_REPORT_MAX_ROWS)) + activity, _ = _activity_half(stores, period, force_refresh=force_refresh, log=log, now=now) + ranked = rank_totals(activity["actors"][entity], "activity_count", limit=limit) + names = resolve_entity_names(stores, {entity: [row["id"] for row in ranked]}, log=log, now=now) + return name_ranked_rows(ranked, names.get(entity, {})) + + +def sign_in_pattern_report( + stores: DashboardStores, + period: DashboardPeriod, + *, + force_refresh: bool = False, + log: Optional[Callable[..., Any]] = None, + now=None, +) -> List[Dict[str, Any]]: + """Sign-ins totalled by UTC weekday and hour across the whole period.""" + activity, _ = _activity_half(stores, period, force_refresh=force_refresh, log=log, now=now) + return [ + { + "weekday": DASHBOARD_WEEKDAYS[cell["weekday"]], + "hour_utc": cell["hour"], + "sign_ins": cell["count"], + } + for cell in activity["login_cells"] + ] + + +def find_entities(stores: DashboardStores, kind: str, search: Any, *, limit: int = DASHBOARD_ENTITY_SEARCH_MAX_RESULTS): + """Find users by display name or email, or groups and public workspaces by name.""" + if kind not in DASHBOARD_ENTITY_KINDS: + raise ValueError(f"kind must be one of: {', '.join(DASHBOARD_ENTITY_KINDS)}.") + normalized = " ".join(str(search or "").split()).lower() + if len(normalized) < DASHBOARD_ENTITY_SEARCH_MIN_LENGTH: + raise ValueError(f"Search for at least {DASHBOARD_ENTITY_SEARCH_MIN_LENGTH} characters.") + if len(normalized) > DASHBOARD_ENTITY_SEARCH_MAX_LENGTH: + raise ValueError(f"Search text may not exceed {DASHBOARD_ENTITY_SEARCH_MAX_LENGTH} characters.") + limit = max(1, min(int(limit), DASHBOARD_ENTITY_SEARCH_MAX_RESULTS)) + container = _entity_container(stores, kind) + if container is None: + return [] + if kind == "users": + query = """ + SELECT TOP @limit c.id, c.display_name, c.email FROM c + WHERE (IS_STRING(c.display_name) AND CONTAINS(LOWER(c.display_name), @search)) + OR (IS_STRING(c.email) AND CONTAINS(LOWER(c.email), @search)) + """ + else: + query = """ + SELECT TOP @limit c.id, c.name FROM c + WHERE IS_STRING(c.name) AND CONTAINS(LOWER(c.name), @search) + """ + matches = container.query_items( + query=query, + parameters=[ + {"name": "@limit", "value": limit}, + {"name": "@search", "value": normalized}, + ], + enable_cross_partition_query=True, + ) + results = [] + for document in matches: + entity_id = document.get("id") if isinstance(document, dict) else None + if not isinstance(entity_id, str) or not entity_id: + continue + described = _describe_entity(kind, document) + result = {"id": entity_id, "name": described["name"]} + if kind == "users": + result["email"] = _clean_text(document.get("email")) + results.append(result) + if len(results) >= limit: + break + return results diff --git a/application/single_app/functions_control_center_dashboard_chat.py b/application/single_app/functions_control_center_dashboard_chat.py new file mode 100644 index 000000000..88f489381 --- /dev/null +++ b/application/single_app/functions_control_center_dashboard_chat.py @@ -0,0 +1,210 @@ +# functions_control_center_dashboard_chat.py +"""Decide whether "Chat with this dashboard" can run, and say what is missing when it cannot. + +The dashboard's chat opens an orchestrated conversation that answers through a Control Center +action, so it depends on three administrator settings and on a Control Center action that an +administrator creates. Each requirement is reported on its own, with what to do about it, so a +dashboard viewer sees exactly which part is missing instead of a chat that silently cannot +answer. Settings, roles and storage results are passed in, so this module needs no application +bootstrap and its answers can be tested directly. +""" + +from typing import Any, Callable, Dict, Iterable, List, Mapping + + +# Admin Settings sections the remedies link to, as V2 /admin/settings/
paths. +DASHBOARD_CHAT_SETTINGS_LINKS = { + "agents": "/admin/settings/agents-config", + "orchestration": "/admin/settings/chat-orchestration-section", + "action_access": "/admin/settings/chat-orchestration-capabilities-section", + "governance": "/admin/settings/governance-feature-toggles-section", + "actions": "/admin/actions", + "new_action": "/admin/actions/new?type=control_center", +} +DASHBOARD_CHAT_ROLES = frozenset({"User", "Admin"}) + + +def _action_name(action: Mapping[str, Any]) -> str: + for field in ("display_name", "displayName", "name"): + value = action.get(field) + if isinstance(value, str) and value.strip(): + return " ".join(value.split())[:200] + return "Control Center" + + +def _requirement(requirement_id, label, met, *, detail="", admin_remedy="", reader_remedy="", link=None, is_admin=False): + return { + "id": requirement_id, + "label": label, + "met": bool(met), + "detail": "" if met else detail, + "remedy": "" if met else (admin_remedy if is_admin else reader_remedy), + "settings_link": link if (is_admin and not met and link) else None, + } + + +def build_dashboard_chat_readiness( + *, + settings: Mapping[str, Any], + user_roles: Iterable[str], + is_admin: bool, + stored_actions: Iterable[Mapping[str, Any]], + available_actions: Iterable[Mapping[str, Any]], + governance_allows: Callable[[Mapping[str, Any]], bool], + action_invoke_allowlisted: bool, +) -> Dict[str, Any]: + """Return each requirement for dashboard chat, whether it is met, and how to meet it. + + ``stored_actions`` are the global Control Center action records; ``available_actions`` are + the Control Center actions the action catalog offers this caller, which already applies + scope, enablement, governance and Control Center access. Remedies name the Admin Settings + location; links are included only for administrators, who can open those pages. + """ + settings = settings or {} + roles = {str(role) for role in (user_roles or [])} + stored = [action for action in (stored_actions or []) if isinstance(action, Mapping)] + available = [action for action in (available_actions or []) if isinstance(action, Mapping)] + requirements: List[Dict[str, Any]] = [] + + agents_on = bool(settings.get("enable_semantic_kernel")) + requirements.append(_requirement( + "agents", + "Agents and actions are turned on", + agents_on, + detail="Dashboard chat answers through a Control Center action, and actions run on the agent runtime.", + admin_remedy="Turn on Enable Agents in Admin Settings › Agents & Actions.", + reader_remedy="Ask a SimpleChat administrator to turn on Enable Agents in Admin Settings › Agents & Actions.", + link=DASHBOARD_CHAT_SETTINGS_LINKS["agents"], + is_admin=is_admin, + )) + + orchestration_on = bool(settings.get("enable_chat_orchestration")) + requirements.append(_requirement( + "orchestration", + "Chat Orchestration is turned on", + orchestration_on, + detail="Chat Orchestration plans the questions and decides when to use the Control Center action.", + admin_remedy="Turn on Enable Chat Orchestration in Admin Settings › Orchestration.", + reader_remedy="Ask a SimpleChat administrator to turn on Enable Chat Orchestration in Admin Settings › Orchestration.", + link=DASHBOARD_CHAT_SETTINGS_LINKS["orchestration"], + is_admin=is_admin, + )) + + action_access_switch = bool(settings.get("enable_chat_orchestration_actions")) + action_access_on = action_access_switch and bool(action_invoke_allowlisted) + if not action_access_switch: + action_detail = "Enable Action Access is off, so orchestration cannot use any action." + admin_remedy = "Turn on Enable Action Access in Admin Settings › Orchestration › Capabilities." + reader_remedy = ( + "Ask a SimpleChat administrator to turn on Enable Action Access in " + "Admin Settings › Orchestration › Capabilities." + ) + else: + action_detail = "The orchestration capability list leaves out Use an action." + admin_remedy = "Add Use an action to the capability list in Admin Settings › Orchestration › Capabilities." + reader_remedy = ( + "Ask a SimpleChat administrator to add Use an action to the capability list in " + "Admin Settings › Orchestration › Capabilities." + ) + requirements.append(_requirement( + "action_access", + "Orchestration can use actions", + action_access_on, + detail=action_detail, + admin_remedy=admin_remedy, + reader_remedy=reader_remedy, + link=DASHBOARD_CHAT_SETTINGS_LINKS["action_access"], + is_admin=is_admin, + )) + + action_ready = bool(available) + enabled_actions = [action for action in stored if action.get("is_enabled", True) is True] + if action_ready: + action_detail, admin_remedy, reader_remedy, link = "", "", "", None + elif not stored: + action_detail = "No Control Center action exists yet." + admin_remedy = ( + "Create one in Admin Settings › Agents & Actions › Global Actions: choose New action, then Control Center. " + "It needs no connection settings." + ) + reader_remedy = "Ask a SimpleChat administrator to create a Control Center action in Admin Settings › Agents & Actions › Global Actions." + link = DASHBOARD_CHAT_SETTINGS_LINKS["new_action"] + elif not enabled_actions: + action_detail = "The Control Center action is turned off." + admin_remedy = "Turn the Control Center action on in Admin Settings › Agents & Actions › Global Actions." + reader_remedy = "Ask a SimpleChat administrator to turn on the Control Center action in Admin Settings › Agents & Actions › Global Actions." + link = DASHBOARD_CHAT_SETTINGS_LINKS["actions"] + elif not agents_on: + action_detail = "A Control Center action exists, and becomes available once agents and actions are on." + admin_remedy = "Turn on Enable Agents in Admin Settings › Agents & Actions." + reader_remedy = "Ask a SimpleChat administrator to turn on Enable Agents in Admin Settings › Agents & Actions." + link = DASHBOARD_CHAT_SETTINGS_LINKS["agents"] + elif settings.get("per_user_semantic_kernel") and not settings.get("merge_global_semantic_kernel_with_workspace"): + action_detail = "Workspace Mode is on, which hides global actions such as the Control Center action." + admin_remedy = "Turn on Add Global Agents and Actions to Workspaces in Admin Settings › Agents & Actions." + reader_remedy = ( + "Ask a SimpleChat administrator to turn on Add Global Agents and Actions to Workspaces in " + "Admin Settings › Agents & Actions." + ) + link = DASHBOARD_CHAT_SETTINGS_LINKS["agents"] + elif not any(governance_allows(action) for action in enabled_actions): + action_detail = "Governance does not let you use the Control Center action." + admin_remedy = "Allow it for you under Govern Global Actions in Admin Settings › Governance." + reader_remedy = "Ask a SimpleChat administrator to allow the Control Center action for you in Admin Settings › Governance." + link = DASHBOARD_CHAT_SETTINGS_LINKS["governance"] + else: + action_detail = "The Control Center action is not available to your account right now." + admin_remedy = "Check that the Control Center action is a global action and that your app roles include Control Center access." + reader_remedy = "Ask a SimpleChat administrator to check the Control Center action and your Control Center access." + link = DASHBOARD_CHAT_SETTINGS_LINKS["actions"] + requirements.append(_requirement( + "control_center_action", + "A Control Center action is set up for you", + action_ready, + detail=action_detail, + admin_remedy=admin_remedy, + reader_remedy=reader_remedy, + link=link, + is_admin=is_admin, + )) + + can_chat = bool(roles & DASHBOARD_CHAT_ROLES) + if not can_chat: + requirements.append(_requirement( + "chat_access", + "Your account can use chat", + False, + detail="Chat is available to accounts with the User or Admin app role.", + admin_remedy="Assign your account the User app role in Microsoft Entra ID.", + reader_remedy="Ask your identity administrator to assign your account the User app role.", + is_admin=is_admin, + )) + + action = None + if available: + chosen = sorted(available, key=lambda item: (_action_name(item).lower(), str(item.get("action_ref") or "")))[0] + action = {"name": _action_name(chosen)} + return { + "ready": all(requirement["met"] for requirement in requirements), + "action": action, + "requirements": requirements, + } + + +def control_center_action_records(records: Iterable[Any], action_type: str) -> List[Dict[str, Any]]: + """Keep the stored global actions of one type, reduced to the fields readiness reads.""" + kept = [] + for record in records or []: + if not isinstance(record, dict): + continue + record_type = record.get("type") + if not isinstance(record_type, str) or record_type.strip().lower() != action_type: + continue + kept.append({ + "id": record.get("id"), + "name": record.get("name"), + "display_name": record.get("display_name") or record.get("displayName"), + "type": record_type, + "is_enabled": record.get("is_enabled", True), + }) + return kept diff --git a/application/single_app/functions_governance.py b/application/single_app/functions_governance.py index 84ae7640a..59ffe4301 100644 --- a/application/single_app/functions_governance.py +++ b/application/single_app/functions_governance.py @@ -17,6 +17,7 @@ from config import cosmos_governance_item_policies_container, cosmos_governance_policies_container from functions_activity_logging import log_governance_change from functions_action_manifest import resolve_action_type +from functions_control_center_dashboard import CONTROL_CENTER_ACTION_TYPE from functions_group import get_user_groups from functions_public_workspaces import get_user_public_workspaces from functions_settings import get_settings @@ -113,6 +114,7 @@ "blob_storage": "blob_storage", "document_search": "document_search", "search": "document_search", + CONTROL_CENTER_ACTION_TYPE: CONTROL_CENTER_ACTION_TYPE, } @@ -135,8 +137,13 @@ "azure_maps": "Azure Maps", "blob_storage": "Blob Storage", "document_search": "Document Search", + CONTROL_CENTER_ACTION_TYPE: "Control Center", } +# Action types that only an administrator may create, as global actions. A Control Center +# action reads organization-wide usage, so a personal or group copy would only ever refuse. +GLOBAL_ONLY_ACTION_TYPES = frozenset({CONTROL_CENTER_ACTION_TYPE}) + LEGACY_ITEM_POLICY_ENTITY_TYPE_ALIASES = { "endpoint": "global_endpoint", @@ -1277,6 +1284,11 @@ def ensure_action_type_access( raise PermissionError("Governance policy blocks access to an unknown action type.") if not action_type_entity_type: raise ValueError(f"Unsupported action type governance scope: {scope}") + if normalized_action_type in GLOBAL_ONLY_ACTION_TYPES and normalized_scope != "global": + action_type_label = get_governed_action_type_label(normalized_action_type) + raise PermissionError( + f"{action_type_label} actions are available only as global actions that an administrator creates." + ) decision_key = ( "action_type_access_decision", diff --git a/application/single_app/functions_orchestration_actions.py b/application/single_app/functions_orchestration_actions.py index 19ce18a40..b311c3a57 100644 --- a/application/single_app/functions_orchestration_actions.py +++ b/application/single_app/functions_orchestration_actions.py @@ -511,6 +511,7 @@ async def invoke_action( manifest = resolve_action_manifest( user_id, action_ref, settings=current_settings, user_groups=getattr(context, 'active_group_ids', None) or None, + user_roles=list(identity.roles), ) if invocation_capture is not None: current_settings = deepcopy(current_settings) @@ -569,6 +570,7 @@ async def guard_function(invocation, next): fresh = resolve_action_manifest( user_id, action_ref, settings=current, user_groups=getattr(context, 'active_group_ids', None) or None, + user_roles=list(identity.roles), ) if fresh != manifest: raise ActionExecutionError('The selected action changed during execution.') diff --git a/application/single_app/functions_orchestration_context.py b/application/single_app/functions_orchestration_context.py index 2a4e0e008..e42f1270b 100644 --- a/application/single_app/functions_orchestration_context.py +++ b/application/single_app/functions_orchestration_context.py @@ -1590,8 +1590,12 @@ def resolve_agent_catalog(user_id, seeds=None, settings=None, user_groups=None): # Run ledger # -------------------------------------------------------------------------------------- -def resolve_action_catalog(user_id, seeds=None, settings=None, user_groups=None): - """Discover action metadata only when this request can use direct actions.""" +def resolve_action_catalog(user_id, seeds=None, settings=None, user_groups=None, user_roles=None): + """Discover action metadata only when this request can use direct actions. + + ``user_roles`` are the caller's authenticated app roles, which decide whether a + Control Center action is offered. + """ settings = settings or {} seeds = seeds or {} if (seeds.get('agent') or {}).get('name'): @@ -1607,7 +1611,7 @@ def resolve_action_catalog(user_id, seeds=None, settings=None, user_groups=None) from functions_action_catalog import build_accessible_action_catalog return build_accessible_action_catalog( - user_id, settings=settings, user_groups=user_groups, + user_id, settings=settings, user_groups=user_groups, user_roles=user_roles, ) diff --git a/application/single_app/functions_orchestration_execution.py b/application/single_app/functions_orchestration_execution.py index 55bebdccb..3b5d5d55f 100644 --- a/application/single_app/functions_orchestration_execution.py +++ b/application/single_app/functions_orchestration_execution.py @@ -675,6 +675,7 @@ def _initialize(self, settings, identity_context, execution_identity): actions = resolve_action_catalog( user_id, seeds=seeds, settings=self.settings, user_groups=seeds.get("active_group_ids") or None, + user_roles=identity["user_roles"], ) user_message = self.record.get("user_message") or "" allowed_urls = list(dict.fromkeys([ diff --git a/application/single_app/functions_orchestration_plan_editing.py b/application/single_app/functions_orchestration_plan_editing.py index 00faec98d..5dc28baca 100644 --- a/application/single_app/functions_orchestration_plan_editing.py +++ b/application/single_app/functions_orchestration_plan_editing.py @@ -312,6 +312,7 @@ def _revision_catalogs( actions = resolve_action_catalog( user_id, seeds=seeds, settings=settings, user_groups=seeds.get('active_group_ids') or None, + user_roles=identity.get('user_roles') or [], ) runtime_options = {} if callable(native_bridge_for_step): diff --git a/application/single_app/json_schema_validation.py b/application/single_app/json_schema_validation.py index 0be79ee75..fe7d7c891 100644 --- a/application/single_app/json_schema_validation.py +++ b/application/single_app/json_schema_validation.py @@ -22,6 +22,7 @@ ) from functions_blob_storage_operations import BLOB_STORAGE_PLUGIN_TYPE, derive_blob_endpoint_from_connection_string from functions_chart_operations import CHART_DEFAULT_ENDPOINT +from functions_control_center_dashboard import CONTROL_CENTER_ACTION_DEFAULT_ENDPOINT, CONTROL_CENTER_ACTION_TYPE from functions_databricks_operations import DATABRICKS_LEGACY_TABLE_PLUGIN_TYPE, DATABRICKS_PLUGIN_TYPE from functions_snowflake_operations import SNOWFLAKE_DEFAULT_ENDPOINT, SNOWFLAKE_PLUGIN_TYPE from functions_m365_operations import ( @@ -37,6 +38,7 @@ 'sql_query': 'sql://sql_query', 'chart': CHART_DEFAULT_ENDPOINT, 'simplechat': 'simplechat://internal', + CONTROL_CENTER_ACTION_TYPE: CONTROL_CENTER_ACTION_DEFAULT_ENDPOINT, 'search': 'internal://document-search', 'document_search': 'internal://document-search', SNOWFLAKE_PLUGIN_TYPE: SNOWFLAKE_DEFAULT_ENDPOINT, diff --git a/application/single_app/route_backend_control_center.py b/application/single_app/route_backend_control_center.py index c0bc7194c..a5814b9a7 100644 --- a/application/single_app/route_backend_control_center.py +++ b/application/single_app/route_backend_control_center.py @@ -27,6 +27,25 @@ parse_activity_filters, query_activity_rows, ) +from functions_control_center_dashboard import ( + CONTROL_CENTER_ACTION_TYPE, + DASHBOARD_INVALID_RANGE_ERROR, + DashboardStores, + append_token_usage_filters, + build_token_usage_query_context, + extract_token_filters, + get_dashboard_insights, + get_dashboard_summary, + normalize_token_filter_value, + parse_dashboard_period, +) +from functions_control_center_dashboard_chat import ( + build_dashboard_chat_readiness, + control_center_action_records, +) +from functions_action_catalog import build_accessible_action_catalog +from functions_governance import is_global_action_access_allowed +from functions_orchestration_registry import CAPABILITY_ACTION_INVOKE, capability_allowlisted from functions_settings import * from functions_logging import * from functions_activity_logging import * @@ -84,18 +103,6 @@ ACTIVITY_LOGS_MAX_PER_PAGE = 200 CONTROL_CENTER_MANAGEMENT_DEFAULT_PER_PAGE = 25 CONTROL_CENTER_MANAGEMENT_MAX_PER_PAGE = 250 -CONTROL_CENTER_DASHBOARD_CACHE_TTL_SECONDS = 90 -CONTROL_CENTER_DASHBOARD_CACHE_MAX_ENTRIES = 128 -CONTROL_CENTER_DASHBOARD_RANKING_LIMIT = 10 -CONTROL_CENTER_DASHBOARD_RANKED_FIELDS = ( - ("users", "user_id"), - ("groups", "group_id"), - ("public_workspaces", "public_workspace_id"), -) -DASHBOARD_INVALID_RANGE_ERROR = ( - "Invalid dashboard date range. Use 7, 30, or 90 days or a valid custom range of up to 366 days." -) -_control_center_dashboard_cache = {} _control_center_group_snapshot_cache = {} @@ -112,6 +119,25 @@ def _control_center_group_inventory(force_refresh=False): } return payload + +def _control_center_dashboard_stores(): + """The containers the V2 dashboard reads, bound from the application configuration.""" + return DashboardStores( + activity_logs=cosmos_activity_logs_container, + user_settings=cosmos_user_settings_container, + groups=cosmos_groups_container, + public_workspaces=cosmos_public_workspaces_container, + user_documents=cosmos_user_documents_container, + group_documents=cosmos_group_documents_container, + public_documents=cosmos_public_documents_container, + approvals=cosmos_approvals_container, + ) + + +def _dashboard_force_refresh(args): + """Whether the request asks to bypass the short-lived dashboard cache.""" + return str(args.get('force_refresh', '') or '').strip().lower() in {'1', 'true', 'yes'} + # The answers an approved ownership change gives when the group's current copy no # longer matches the request. Each is stored on the approval as its failure reason. GROUP_OWNERSHIP_CHANGED_MESSAGE = ( @@ -164,372 +190,6 @@ def __init__(self, answer): self.answer = answer -def _dashboard_cache_get(key, force_refresh=False, now=None): - """Return an unexpired dashboard cache entry, if one is available.""" - if force_refresh: - return None - current_time = time.monotonic() if now is None else now - entry = _control_center_dashboard_cache.get(key) - if not entry: - return None - expires_at, payload = entry - if expires_at <= current_time: - _control_center_dashboard_cache.pop(key, None) - return None - return payload - - -def _dashboard_cache_set(key, payload, now=None): - """Store a short-lived dashboard response and bound cache growth.""" - current_time = time.monotonic() if now is None else now - expired_keys = [ - cache_key - for cache_key, (expires_at, _) in _control_center_dashboard_cache.items() - if expires_at <= current_time - ] - for cache_key in expired_keys: - _control_center_dashboard_cache.pop(cache_key, None) - if len(_control_center_dashboard_cache) >= CONTROL_CENTER_DASHBOARD_CACHE_MAX_ENTRIES: - _control_center_dashboard_cache.pop(next(iter(_control_center_dashboard_cache))) - _control_center_dashboard_cache[key] = ( - current_time + CONTROL_CENTER_DASHBOARD_CACHE_TTL_SECONDS, - payload, - ) - - -def _dashboard_metric(current_value, previous_value=None): - """Format a current-period value with a comparable prior-period delta.""" - current_value = int(current_value or 0) - if previous_value is None: - return {"value": current_value, "delta": None, "percent_change": None} - previous_value = int(previous_value or 0) - delta = current_value - previous_value - percent_change = ( - round((delta / previous_value) * 100, 1) - if previous_value - else (0.0 if delta == 0 else None) - ) - return { - "value": current_value, - "delta": delta, - "previous": previous_value, - "percent_change": percent_change, - } - - -def _dashboard_status_counts(total, grouped_statuses, unknown_status='active'): - """Normalize stored status counts while preserving each workspace type's default.""" - counts = { - "active": 0, - "locked": 0, - "upload_disabled": 0, - "inactive": 0, - } - for row in grouped_statuses: - raw_status = row.get("status") - status = str(raw_status or "active").strip().lower() - count = int(row.get("count") or 0) - if status in {"locked", "upload_disabled", "inactive"}: - counts[status] += count - elif status != "active" and raw_status and unknown_status == 'inactive': - counts["inactive"] += count - counts["active"] = max( - int(total or 0) - counts["locked"] - counts["upload_disabled"] - counts["inactive"], - 0, - ) - return counts - - -def _dashboard_parse_period(args): - """Resolve the requested day preset or custom UTC calendar range.""" - custom_start = args.get("start_date") - custom_end = args.get("end_date") - if custom_start or custom_end: - if not custom_start or not custom_end: - raise ValueError("Both start_date and end_date are required.") - try: - start_day = datetime.strptime(custom_start, "%Y-%m-%d").date() - end_day = datetime.strptime(custom_end, "%Y-%m-%d").date() - except (TypeError, ValueError) as ex: - raise ValueError("Dates must use YYYY-MM-DD format.") from ex - if end_day < start_day: - raise ValueError("end_date must be on or after start_date.") - days = (end_day - start_day).days + 1 - if days > 366: - raise ValueError("Custom ranges may not exceed 366 days.") - else: - try: - days = int(args.get("days", 30)) - except (TypeError, ValueError) as ex: - raise ValueError("days must be 7, 30, or 90.") from ex - if days not in {7, 30, 90}: - raise ValueError("days must be 7, 30, or 90.") - end_day = datetime.utcnow().date() - start_day = end_day - timedelta(days=days - 1) - - start_date = datetime.combine(start_day, datetime.min.time()) - end_date = datetime.combine(end_day, datetime.max.time()) - period_days = (end_day - start_day).days + 1 - previous_end = start_date - timedelta(microseconds=1) - previous_start = previous_end - timedelta(days=period_days) + timedelta(microseconds=1) - return start_date, end_date, previous_start, previous_end, period_days - - -def _dashboard_query_count(container, query, parameters=None): - """Execute a cross-partition aggregate count.""" - rows = list(container.query_items( - query=query, - parameters=parameters or [], - enable_cross_partition_query=True, - )) - return int(rows[0] or 0) if rows else 0 - - -def _dashboard_count_active_users(start_date, end_date): - """Count distinct users with recorded login activity in a UTC interval. - - The Python Cosmos SDK cannot run COUNT over a DISTINCT subquery across partitions, - so the distinct user IDs are read with SELECT DISTINCT VALUE and counted here. - """ - user_ids = cosmos_activity_logs_container.query_items( - query=""" - SELECT DISTINCT VALUE c.user_id FROM c - WHERE c.activity_type = 'user_login' - AND IS_DEFINED(c.user_id) - AND c.timestamp >= @start_date - AND c.timestamp <= @end_date - """, - parameters=[ - {"name": "@start_date", "value": start_date.isoformat()}, - {"name": "@end_date", "value": end_date.isoformat()}, - ], - enable_cross_partition_query=True, - ) - return len(set(user_ids)) - - -def _dashboard_activity_count(activity_type, start_date, end_date): - """Count recorded creation events in a UTC interval.""" - return _dashboard_query_count( - cosmos_activity_logs_container, - """ - SELECT VALUE COUNT(1) FROM c - WHERE c.activity_type = @activity_type - AND ( - (c.timestamp >= @start_date AND c.timestamp <= @end_date) - OR (c.created_at >= @start_date AND c.created_at <= @end_date) - ) - """, - [ - {"name": "@activity_type", "value": activity_type}, - {"name": "@start_date", "value": start_date.isoformat()}, - {"name": "@end_date", "value": end_date.isoformat()}, - ], - ) - - -def _dashboard_document_upload_counts(start_date, end_date): - """Count document-creation activity by the recorded workspace type. - - Cross-partition GROUP BY is unavailable to the Python Cosmos SDK, so group and - public uploads are counted directly. Every other recorded or missing type is - personal, so personal is the remainder of the period total. - """ - window = """ - c.activity_type = 'document_creation' - AND ( - (c.timestamp >= @start_date AND c.timestamp <= @end_date) - OR (c.created_at >= @start_date AND c.created_at <= @end_date) - ) - """ - parameters = [ - {"name": "@start_date", "value": start_date.isoformat()}, - {"name": "@end_date", "value": end_date.isoformat()}, - ] - total = _dashboard_query_count( - cosmos_activity_logs_container, - f"SELECT VALUE COUNT(1) FROM c WHERE {window}", - parameters, - ) - counts = { - workspace_type: _dashboard_query_count( - cosmos_activity_logs_container, - f"SELECT VALUE COUNT(1) FROM c WHERE {window} AND c.workspace_type = @workspace_type", - parameters + [{"name": "@workspace_type", "value": workspace_type}], - ) - for workspace_type in ("group", "public") - } - return {"personal": max(total - counts["group"] - counts["public"], 0), **counts} - - -def _dashboard_status_rows(container): - """Tally stored status values for _dashboard_status_counts without a GROUP BY query.""" - statuses = Counter( - status if status is None or isinstance(status, str) else str(status) - for status in container.query_items( - query="SELECT VALUE c.status FROM c WHERE IS_DEFINED(c.status)", - enable_cross_partition_query=True, - ) - ) - return [{"status": status, "count": count} for status, count in statuses.items()] - - -def _dashboard_token_total(start_date, end_date, token_filters): - """Aggregate token usage without returning individual activity records.""" - where_clause, parameters = build_token_usage_query_context( - start_date, - end_date, - token_filters=token_filters, - ) - rows = list(cosmos_activity_logs_container.query_items( - query=f"SELECT VALUE SUM(c.usage.total_tokens) FROM c WHERE {where_clause}", - parameters=parameters, - enable_cross_partition_query=True, - )) - return int(rows[0] or 0) if rows else 0 - - -def _dashboard_document_failure_count(container, start_date, end_date): - """Count stored document processing states that explicitly report failure.""" - return _dashboard_query_count( - container, - """ - SELECT VALUE COUNT(1) FROM c - WHERE IS_STRING(c.status) - AND (CONTAINS(LOWER(c.status), 'failed') OR CONTAINS(LOWER(c.status), 'error')) - AND ( - (c.upload_date >= @start_date AND c.upload_date <= @end_date) - OR (c.created_at >= @start_date AND c.created_at <= @end_date) - ) - """, - [ - {"name": "@start_date", "value": start_date.isoformat()}, - {"name": "@end_date", "value": end_date.isoformat()}, - ], - ) - - -def _dashboard_document_failures(start_date, end_date): - """Count failed processing records across personal, group, and public workspaces.""" - return sum( - _dashboard_document_failure_count(container, start_date, end_date) - for container in ( - cosmos_user_documents_container, - cosmos_group_documents_container, - cosmos_public_documents_container, - ) - ) - - -def _dashboard_token_value(value): - """Return the numeric token count Cosmos SUM would add; other values count as zero.""" - if isinstance(value, bool) or not isinstance(value, (int, float)): - return 0 - return value - - -def _dashboard_ranked(totals, value_key): - """Return the largest totals, with ties ordered by ID so the ranking is stable.""" - ranked = sorted(totals.items(), key=lambda item: (-item[1], item[0])) - return [ - {"id": entity_id, value_key: int(total)} - for entity_id, total in ranked[:CONTROL_CENTER_DASHBOARD_RANKING_LIMIT] - ] - - -def _dashboard_token_insights(start_date, end_date, token_filters): - """Total filtered token usage by day and model and rank the largest consumers. - - The Python Cosmos SDK cannot run cross-partition GROUP BY, so one narrow projection - of the filtered token records is streamed and totalled here. - """ - where_clause, parameters = build_token_usage_query_context( - start_date, - end_date, - token_filters=token_filters, - ) - by_model = defaultdict(int) - consumers = {key: defaultdict(int) for key, _ in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS} - rows = cosmos_activity_logs_container.query_items( - query=f""" - SELECT c.timestamp, - c.usage.model AS model, - c.usage.total_tokens AS tokens, - c.user_id AS user_id, - c.workspace_context.group_id AS group_id, - c.workspace_context.public_workspace_id AS public_workspace_id - FROM c - WHERE {where_clause} - """, - parameters=parameters, - enable_cross_partition_query=True, - ) - for row in rows: - tokens = _dashboard_token_value(row.get("tokens")) - timestamp = row.get("timestamp") - if isinstance(timestamp, str) and "model" in row: - by_model[(timestamp[:10], str(row.get("model") or "Unknown model"))] += tokens - for key, field in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS: - entity_id = row.get(field) - if isinstance(entity_id, str) and entity_id: - consumers[key][entity_id] += tokens - token_usage_by_model = [ - {"date": date, "model": model, "tokens": int(tokens)} - for (date, model), tokens in sorted(by_model.items()) - ] - return token_usage_by_model, { - key: _dashboard_ranked(totals, "tokens") for key, totals in consumers.items() - } - - -def _dashboard_activity_insights(start_date, end_date): - """Rank recorded activity and build the UTC login heatmap from one projection. - - Each heatmap cell totals every login in the period for one weekday and hour. - """ - actors = {key: defaultdict(int) for key, _ in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS} - logins = Counter() - rows = cosmos_activity_logs_container.query_items( - query=""" - SELECT c.timestamp, - c.activity_type, - c.user_id AS user_id, - c.workspace_context.group_id AS group_id, - c.workspace_context.public_workspace_id AS public_workspace_id - FROM c - WHERE c.timestamp >= @start_date - AND c.timestamp <= @end_date - """, - parameters=[ - {"name": "@start_date", "value": start_date.isoformat()}, - {"name": "@end_date", "value": end_date.isoformat()}, - ], - enable_cross_partition_query=True, - ) - for row in rows: - for key, field in CONTROL_CENTER_DASHBOARD_RANKED_FIELDS: - entity_id = row.get(field) - if isinstance(entity_id, str) and entity_id: - actors[key][entity_id] += 1 - timestamp = row.get("timestamp") - if row.get("activity_type") != "user_login" or not isinstance(timestamp, str): - continue - try: - weekday = datetime.strptime(timestamp[:10], "%Y-%m-%d").weekday() - hour = int(timestamp[11:13]) - except ValueError: - continue - if 0 <= hour <= 23: - logins[(weekday, hour)] += 1 - login_cells = [ - {"weekday": weekday, "hour": hour, "count": count} - for (weekday, hour), count in sorted(logins.items()) - ] - return { - key: _dashboard_ranked(counts, "activity_count") for key, counts in actors.items() - }, login_cells - - def parse_control_center_management_pagination(request_args): """Parse shared Control Center management pagination query parameters.""" try: @@ -711,103 +371,6 @@ def get_public_workspace_name_map(public_workspace_ids): return public_workspace_names -def normalize_token_filter_value(value): - """Normalize optional token filter values from query params or request JSON.""" - if value is None: - return None - - normalized = str(value).strip() - if not normalized or normalized.lower() == 'all': - return None - - return normalized - - -def extract_token_filters(source): - """Extract supported token filters from a query string or JSON payload.""" - if not source: - return {} - - token_source = source - nested_filters = source.get('token_filters') if hasattr(source, 'get') else None - if isinstance(nested_filters, dict): - token_source = nested_filters - - token_filters = { - 'user_id': normalize_token_filter_value(token_source.get('user_id')), - 'workspace_type': normalize_token_filter_value(token_source.get('workspace_type')), - 'group_id': normalize_token_filter_value(token_source.get('group_id')), - 'public_workspace_id': normalize_token_filter_value(token_source.get('public_workspace_id')), - 'model': normalize_token_filter_value(token_source.get('model')), - 'token_type': normalize_token_filter_value(token_source.get('token_type')) - } - - if token_filters['workspace_type'] not in {None, 'personal', 'group', 'public'}: - token_filters['workspace_type'] = None - - if token_filters['token_type'] not in {None, 'embedding', 'chat', 'web_search'}: - token_filters['token_type'] = None - - if token_filters['group_id'] and not token_filters['workspace_type']: - token_filters['workspace_type'] = 'group' - - if token_filters['public_workspace_id'] and not token_filters['workspace_type']: - token_filters['workspace_type'] = 'public' - - return token_filters - - -def append_token_usage_filters(query_conditions, parameters, token_filters): - """Append optional token usage filters to the Cosmos query state.""" - if not token_filters: - return - - user_id = token_filters.get('user_id') - if user_id: - query_conditions.append("c.user_id = @token_user_id") - parameters.append({"name": "@token_user_id", "value": user_id}) - - workspace_type = token_filters.get('workspace_type') - if workspace_type: - query_conditions.append("c.workspace_type = @token_workspace_type") - parameters.append({"name": "@token_workspace_type", "value": workspace_type}) - - group_id = token_filters.get('group_id') - if group_id: - query_conditions.append("c.workspace_context.group_id = @token_group_id") - parameters.append({"name": "@token_group_id", "value": group_id}) - - public_workspace_id = token_filters.get('public_workspace_id') - if public_workspace_id: - query_conditions.append("c.workspace_context.public_workspace_id = @token_public_workspace_id") - parameters.append({"name": "@token_public_workspace_id", "value": public_workspace_id}) - - model = token_filters.get('model') - if model: - query_conditions.append("c.usage.model = @token_model") - parameters.append({"name": "@token_model", "value": model}) - - token_type = token_filters.get('token_type') - if token_type: - query_conditions.append("c.token_type = @token_type") - parameters.append({"name": "@token_type", "value": token_type}) - - -def build_token_usage_query_context(start_date, end_date, token_filters=None): - """Build shared WHERE-clause state for token usage queries.""" - parameters = [ - {"name": "@start_date", "value": start_date.isoformat()}, - {"name": "@end_date", "value": end_date.isoformat()} - ] - query_conditions = [ - "c.activity_type = 'token_usage'", - "((c.timestamp >= @start_date AND c.timestamp <= @end_date) OR (c.created_at >= @start_date AND c.created_at <= @end_date))" - ] - - append_token_usage_filters(query_conditions, parameters, token_filters or {}) - return " AND ".join(query_conditions), parameters - - def get_distinct_token_filter_values(query): """Run a DISTINCT VALUE token filter query and return sorted non-empty values.""" try: @@ -6887,201 +6450,19 @@ def api_delete_public_workspace_admin(workspace_id): def api_v2_control_center_dashboard_summary(): """Return cached aggregate dashboard metrics for the requested UTC period.""" try: - start_date, end_date, previous_start, previous_end, period_days = ( - _dashboard_parse_period(request.args) - ) + period = parse_dashboard_period(request.args) except ValueError: return jsonify({'error': DASHBOARD_INVALID_RANGE_ERROR}), 400 - token_filters = extract_token_filters(request.args) - cache_key = ( - start_date.isoformat(), - end_date.isoformat(), - json.dumps(token_filters, sort_keys=True), - ) - force_refresh = request.args.get('force_refresh', '').strip().lower() in { - '1', 'true', 'yes' - } - cached = _dashboard_cache_get(cache_key, force_refresh=force_refresh) - if cached is not None: - return jsonify({**cached, 'cached': True}) - try: - total_users = _dashboard_query_count( - cosmos_user_settings_container, - 'SELECT VALUE COUNT(1) FROM c', - ) - blocked_users = _dashboard_query_count( - cosmos_user_settings_container, - """ - SELECT VALUE COUNT(1) FROM c - WHERE c.settings.access.status = 'deny' - """, - ) - - groups_total = _dashboard_query_count( - cosmos_groups_container, - 'SELECT VALUE COUNT(1) FROM c', - ) - group_statuses = _dashboard_status_rows(cosmos_groups_container) - groups_by_status = _dashboard_status_counts(groups_total, group_statuses) - - workspaces_total = _dashboard_query_count( - cosmos_public_workspaces_container, - 'SELECT VALUE COUNT(1) FROM c', - ) - workspace_statuses = _dashboard_status_rows(cosmos_public_workspaces_container) - workspaces_by_status = _dashboard_status_counts( - workspaces_total, - workspace_statuses, - unknown_status='inactive', - ) - - active_users = _dashboard_count_active_users(start_date, end_date) - previous_active_users = _dashboard_count_active_users( - previous_start, - previous_end, - ) - current_dau = _dashboard_count_active_users( - datetime.combine(end_date.date(), datetime.min.time()), - end_date, - ) - current_wau = _dashboard_count_active_users( - datetime.combine(end_date.date() - timedelta(days=6), datetime.min.time()), - end_date, - ) - current_mau = _dashboard_count_active_users( - datetime.combine(end_date.date() - timedelta(days=29), datetime.min.time()), - end_date, - ) - - conversations = _dashboard_activity_count( - 'conversation_creation', - start_date, - end_date, - ) - previous_conversations = _dashboard_activity_count( - 'conversation_creation', - previous_start, - previous_end, + payload, cached = get_dashboard_summary( + _control_center_dashboard_stores(), + period, + extract_token_filters(request.args), + force_refresh=_dashboard_force_refresh(request.args), + log=log_event, ) - document_uploads = _dashboard_document_upload_counts(start_date, end_date) - previous_document_uploads = _dashboard_document_upload_counts( - previous_start, - previous_end, - ) - current_token_total = _dashboard_token_total( - start_date, - end_date, - token_filters, - ) - previous_token_total = _dashboard_token_total( - previous_start, - previous_end, - token_filters, - ) - - try: - processing_failures = _dashboard_document_failures(start_date, end_date) - previous_processing_failures = _dashboard_document_failures( - previous_start, - previous_end, - ) - failure_metric = { - **_dashboard_metric( - processing_failures, - previous_processing_failures, - ), - 'available': True, - } - except Exception as ex: - log_event( - message='[CONTROL_CENTER] Document failure metrics are unavailable.', - extra={'error_type': type(ex).__name__}, - level=logging.WARNING, - ) - failure_metric = { - 'value': None, - 'delta': None, - 'percent_change': None, - 'available': False, - } - - try: - pending_approvals = _dashboard_query_count( - cosmos_approvals_container, - """ - SELECT VALUE COUNT(1) FROM c - WHERE c.status = 'pending' - """, - ) - except Exception as ex: - log_event( - message='[CONTROL_CENTER] Pending approval count is unavailable.', - extra={'error_type': type(ex).__name__}, - level=logging.WARNING, - ) - pending_approvals = None - - total_document_uploads = sum(document_uploads.values()) - previous_total_document_uploads = sum(previous_document_uploads.values()) - response_data = { - 'period': { - 'start_date': start_date.date().isoformat(), - 'end_date': end_date.date().isoformat(), - 'days': period_days, - 'timezone': 'UTC', - }, - 'refreshed_at': datetime.utcnow().isoformat() + 'Z', - 'users': { - 'total': _dashboard_metric(total_users), - 'active': _dashboard_metric(active_users, previous_active_users), - 'dau': _dashboard_metric(current_dau), - 'wau': _dashboard_metric(current_wau), - 'mau': _dashboard_metric(current_mau), - 'blocked': _dashboard_metric(blocked_users), - }, - 'groups': { - 'total': _dashboard_metric(groups_total), - 'by_status': { - status: _dashboard_metric(count) - for status, count in groups_by_status.items() - }, - }, - 'public_workspaces': { - 'total': _dashboard_metric(workspaces_total), - 'by_status': { - status: _dashboard_metric(count) - for status, count in workspaces_by_status.items() - }, - }, - 'conversations': _dashboard_metric( - conversations, - previous_conversations, - ), - 'document_uploads': { - 'total': _dashboard_metric( - total_document_uploads, - previous_total_document_uploads, - ), - 'by_workspace_type': { - workspace_type: _dashboard_metric( - count, - previous_document_uploads.get(workspace_type, 0), - ) - for workspace_type, count in document_uploads.items() - }, - }, - 'document_processing_failures': failure_metric, - 'tokens': _dashboard_metric(current_token_total, previous_token_total), - 'pending_approvals': ( - _dashboard_metric(pending_approvals) - if pending_approvals is not None else None - ), - 'status_history_available': False, - } - _dashboard_cache_set(cache_key, response_data) - return jsonify({**response_data, 'cached': False}) + return jsonify({**payload, 'cached': cached}) except Exception as ex: log_event( message='[CONTROL_CENTER] Dashboard summary query failed.', @@ -7095,52 +6476,21 @@ def api_v2_control_center_dashboard_summary(): @login_required @control_center_required('dashboard') def api_v2_control_center_dashboard_insights(): - """Return aggregate-only dashboard insights available from recorded activity fields.""" + """Return dashboard chart series and rankings, with names, for the requested UTC period.""" try: - start_date, end_date, _, _, period_days = _dashboard_parse_period(request.args) + period = parse_dashboard_period(request.args) except ValueError: return jsonify({'error': DASHBOARD_INVALID_RANGE_ERROR}), 400 - token_filters = extract_token_filters(request.args) - cache_key = ( - 'insights', - start_date.isoformat(), - end_date.isoformat(), - json.dumps(token_filters, sort_keys=True), - ) - force_refresh = request.args.get('force_refresh', '').strip().lower() in { - '1', 'true', 'yes' - } - cached = _dashboard_cache_get(cache_key, force_refresh=force_refresh) - if cached is not None: - return jsonify({**cached, 'cached': True}) - try: - token_usage_by_model, top_tokens = _dashboard_token_insights( - start_date, - end_date, - token_filters, + payload, cached = get_dashboard_insights( + _control_center_dashboard_stores(), + period, + extract_token_filters(request.args), + force_refresh=_dashboard_force_refresh(request.args), + log=log_event, ) - top_activity, login_cells = _dashboard_activity_insights(start_date, end_date) - - response_data = { - 'period': { - 'start_date': start_date.date().isoformat(), - 'end_date': end_date.date().isoformat(), - 'days': period_days, - 'timezone': 'UTC', - }, - 'token_usage_by_model': token_usage_by_model, - 'top_tokens': top_tokens, - 'top_activity': top_activity, - 'login_heatmap': { - 'weekday_convention': 'Monday=0 through Sunday=6', - 'timezone': 'UTC', - 'cells': login_cells, - }, - } - _dashboard_cache_set(cache_key, response_data) - return jsonify({**response_data, 'cached': False}) + return jsonify({**payload, 'cached': cached}) except Exception as ex: log_event( message='[CONTROL_CENTER] Dashboard insights query failed.', @@ -7149,6 +6499,57 @@ def api_v2_control_center_dashboard_insights(): ) return jsonify({'error': 'Failed to retrieve dashboard insights.'}), 500 + @bp.route('/api/v2/control-center/dashboard/chat-readiness', methods=['GET']) + @swagger_route(security=get_auth_security()) + @login_required + @control_center_required('dashboard') + def api_v2_control_center_dashboard_chat_readiness(): + """Report what "Chat with this dashboard" needs and whether each requirement is met. + + The answer lists settings states and the caller's own access, never settings values, + and links to Admin Settings only for administrators, who can open those pages. + """ + try: + settings = get_settings() + user = session.get('user') or {} + roles = [role for role in (user.get('roles') or []) if isinstance(role, str)] + user_id = get_current_user_id() + stored_actions = control_center_action_records( + cosmos_global_actions_container.query_items( + query=( + "SELECT c.id, c.name, c.display_name, c.displayName, c.type, c.is_enabled FROM c " + "WHERE IS_STRING(c.type) AND LOWER(c.type) = @action_type" + ), + parameters=[{'name': '@action_type', 'value': CONTROL_CENTER_ACTION_TYPE}], + enable_cross_partition_query=True, + ), + CONTROL_CENTER_ACTION_TYPE, + ) + available_actions = [ + action + for action in build_accessible_action_catalog(user_id, settings=settings, user_roles=roles) + if str(action.get('type') or '').strip().lower() == CONTROL_CENTER_ACTION_TYPE + ] if user_id else [] + readiness = build_dashboard_chat_readiness( + settings=settings, + user_roles=roles, + is_admin='Admin' in roles, + stored_actions=stored_actions, + available_actions=available_actions, + governance_allows=lambda action: is_global_action_access_allowed(user_id, action), + action_invoke_allowlisted=capability_allowlisted(settings, CAPABILITY_ACTION_INVOKE), + ) + response = jsonify(readiness) + response.headers['Cache-Control'] = 'no-store' + return response + except Exception as ex: + log_event( + message='[CONTROL_CENTER] Dashboard chat readiness could not be checked.', + extra={'error_type': type(ex).__name__}, + level=logging.ERROR, + ) + return jsonify({'error': 'Dashboard chat requirements could not be checked. Try again.'}), 500 + # Activity Trends API @bp.route('/api/admin/control-center/activity-trends', methods=['GET']) @swagger_route(security=get_auth_security()) @@ -7498,163 +6899,6 @@ def api_export_activity_trends(): debug_print(f"Error exporting activity trends: {e}") return jsonify({'error': 'Failed to export data'}), 500 - @bp.route('/api/admin/control-center/activity-trends/chat', methods=['POST']) - @swagger_route(security=get_auth_security()) - @login_required - @control_center_required('dashboard') - def api_chat_activity_trends(): - """ - Create a new chat conversation with activity trends data as CSV message. - """ - try: - data = request.get_json() or {} - - # Parse request parameters - charts = data.get('charts', ['logins', 'chats', 'documents']) # Default to all charts - time_window = data.get('time_window', '30') # Default to 30 days - start_date = data.get('start_date') # For custom range - end_date = data.get('end_date') # For custom range - token_filters = extract_token_filters(data) - - # Determine date range - if time_window == 'custom' and start_date and end_date: - try: - start_date_obj = datetime.fromisoformat(start_date.replace('Z', '+00:00') if 'Z' in start_date else start_date) - end_date_obj = datetime.fromisoformat(end_date.replace('Z', '+00:00') if 'Z' in end_date else end_date) - end_date_obj = end_date_obj.replace(hour=23, minute=59, second=59, microsecond=999999) - except ValueError: - return jsonify({'error': 'Invalid date format'}), 400 - else: - # Use predefined ranges - days = int(time_window) if time_window.isdigit() else 30 - end_date_obj = datetime.now().replace(hour=23, minute=59, second=59, microsecond=999999) - start_date_obj = end_date_obj - timedelta(days=days-1) - - # Get activity data using existing function - activity_data = get_activity_trends_data( - start_date_obj, - end_date_obj, - token_filters=token_filters - ) - - # Prepare CSV data - csv_rows = [] - csv_rows.append(['Date', 'Chart Type', 'Activity Count']) - - # Process each requested chart type - for chart_type in charts: - if chart_type in activity_data: - chart_data = activity_data[chart_type] - # Sort dates for consistent output - sorted_dates = sorted(chart_data.keys()) - - for date_key in sorted_dates: - count = chart_data[date_key] - chart_display_name = { - 'logins': 'Logins', - 'chats': 'Chats', - 'documents': 'Documents', - 'personal_documents': 'Personal Documents', - 'group_documents': 'Group Documents', - 'public_documents': 'Public Documents' - }.get(chart_type, chart_type.title()) - - csv_rows.append([date_key, chart_display_name, count]) - - # Generate CSV content - import io - import csv - output = io.StringIO() - writer = csv.writer(output) - writer.writerows(csv_rows) - csv_content = output.getvalue() - output.close() - - # Get current user info - user_id = session.get('user_id') - user_email = session.get('email') - user_display_name = session.get('display_name', user_email) - - if not user_id: - return jsonify({'error': 'User not authenticated'}), 401 - - # Create new conversation - conversation_id = str(uuid.uuid4()) - timestamp = datetime.now(timezone.utc).isoformat() - - # Generate descriptive title with date range - if time_window == 'custom': - date_range = f"{start_date} to {end_date}" - else: - date_range = f"Last {time_window} Days" - - charts_text = ", ".join([c.title() for c in charts]) - conversation_title = f"Activity Trends - {charts_text} ({date_range})" - - # Create conversation document - conversation_doc = { - "id": conversation_id, - "title": conversation_title, - "user_id": user_id, - "user_email": user_email, - "user_display_name": user_display_name, - "created": timestamp, - "last_updated": timestamp, - "messages": [], - "system_message": "You are analyzing activity trends data from a control center dashboard. The user has provided activity data as a CSV file. Please analyze the data and provide insights about user activity patterns, trends, and any notable observations.", - "message_count": 0, - "settings": { - "model": "gpt-4o", - "temperature": 0.7, - "max_tokens": 4000 - } - } - - # Create the initial message with CSV data (simulate file upload) - message_id = str(uuid.uuid4()) - csv_filename = f"activity_trends_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" - - # Create message with file attachment structure - initial_message = { - "id": message_id, - "role": "user", - "content": f"Please analyze this activity trends data from our system dashboard. The data covers {date_range} and includes {charts_text} activity.", - "timestamp": timestamp, - "files": [{ - "name": csv_filename, - "type": "text/csv", - "size": len(csv_content.encode('utf-8')), - "content": csv_content, - "id": str(uuid.uuid4()) - }] - } - - conversation_doc["messages"].append(initial_message) - conversation_doc["message_count"] = 1 - - # Save conversation to database - cosmos_conversations_container.create_item(conversation_doc) - - # Log the activity - log_event("[CONTROL_CENTER] Activity Trends Chat Created", { - "conversation_id": conversation_id, - "user_id": user_id, - "charts": charts, - "time_window": time_window, - "date_range": date_range - }) - - return jsonify({ - 'success': True, - 'conversation_id': conversation_id, - 'conversation_title': conversation_title, - 'redirect_url': f'/chat/{conversation_id}' - }), 200 - - except Exception as e: - debug_print(f"Error creating activity trends chat: {e}") - return jsonify({'error': 'Failed to create chat conversation'}), 500 - # Data Refresh API @bp.route('/api/admin/control-center/refresh', methods=['POST']) @swagger_route(security=get_auth_security()) diff --git a/application/single_app/route_backend_orchestration.py b/application/single_app/route_backend_orchestration.py index 9f21ad4d2..300b95fe8 100644 --- a/application/single_app/route_backend_orchestration.py +++ b/application/single_app/route_backend_orchestration.py @@ -1356,6 +1356,7 @@ def _validate_retry_context(record, user_id, settings, *, preparing=False): ) if identity.get('user_enable_agents', True) else [] actions = resolve_action_catalog( user_id, seeds=seeds, settings=settings, user_groups=seeds.get('active_group_ids') or None, + user_roles=identity.get('user_roles') or [], ) allowed_urls = revision_allowed_urls(record) workflow_planning = _current_workflow_planning(record, user_id, settings, conversation_id) @@ -2060,6 +2061,7 @@ def generate(): action_catalog = resolve_action_catalog( user_id, seeds=seeds, settings=settings, user_groups=seeds.get('active_group_ids') or None, + user_roles=identity.get('user_roles') or [], ) planning_identity = dict(identity) if isinstance(seeds.get('agent'), dict) and seeds['agent'].get('name'): @@ -2639,6 +2641,7 @@ def orchestration_run(): action_catalog = resolve_action_catalog( user_id, seeds=seeds, settings=settings, user_groups=seeds.get('active_group_ids') or None, + user_roles=identity.get('user_roles') or [], ) effective_plan = apply_plan_edits( deepcopy(plan), data.get('edits', record.get('edit_narrowing')), diff --git a/application/single_app/route_backend_plugins.py b/application/single_app/route_backend_plugins.py index 6933335c0..19a2be111 100644 --- a/application/single_app/route_backend_plugins.py +++ b/application/single_app/route_backend_plugins.py @@ -186,6 +186,11 @@ ) from functions_m365_operations import M365_PLUGIN_TYPES, get_m365_action_definition, get_m365_default_config from functions_simplechat_operations import SIMPLECHAT_DEFAULT_ENDPOINT, SIMPLECHAT_PLUGIN_TYPE +from functions_control_center_dashboard import ( + CONTROL_CENTER_ACTION_DEFAULT_DESCRIPTION, + CONTROL_CENTER_ACTION_DEFAULT_ENDPOINT, + CONTROL_CENTER_ACTION_TYPE, +) from functions_agent_delegation import ( AGENT_DEFAULT_ENDPOINT, AGENT_PLUGIN_TYPE, @@ -395,6 +400,15 @@ def _apply_plugin_runtime_defaults(plugin_payload): auth['type'] = 'user' plugin_payload['auth'] = auth + elif plugin_type == CONTROL_CENTER_ACTION_TYPE: + # Runs as the signed-in Control Center viewer; there is no connection to configure. + if not str(plugin_payload.get('endpoint') or '').strip(): + plugin_payload['endpoint'] = CONTROL_CENTER_ACTION_DEFAULT_ENDPOINT + plugin_payload['auth'] = {'type': 'user'} + # Orchestration chooses actions by their description, so a blank one is never saved. + if not str(plugin_payload.get('description') or '').strip(): + plugin_payload['description'] = CONTROL_CENTER_ACTION_DEFAULT_DESCRIPTION + elif plugin_type == AGENT_PLUGIN_TYPE: if not str(plugin_payload.get('endpoint') or '').strip(): plugin_payload['endpoint'] = AGENT_DEFAULT_ENDPOINT diff --git a/application/single_app/semantic_kernel_plugins/control_center_plugin.py b/application/single_app/semantic_kernel_plugins/control_center_plugin.py new file mode 100644 index 000000000..7e347ee84 --- /dev/null +++ b/application/single_app/semantic_kernel_plugins/control_center_plugin.py @@ -0,0 +1,402 @@ +# control_center_plugin.py +"""Read-only Control Center insights for chat, answered for the signed-in Control Center viewer. + +Every function re-checks that the person the action runs for may view the Control Center +dashboard: an Admin, or a ControlCenterAdmin or ControlCenterDashboardReader where the +deployment requires those roles. Agent and orchestration runs carry the authenticated user's +session into the tool call; a scheduled run has none, so it is refused. + +The figures come from the same aggregation as the V2 dashboard, so an answer in chat matches +the dashboard for the same dates. Nothing here writes data or accepts query text: each function +runs fixed, parameterized queries and returns bounded rows that orchestration can chart. +""" + +import logging +import re +from typing import Annotated, Any, Callable, Dict, List, Optional + +from semantic_kernel.functions import kernel_function + +from functions_appinsights import log_event +from functions_control_center_dashboard import ( + CONTROL_CENTER_ACTION_DEFAULT_DESCRIPTION, + CONTROL_CENTER_ACTION_TYPE, + CONTROL_CENTER_DASHBOARD_RANKING_LIMIT, + DASHBOARD_ENTITY_KINDS, + DASHBOARD_REPORT_MAX_ROWS, + DASHBOARD_TOKEN_GROUPINGS, + DASHBOARD_TOKEN_TYPES, + DASHBOARD_WORKSPACE_TYPES, + DashboardStores, + daily_activity_report, + extract_token_filters, + find_entities as search_dashboard_entities, + get_dashboard_summary as load_dashboard_summary, + resolve_entity_names, + resolve_requested_period, + sign_in_pattern_report, + summary_report_rows, + token_usage_report, + top_activity_report, +) +from semantic_kernel_plugins.base_plugin import BasePlugin +from semantic_kernel_plugins.plugin_invocation_logger import plugin_function_logger + + +CONTROL_CENTER_PLUGIN_TYPE = CONTROL_CENTER_ACTION_TYPE +CONTROL_CENTER_ACCESS_DENIED = ( + "Only Control Center administrators and dashboard readers can use the Control Center action." +) +CONTROL_CENTER_READ_FAILED = "Control Center data could not be read. Try again in a moment." +_ENTITY_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:@-]{0,127}$") +_MODEL_MAX_LENGTH = 200 +_DATE_PARAMETER = "Start date, YYYY-MM-DD (UTC). Leave empty and use days for a range ending today." +_END_PARAMETER = "End date, YYYY-MM-DD (UTC). Required when start_date is given." +_DAYS_PARAMETER = "Number of whole days ending today (UTC), 1 to 366. Ignored when dates are given." +_FUNCTION_DEFINITIONS = ( + ( + "get_dashboard_summary", + "Headline Control Center figures for a UTC date range compared with the previous equal " + "range: users, sign-ins, daily/weekly/monthly active users, conversations, document " + "uploads, processing failures, tokens, groups, public workspaces and pending approvals.", + ), + ( + "get_daily_activity", + "Daily sign-ins, conversations created and document uploads by workspace type, optionally " + "for one user or workspace.", + ), + ( + "get_token_usage", + "Model token usage totalled by day, model, usage type, user, group or public workspace, " + "optionally filtered.", + ), + ( + "get_top_activity", + "The users, groups or public workspaces with the most recorded actions in a date range.", + ), + ( + "get_sign_in_pattern", + "Sign-ins totalled by UTC weekday and hour, showing recurring busy times.", + ), + ( + "find_entities", + "Look up the ID of a user by name or email, or of a group or public workspace by name.", + ), +) + + +def _default_stores() -> DashboardStores: + # The storage handles belong to the configured application; they are bound when a tool runs + # so plugin type discovery can import this module without initializing Azure clients. + import config + + return DashboardStores( + activity_logs=config.cosmos_activity_logs_container, + user_settings=config.cosmos_user_settings_container, + groups=config.cosmos_groups_container, + public_workspaces=config.cosmos_public_workspaces_container, + user_documents=config.cosmos_user_documents_container, + group_documents=config.cosmos_group_documents_container, + public_documents=config.cosmos_public_documents_container, + approvals=config.cosmos_approvals_container, + ) + + +def _default_capabilities() -> Dict[str, bool]: + # Imported at the call for the same reason: the authentication module loads the bootstrap. + from functions_authentication import get_request_control_center_capabilities + + return get_request_control_center_capabilities() + + +def _entity_id(value: Any, label: str) -> str: + """Accept an empty filter or an ID; names must be looked up with find_entities first.""" + text = str(value or "").strip() + if not text or text.lower() == "all": + return "" + if not _ENTITY_ID_PATTERN.fullmatch(text): + raise ValueError( + f"{label} must be an ID. Use find_entities to look up the ID for {text[:60]!r} first." + ) + return text + + +def _choice(value: Any, allowed, label: str) -> str: + text = str(value or "").strip().lower() + if not text or text == "all": + return "" + if text not in allowed: + raise ValueError(f"{label} must be one of: {', '.join(allowed)}.") + return text + + +def _limit(value: Any, default: int = CONTROL_CENTER_DASHBOARD_RANKING_LIMIT) -> int: + if value is None or (isinstance(value, str) and not value.strip()): + return default + if isinstance(value, bool): + raise ValueError(f"limit must be a whole number from 1 to {DASHBOARD_REPORT_MAX_ROWS}.") + try: + number = int(str(value).strip()) + except (TypeError, ValueError) as ex: + raise ValueError(f"limit must be a whole number from 1 to {DASHBOARD_REPORT_MAX_ROWS}.") from ex + if not 1 <= number <= DASHBOARD_REPORT_MAX_ROWS: + raise ValueError(f"limit must be a whole number from 1 to {DASHBOARD_REPORT_MAX_ROWS}.") + return number + + +class ControlCenterPlugin(BasePlugin): + """Answer questions about SimpleChat usage from the Control Center dashboard's data.""" + + def __init__( + self, + manifest: Optional[Dict[str, Any]] = None, + *, + stores_factory: Optional[Callable[[], DashboardStores]] = None, + capabilities: Optional[Callable[[], Dict[str, bool]]] = None, + ): + super().__init__(manifest) + self.manifest = manifest or {} + self._stores_factory = stores_factory or _default_stores + self._capabilities = capabilities or _default_capabilities + + @property + def display_name(self) -> str: + return "Control Center" + + @property + def metadata(self) -> Dict[str, Any]: + return { + "name": self.manifest.get("name", CONTROL_CENTER_PLUGIN_TYPE), + "type": CONTROL_CENTER_PLUGIN_TYPE, + "description": CONTROL_CENTER_ACTION_DEFAULT_DESCRIPTION, + "methods": [ + { + "name": name, + "description": description, + "parameters": [], + "returns": {"type": "dict", "description": description}, + } + for name, description in _FUNCTION_DEFINITIONS + ], + } + + def get_functions(self) -> List[str]: + return [name for name, _ in _FUNCTION_DEFINITIONS] + + def _run(self, operation: str, callback: Callable[[DashboardStores], Dict[str, Any]]) -> Dict[str, Any]: + try: + capabilities = self._capabilities() or {} + except Exception as ex: + log_event( + "[CONTROL_CENTER_ACTION] Control Center access could not be checked.", + extra={"operation": operation, "error_type": type(ex).__name__}, + level=logging.WARNING, + ) + capabilities = {} + if not capabilities.get("can_view_dashboard"): + log_event( + "[CONTROL_CENTER_ACTION] Refused a Control Center tool call without dashboard access.", + extra={"operation": operation}, + level=logging.WARNING, + ) + return {"success": False, "error": CONTROL_CENTER_ACCESS_DENIED, "error_type": "permission"} + try: + payload = callback(self._stores_factory()) + return {"success": True, **payload} + except ValueError as ex: + return {"success": False, "error": str(ex), "error_type": "validation"} + except Exception as ex: + log_event( + "[CONTROL_CENTER_ACTION] A Control Center tool call failed.", + extra={"operation": operation, "error_type": type(ex).__name__}, + level=logging.ERROR, + ) + return {"success": False, "error": CONTROL_CENTER_READ_FAILED, "error_type": "unexpected"} + + @staticmethod + def _describe_filters(stores: DashboardStores, filters: Dict[str, Any]) -> Dict[str, Any]: + """Echo the applied filters with names, so an answer can say whose data it describes.""" + lookups = { + "users": [filters["user_id"]] if filters.get("user_id") else [], + "groups": [filters["group_id"]] if filters.get("group_id") else [], + "public_workspaces": [filters["public_workspace_id"]] if filters.get("public_workspace_id") else [], + } + names = resolve_entity_names(stores, lookups, log=log_event) + described = {} + for key, kind, field in ( + ("user", "users", "user_id"), + ("group", "groups", "group_id"), + ("public_workspace", "public_workspaces", "public_workspace_id"), + ): + entity_id = filters.get(field) + if entity_id: + described[key] = {"id": entity_id, "name": names[kind].get(entity_id, {}).get("name", "")} + for key in ("workspace_type", "model", "token_type"): + if filters.get(key): + described[key] = filters[key] + return described + + @plugin_function_logger("ControlCenterPlugin") + @kernel_function(description=_FUNCTION_DEFINITIONS[0][1]) + def get_dashboard_summary( + self, + start_date: Annotated[str, _DATE_PARAMETER] = "", + end_date: Annotated[str, _END_PARAMETER] = "", + days: Annotated[int, _DAYS_PARAMETER] = 30, + ) -> dict: + def run(stores): + period = resolve_requested_period(start_date, end_date, days) + summary, _ = load_dashboard_summary(stores, period, {}, log=log_event) + return { + "period": period.describe(), + "previous_period": period.describe_previous(), + "rows": summary_report_rows(summary), + "notes": [ + "Changes compare the period with the equally long period just before it.", + "Directory figures (users, groups, public workspaces, approvals) are current totals.", + ], + } + + return self._run("get_dashboard_summary", run) + + @plugin_function_logger("ControlCenterPlugin") + @kernel_function(description=_FUNCTION_DEFINITIONS[1][1]) + def get_daily_activity( + self, + start_date: Annotated[str, _DATE_PARAMETER] = "", + end_date: Annotated[str, _END_PARAMETER] = "", + days: Annotated[int, _DAYS_PARAMETER] = 30, + user: Annotated[str, "Optional user ID from find_entities, to show one person's activity."] = "", + workspace_type: Annotated[str, "Optional workspace type: personal, group or public."] = "", + group: Annotated[str, "Optional group ID from find_entities."] = "", + public_workspace: Annotated[str, "Optional public workspace ID from find_entities."] = "", + ) -> dict: + def run(stores): + period = resolve_requested_period(start_date, end_date, days) + filters = { + "user_id": _entity_id(user, "user"), + "workspace_type": _choice(workspace_type, DASHBOARD_WORKSPACE_TYPES, "workspace_type"), + "group_id": _entity_id(group, "group"), + "public_workspace_id": _entity_id(public_workspace, "public_workspace"), + } + rows = daily_activity_report(stores, period, filters, log=log_event) + notes = ["Each day counts events recorded on that UTC date."] + if filters["workspace_type"] or filters["group_id"] or filters["public_workspace_id"]: + notes.append( + "Sign-ins are not recorded against a workspace, so the workspace filters narrow " + "conversations and uploads only; sign_ins honor the user filter alone." + ) + totals = { + field: sum(int(row.get(field) or 0) for row in rows) + for field in ("sign_ins", "conversations_created", "uploads_personal", "uploads_group", "uploads_public") + } + return { + "period": period.describe(), + "filters": self._describe_filters(stores, filters), + "totals": totals, + "rows": rows, + "notes": notes, + } + + return self._run("get_daily_activity", run) + + @plugin_function_logger("ControlCenterPlugin") + @kernel_function(description=_FUNCTION_DEFINITIONS[2][1]) + def get_token_usage( + self, + group_by: Annotated[str, "How to total tokens: day, model, token_type, user, group or public_workspace."] = "day", + start_date: Annotated[str, _DATE_PARAMETER] = "", + end_date: Annotated[str, _END_PARAMETER] = "", + days: Annotated[int, _DAYS_PARAMETER] = 30, + user: Annotated[str, "Optional user ID from find_entities."] = "", + workspace_type: Annotated[str, "Optional workspace type: personal, group or public."] = "", + group: Annotated[str, "Optional group ID from find_entities."] = "", + public_workspace: Annotated[str, "Optional public workspace ID from find_entities."] = "", + model: Annotated[str, "Optional model deployment name, exactly as recorded."] = "", + token_type: Annotated[str, "Optional usage type: chat, embedding or web_search."] = "", + limit: Annotated[int, "Rows to return for model, user, group or public_workspace totals, 1 to 50."] = 10, + ) -> dict: + def run(stores): + grouping = _choice(group_by, DASHBOARD_TOKEN_GROUPINGS, "group_by") or "day" + period = resolve_requested_period(start_date, end_date, days) + model_name = " ".join(str(model or "").split()) + if len(model_name) > _MODEL_MAX_LENGTH: + raise ValueError(f"model may not exceed {_MODEL_MAX_LENGTH} characters.") + filters = extract_token_filters({ + "user_id": _entity_id(user, "user"), + "workspace_type": _choice(workspace_type, DASHBOARD_WORKSPACE_TYPES, "workspace_type"), + "group_id": _entity_id(group, "group"), + "public_workspace_id": _entity_id(public_workspace, "public_workspace"), + "model": model_name, + "token_type": _choice(token_type, DASHBOARD_TOKEN_TYPES, "token_type"), + }) + report = token_usage_report( + stores, period, filters, group_by=grouping, limit=_limit(limit), log=log_event, + ) + return { + "period": period.describe(), + "group_by": grouping, + "filters": self._describe_filters(stores, filters), + "total_tokens": report["total_tokens"], + "rows": report["rows"], + } + + return self._run("get_token_usage", run) + + @plugin_function_logger("ControlCenterPlugin") + @kernel_function(description=_FUNCTION_DEFINITIONS[3][1]) + def get_top_activity( + self, + entity: Annotated[str, "Which ranking: users, groups or public_workspaces."] = "users", + start_date: Annotated[str, _DATE_PARAMETER] = "", + end_date: Annotated[str, _END_PARAMETER] = "", + days: Annotated[int, _DAYS_PARAMETER] = 30, + limit: Annotated[int, "Rows to return, 1 to 50."] = 10, + ) -> dict: + def run(stores): + kind = _choice(entity, DASHBOARD_ENTITY_KINDS, "entity") or "users" + period = resolve_requested_period(start_date, end_date, days) + return { + "period": period.describe(), + "entity": kind, + "rows": top_activity_report(stores, period, entity=kind, limit=_limit(limit), log=log_event), + "notes": [ + "activity_count counts every recorded action: sign-ins, conversations, uploads, " + "token records and administrative changes.", + ], + } + + return self._run("get_top_activity", run) + + @plugin_function_logger("ControlCenterPlugin") + @kernel_function(description=_FUNCTION_DEFINITIONS[4][1]) + def get_sign_in_pattern( + self, + start_date: Annotated[str, _DATE_PARAMETER] = "", + end_date: Annotated[str, _END_PARAMETER] = "", + days: Annotated[int, _DAYS_PARAMETER] = 30, + ) -> dict: + def run(stores): + period = resolve_requested_period(start_date, end_date, days) + return { + "period": period.describe(), + "rows": sign_in_pattern_report(stores, period, log=log_event), + "notes": ["Each row totals every sign-in in the period for one UTC weekday and hour."], + } + + return self._run("get_sign_in_pattern", run) + + @plugin_function_logger("ControlCenterPlugin") + @kernel_function(description=_FUNCTION_DEFINITIONS[5][1]) + def find_entities( + self, + kind: Annotated[str, "What to look up: users, groups or public_workspaces."] = "users", + search: Annotated[str, "Part of a name or email, at least 2 characters."] = "", + limit: Annotated[int, "Matches to return, 1 to 10."] = 10, + ) -> dict: + def run(stores): + entity_kind = _choice(kind, DASHBOARD_ENTITY_KINDS, "kind") or "users" + matches = search_dashboard_entities(stores, entity_kind, search, limit=min(_limit(limit), 10)) + return {"kind": entity_kind, "rows": matches} + + return self._run("find_entities", run) diff --git a/application/single_app/static/js/admin/admin_governance.js b/application/single_app/static/js/admin/admin_governance.js index 7361a23eb..45fac45dc 100644 --- a/application/single_app/static/js/admin/admin_governance.js +++ b/application/single_app/static/js/admin/admin_governance.js @@ -120,6 +120,7 @@ const GOVERNANCE_ACTION_TYPE_ALIASES = { blob_storage: 'blob_storage', document_search: 'document_search', search: 'document_search', + control_center: 'control_center', }; const GOVERNANCE_ACTION_TYPE_LABELS = { @@ -139,6 +140,7 @@ const GOVERNANCE_ACTION_TYPE_LABELS = { azure_maps: 'Azure Maps', blob_storage: 'Blob Storage', document_search: 'Document Search', + control_center: 'Control Center', }; const GOVERNANCE_PRIMARY_TOGGLE_MAP = { diff --git a/application/single_app/static/js/control-center.js b/application/single_app/static/js/control-center.js index 2c11536c3..3d422a3f8 100644 --- a/application/single_app/static/js/control-center.js +++ b/application/single_app/static/js/control-center.js @@ -365,15 +365,6 @@ class ControlCenter { radio.addEventListener('change', () => this.toggleExportCustomDateRange()); }); - // Chat functionality - document.getElementById('executeChatBtn')?.addEventListener('click', - () => this.chatActivityTrends()); - - // Chat modal - show/hide custom date range based on radio selection - document.querySelectorAll('input[name="chatTimeWindow"]').forEach(radio => { - radio.addEventListener('change', () => this.toggleChatCustomDateRange()); - }); - // Activity Logs event handlers document.getElementById('activityLogsSearchInput')?.addEventListener('input', this.debounce(() => this.handleActivityLogsSearchChange(), 300)); @@ -3851,115 +3842,6 @@ class ControlCenter { return text.replace(/[&<>"']/g, m => map[m]); } - toggleChatCustomDateRange() { - const customRadio = document.getElementById('chatCustom'); - const customDateRange = document.getElementById('chatCustomDateRange'); - - if (customRadio.checked) { - customDateRange.style.display = 'block'; - // Set default dates - const endDate = new Date(); - const startDate = new Date(); - startDate.setDate(startDate.getDate() - 29); - - document.getElementById('chatStartDate').value = startDate.toISOString().split('T')[0]; - document.getElementById('chatEndDate').value = endDate.toISOString().split('T')[0]; - } else { - customDateRange.style.display = 'none'; - } - } - - async chatActivityTrends() { - try { - // Get selected charts - const selectedCharts = []; - if (document.getElementById('chatLogins').checked) selectedCharts.push('logins'); - if (document.getElementById('chatChats').checked) selectedCharts.push('chats'); - if (document.getElementById('chatDocuments').checked) selectedCharts.push('documents'); - - if (selectedCharts.length === 0) { - showToast('Please select at least one chart to include in the chat.', 'warning'); - return; - } - - // Get selected time window - const timeWindowRadio = document.querySelector('input[name="chatTimeWindow"]:checked'); - const timeWindow = timeWindowRadio.value; - - let chatData = { - charts: selectedCharts, - time_window: timeWindow - }; - - const tokenFilters = this.getTokenFilterRequestPayload(); - if (Object.keys(tokenFilters).length > 0) { - chatData.token_filters = tokenFilters; - } - - // Add custom dates if selected - if (timeWindow === 'custom') { - const startDate = document.getElementById('chatStartDate').value; - const endDate = document.getElementById('chatEndDate').value; - - if (!startDate || !endDate) { - showToast('Please select both start and end dates for custom range.', 'warning'); - return; - } - - if (new Date(startDate) > new Date(endDate)) { - showToast('Start date must be before end date.', 'warning'); - return; - } - - chatData.start_date = startDate; - chatData.end_date = endDate; - } - - // Show loading state - const chatBtn = document.getElementById('executeChatBtn'); - const originalText = chatBtn.innerHTML; - chatBtn.innerHTML = 'Creating Chat...'; - chatBtn.disabled = true; - - // Make API call - const response = await fetch('/api/admin/control-center/activity-trends/chat', { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - }, - body: JSON.stringify(chatData) - }); - - const result = await response.json(); - - if (response.ok && result.success) { - // Close modal - const modal = bootstrap.Modal.getInstance(document.getElementById('chatModal')); - modal.hide(); - - // Show success message - this.showAlert('success', 'Chat conversation created successfully! Redirecting...'); - - // Redirect to the new conversation - setTimeout(() => { - window.location.href = result.redirect_url; - }, 1500); - - } else { - throw new Error(result.error || 'Failed to create chat conversation'); - } - - } catch (error) { - console.error('Chat creation error:', error); - this.showAlert('danger', `Failed to create chat: ${error.message}`); - } finally { - // Reset button state - const chatBtn = document.getElementById('executeChatBtn'); - chatBtn.innerHTML = 'Start Chat'; - chatBtn.disabled = false; - } - } - destroyAllCharts() { // Destroy all chart instances if (this.loginsChart) { diff --git a/application/single_app/static/js/plugin_modal_stepper.js b/application/single_app/static/js/plugin_modal_stepper.js index c6b3dd196..f17bed257 100644 --- a/application/single_app/static/js/plugin_modal_stepper.js +++ b/application/single_app/static/js/plugin_modal_stepper.js @@ -3,8 +3,9 @@ import { showToast } from "./chat/chat-toast.js"; import { getTypeIcon, getMcpRetirementStatus, isMcpActionType } from "./workspace/view-utils.js"; -// Action types hidden from the creation UI (backend plugins remain intact) -const HIDDEN_ACTION_TYPES = ['sql_schema', 'ui_test', 'queue_storage', 'embedding_model', 'databricks_table']; +// Action types hidden from the creation UI (backend plugins remain intact). +// control_center is created in the V2 admin action editor, the one surface that configures it. +const HIDDEN_ACTION_TYPES = ['sql_schema', 'ui_test', 'queue_storage', 'embedding_model', 'databricks_table', 'control_center']; const M365_ACTION_TYPES = ['m365_calendar', 'm365_email', 'm365_onedrive', 'm365_sharepoint']; const ACTION_IDENTITY_AUTH_TYPES = ['api_key', 'bearer_token', 'client_secret', 'connection_string', 'managed_identity', 'username_password']; const SQL_ACTION_IDENTITY_AUTH_TYPES = ['connection_string', 'managed_identity', 'username_password']; diff --git a/application/single_app/static/js/workspace/view-utils.js b/application/single_app/static/js/workspace/view-utils.js index 6f64df131..1b22147f3 100644 --- a/application/single_app/static/js/workspace/view-utils.js +++ b/application/single_app/static/js/workspace/view-utils.js @@ -99,6 +99,7 @@ export function getTypeIcon(type) { if (t.includes("rocksdb")) return "bi-hdd-stack"; if (t.includes("chart")) return "bi-bar-chart-line"; if (t.includes("simplechat")) return "bi-chat-square-dots"; + if (t === "control_center") return "bi-speedometer2"; if (t === "mcp" || t.includes("model_context_protocol")) return "bi-diagram-3"; if (t.includes("openapi")) return "bi-globe"; if (t.includes("log_analytics")) return "bi-graph-up"; diff --git a/application/single_app/static/json/schemas/control_center.definition.json b/application/single_app/static/json/schemas/control_center.definition.json new file mode 100644 index 000000000..92fcefed0 --- /dev/null +++ b/application/single_app/static/json/schemas/control_center.definition.json @@ -0,0 +1,7 @@ +{ + "$schema": "./plugin.definition.schema.json", + "allowedAuthTypes": [ + "user" + ], + "defaultAuthType": "user" +} diff --git a/application/single_app/templates/control_center.html b/application/single_app/templates/control_center.html index 95959d0f2..2c1b9d4ff 100644 --- a/application/single_app/templates/control_center.html +++ b/application/single_app/templates/control_center.html @@ -1733,88 +1733,6 @@
Select Time Window:
- - -