diff --git a/server/chat/backend/agent/orchestrator/sub_agent.py b/server/chat/backend/agent/orchestrator/sub_agent.py index 378c5d435..2cdca9458 100644 --- a/server/chat/backend/agent/orchestrator/sub_agent.py +++ b/server/chat/backend/agent/orchestrator/sub_agent.py @@ -9,13 +9,17 @@ from chat.backend.agent.orchestrator.inputs import FindingRef, SubAgentInput from chat.backend.agent.orchestrator.findings_schema import make_stub +from chat.backend.agent.utils.tool_call_history import ( + MAX_HISTORY_ENTRIES, + OUTPUT_EXCERPT_MAX_CHARS, + derive_command, +) from utils.log_sanitizer import hash_for_log +from utils.text.text_utils import truncate logger = logging.getLogger(__name__) _DEFAULT_TIMEOUT_SECONDS = 600 -_MAX_HISTORY_ENTRIES = 30 -_MAX_HISTORY_FIELD_CHARS = 1000 _FINDING_REF_STATUSES = frozenset({"succeeded", "failed", "timeout", "cancelled", "inconclusive"}) # Loop-guard thresholds: number of consecutive empty/error results from the @@ -127,54 +131,7 @@ def wrapped_func(*args, **kwargs): return wrapped -def _truncate(value, limit: int = _MAX_HISTORY_FIELD_CHARS) -> str: - if value is None: - return "" - s = value if isinstance(value, str) else str(value) - return s if len(s) <= limit else s[:limit] + "...[truncated]" - - -# Map cloud_exec providers to their CLI prefix. Mirrors the frontend's -# getProviderCli so the captured command string already includes the prefix -# (e.g. "aws cloudwatch ...") and CommandLogo can pick the right icon from -# command.startsWith without needing a separate provider field. -_PROVIDER_CLI = { - "aws": "aws", - "gcp": "gcloud", "gcloud": "gcloud", - "azure": "az", "az": "az", - "ovh": "ovhcloud", "ovhcloud": "ovhcloud", - "scaleway": "scw", "scw": "scw", -} -_RECOGNIZED_CLI_PREFIXES = ( - "aws ", "gcloud ", "gsutil ", "bq ", "az ", - "ovhcloud ", "scw ", - "kubectl ", "helm ", "docker ", -) - - -def _entry_command(input_value, limit: int = 1024) -> str: - """Pull the display-ready command/query out of a tool's input dict before - the full args blob is truncated. Without this, large inputs (e.g. cloudwatch - get-metric-data with embedded JSON queries) get cut mid-string and downstream - consumers can't recover the command for the citation/Thoughts UI.""" - if not isinstance(input_value, dict): - return "" - cmd = input_value.get("command") - if isinstance(cmd, str) and cmd.strip(): - provider = input_value.get("provider") - if provider: - cli = _PROVIDER_CLI.get(str(provider).lower()) - if cli and not cmd.lstrip().startswith(_RECOGNIZED_CLI_PREFIXES): - cmd = f"{cli} {cmd.lstrip()}" - return _truncate(cmd, limit) - for key in ("query", "path", "promql"): - v = input_value.get(key) - if v: - return _truncate(str(v), limit) - return "" - - -def _serialize_args(value, limit: int = _MAX_HISTORY_FIELD_CHARS) -> str: +def _serialize_args(value, limit: int = OUTPUT_EXCERPT_MAX_CHARS) -> str: """JSON-encode tool args so downstream consumers can json.loads without needing a Python-repr fallback. Falls back to str() for non-serializable values (rare; keeps the field non-empty).""" @@ -216,10 +173,10 @@ def _extract_tool_call_history(tool_capture) -> list[dict]: seen_ids.add(call_id) input_dict = entry.get("input") items.append({ - "tool_name": _truncate(entry.get("tool_name") or "unknown", 128), + "tool_name": truncate(entry.get("tool_name") or "unknown", 128), "args": _serialize_args(input_dict), - "command": _entry_command(input_dict), - "output_excerpt": _truncate(entry.get("output_excerpt") or "", _MAX_HISTORY_FIELD_CHARS), + "command": derive_command(input_dict), + "output_excerpt": truncate(entry.get("output_excerpt") or "", OUTPUT_EXCERPT_MAX_CHARS), "is_error": bool(entry.get("is_error", False)), "status": "error" if entry.get("is_error", False) else "completed", "started_at": entry.get("started_at"), @@ -238,9 +195,9 @@ def _extract_tool_call_history(tool_capture) -> list[dict]: started_iso = None input_dict = info.get("input") items.append({ - "tool_name": _truncate(info.get("tool_name") or "unknown", 128), + "tool_name": truncate(info.get("tool_name") or "unknown", 128), "args": _serialize_args(input_dict), - "command": _entry_command(input_dict), + "command": derive_command(input_dict), "output_excerpt": "", "is_error": False, "status": "running", @@ -252,7 +209,7 @@ def _extract_tool_call_history(tool_capture) -> list[dict]: items.sort(key=lambda d: d.get("started_at") or "") except Exception: logger.debug("sub_agent: tool_history sort skipped due to malformed entry", exc_info=True) - return items[:_MAX_HISTORY_ENTRIES] + return items[:MAX_HISTORY_ENTRIES] except Exception: logger.exception("sub_agent: tool_call_history extraction failed") return [] diff --git a/server/chat/backend/agent/tools/cloud_tools.py b/server/chat/backend/agent/tools/cloud_tools.py index 60c0ff482..e9308a0e8 100644 --- a/server/chat/backend/agent/tools/cloud_tools.py +++ b/server/chat/backend/agent/tools/cloud_tools.py @@ -1029,8 +1029,10 @@ def get_cloud_tools(): is_postmortem_action = getattr(state_context, 'is_postmortem_action', False) if state_context else False is_pr_review = getattr(state_context, 'is_pr_review', False) if state_context else False is_rca_context = _is_background_rca(state_context, is_background) + _action_id = getattr(state_context, 'trigger_action_id', None) if state_context else None + _incident_id = getattr(state_context, 'incident_id', None) if state_context else None capture_tag = "capture" if tool_capture else "nocapture" - cache_key = f"{user_id}:{capture_tag}:{mode_suffix}:background={is_background}:rca={rca_flag}:postmortem={is_postmortem_action}:is_rca_ctx={is_rca_context}:pr_review={is_pr_review}" + cache_key = f"{user_id}:{capture_tag}:{mode_suffix}:background={is_background}:rca={rca_flag}:postmortem={is_postmortem_action}:is_rca_ctx={is_rca_context}:pr_review={is_pr_review}:action_id={_action_id}:incident={_incident_id}" current_time = time.time() if ( @@ -1332,12 +1334,12 @@ def cloud_exec_wrapper(provider: str, command: str, output_file: Optional[str] = tool_functions.append((run_iac_tool, "iac_tool")) tool_functions.append((cloud_exec_wrapper, "cloud_exec")) - # Only include trigger_rca when the user explicitly requested it via the UI button - if state_context and getattr(state_context, 'trigger_rca_requested', False): + # trigger_rca: available when user clicked the RCA button (UI) OR when + # running as a background agent (Slack, Celery) where there's no UI to gate it. + if is_background or (state_context and getattr(state_context, 'trigger_rca_requested', False)): tool_functions.append((trigger_rca, "trigger_rca")) - # Only include trigger_action when the user explicitly used /action command - _action_id = getattr(state_context, 'trigger_action_id', None) if state_context else None + # trigger_action: only when an action id is pinned (UI /action or background action run). if _action_id: tool_functions.append((trigger_action, "trigger_action")) @@ -1825,6 +1827,197 @@ def _pinned_trigger(action_id: str = "", _pid=pinned_id, _fn=final_func, **kw): except Exception as e: logging.warning(f"Failed to add get_infrastructure_context tool: {e}") + # Introspection tools — incident/infra/actions/metrics self-audit (always available) + if user_id: + try: + from chat.backend.agent.tools.introspection_tools import ( + list_incidents, ListIncidentsArgs as IntrospectionListIncidentsArgs, + get_incident, GetIncidentArgs as IntrospectionGetIncidentArgs, + incident_list_alerts, IncidentListAlertsArgs, + list_services, ListServicesArgs, + service_impact, ServiceImpactArgs, + list_actions, ListActionsArgs, + list_action_runs, ListActionRunsArgs, + get_metrics_summary, GetMetricsSummaryArgs, + get_mttr, GetMttrArgs, + get_incident_frequency, GetIncidentFrequencyArgs, + get_change_failure_rate, GetChangeFailureRateArgs, + get_llm_usage_summary, GetLlmUsageSummaryArgs, + incident_findings, IncidentFindingsArgs, + incident_finding_detail, IncidentFindingDetailArgs, + get_action, GetActionArgs, + graph_get_service, GraphGetServiceArgs, + postmortem_list, PostmortemListArgs, + kb_get_memory, KbGetMemoryArgs, + grafana_list_alerts, GrafanaListAlertsArgs, + ) + + _INTROSPECTION_TOOLS = [ + ( + list_incidents, + "list_incidents", + "List Aurora incidents. Optionally filter by status " + "(investigating/analyzed/merged/resolved) with pagination. " + "Returns id, title, status, severity, service, summary, and timestamps.", + IntrospectionListIncidentsArgs, + ), + ( + get_incident, + "get_incident", + "Get full incident details: summary, suggestions, correlated alerts, " + "affected services. Use this to deep-dive into a specific incident.", + IntrospectionGetIncidentArgs, + ), + ( + incident_list_alerts, + "incident_list_alerts", + "List the alerts correlated to an incident: source, title, service, " + "severity, and correlation score. Use to answer 'what alerts fired " + "for incident X'.", + IncidentListAlertsArgs, + ), + ( + list_services, + "list_services", + "List services in the infrastructure dependency graph. " + "Optional filters: resource_type, provider. Use to enumerate " + "what exists before drilling into a specific service.", + ListServicesArgs, + ), + ( + service_impact, + "service_impact", + "Get a service's blast radius — the downstream services that depend " + "on it. Use to answer 'what breaks if goes down'.", + ServiceImpactArgs, + ), + ( + list_actions, + "list_actions", + "List this org's Aurora actions (automations): name, trigger type, " + "mode, enabled, run count, and last-run status.", + ListActionsArgs, + ), + ( + list_action_runs, + "list_action_runs", + "List an Aurora action's run history: status, timing, linked incident, " + "and any error. Use to check whether an automation ran and how it went.", + ListActionRunsArgs, + ), + ( + get_metrics_summary, + "get_metrics_summary", + "SRE dashboard overview: total/active/resolved incident counts, " + "average MTTR, MTTS (time to RCA), and top affected services. " + "Use to answer 'how are we doing operationally'. Period: 7d/30d/90d/180d/365d.", + GetMetricsSummaryArgs, + ), + ( + get_mttr, + "get_mttr", + "Mean Time to Resolve with p50/p95 percentiles, broken down by severity " + "and trended over time. Filterable by period, severity, and service.", + GetMttrArgs, + ), + ( + get_incident_frequency, + "get_incident_frequency", + "Incident count over time, grouped by severity, service, or source type. " + "Use to answer 'how many incidents this week' or 'which service has the most'.", + GetIncidentFrequencyArgs, + ), + ( + get_change_failure_rate, + "get_change_failure_rate", + "DORA Change Failure Rate — percentage of deployments that caused " + "incidents. Correlates Jenkins/CloudBees deploy events with incidents.", + GetChangeFailureRateArgs, + ), + ( + get_llm_usage_summary, + "get_llm_usage_summary", + "Aggregate LLM token usage and estimated cost for the org: total cost, " + "tokens (input/output), request count, error rate, avg latency. " + "Period: 7d/30d/90d/180d/365d.", + GetLlmUsageSummaryArgs, + ), + ( + incident_findings, + "incident_findings", + "List RCA sub-agent findings for an incident — shows each agent's role, " + "purpose, status, strength rating, tools used, and citations. " + "Use to answer 'what did the RCA investigate' or 'which agents ran'.", + IncidentFindingsArgs, + ), + ( + incident_finding_detail, + "incident_finding_detail", + "Get a single RCA sub-agent's full finding body (markdown) and its " + "step-by-step tool call history. Use after incident_findings to " + "deep-dive into what a specific agent discovered.", + IncidentFindingDetailArgs, + ), + ( + get_action, + "get_action", + "Get full action config (name, description, instructions, trigger, mode) " + "plus its 20 most recent runs with status and duration. " + "Use to inspect a specific automation.", + GetActionArgs, + ), + ( + graph_get_service, + "graph_get_service", + "Get a single service with its direct upstream (dependencies) and " + "downstream (dependants) from the infrastructure graph. " + "Richer than service_impact — includes all metadata and both directions.", + GraphGetServiceArgs, + ), + ( + postmortem_list, + "postmortem_list", + "List all postmortems for the organization with pagination. " + "Returns incident title, generation date, and export URLs " + "(Confluence/Jira/Notion). Use to discover existing postmortems.", + PostmortemListArgs, + ), + ( + kb_get_memory, + "kb_get_memory", + "Read the org's persistent knowledge base memory — a shared context " + "document that teams maintain with org-specific conventions, runbook " + "references, and operational notes.", + KbGetMemoryArgs, + ), + ( + grafana_list_alerts, + "grafana_list_alerts", + "List Grafana alerts ingested via webhook. Optionally filter by state " + "(alerting, ok, pending). Returns title, state, rule info, and dashboard link.", + GrafanaListAlertsArgs, + ), + ] + + for _fn, _name, _desc, _schema in _INTROSPECTION_TOOLS: + _ctx_wrapped = with_user_context(_fn) + _notif_wrapped = with_completion_notification(_ctx_wrapped) + if tool_capture: + _final_fn = wrap_func_with_capture(_notif_wrapped, _name) + else: + _final_fn = _notif_wrapped + + tools.append(StructuredTool.from_function( + func=_final_fn, + name=_name, + description=_desc, + args_schema=_schema, + )) + + logging.info(f"Added introspection tools for user {user_id}") + except Exception as e: + logging.warning(f"Failed to add introspection tools: {e}") + # Add discovery finding tool for prediscovery mode if mode_suffix == "prediscovery": try: diff --git a/server/chat/backend/agent/tools/introspection_tools.py b/server/chat/backend/agent/tools/introspection_tools.py new file mode 100644 index 000000000..d0c8a4adf --- /dev/null +++ b/server/chat/backend/agent/tools/introspection_tools.py @@ -0,0 +1,866 @@ +""" +Introspection Tools — self-audit capabilities for the internal agent. + +These give the internal agent (web chat, Slack, background RCA, Actions) the +same incident / infra / actions / metrics visibility that MCP clients have. +Everything reads Postgres or the graph DB directly — no Flask HTTP round-trip. + +Each tool is a thin query wrapped by ``@introspection_tool``, which removes the +boilerplate every tool would otherwise repeat: requiring a user context, +JSON-serializing the result, and turning failures into a clean ``{"error": …}`` +payload. Shared DB access goes through the ``_cursor`` helper, which scopes RLS +to the caller's org and hands back the resolved ``org_id`` in one step. +""" + +import functools +import json +import logging +import re +from contextlib import contextmanager +from typing import Any, Callable, Optional + +from pydantic import BaseModel, Field + +from chat.backend.agent.utils.tool_call_history import ( + MAX_HISTORY_ENTRIES, + OUTPUT_EXCERPT_MAX_CHARS, + TERMINAL_STATUSES, + history_from_step_rows, +) +from utils.metrics_periods import period_to_interval +from utils.query_helpers import clamp, duration_ms, fetch_dicts, iso_utc +from utils.validation import is_valid_uuid +from utils.db.connection_pool import db_pool +from utils.auth.stateless_auth import set_rls_context + +logger = logging.getLogger(__name__) + +_LOG_PREFIX = "[IntrospectionTools]" + + +# --------------------------------------------------------------------------- +# Shared infrastructure +# --------------------------------------------------------------------------- + +class IntrospectionError(Exception): + """Raised inside a tool to return a clean, user-facing error message.""" + + +def introspection_tool(fn: Callable[..., Any]) -> Callable[..., str]: + """Wrap a tool so it only has to contain its query logic. + + Responsibilities lifted out of every tool: + * require an agent ``user_id`` (injected by ``with_user_context``), + * serialize the returned dict/list to JSON, + * surface ``IntrospectionError`` as ``{"error": message}``, + * log and mask any unexpected exception. + + Wrapped tools take ``user_id`` as a keyword arg and return a plain dict. + """ + @functools.wraps(fn) + def wrapper(*args, user_id: Optional[str] = None, **kwargs) -> str: + if not user_id: + return json.dumps({"error": "No user context available."}) + try: + return json.dumps(fn(*args, user_id=user_id, **kwargs), default=str) + except IntrospectionError as exc: + return json.dumps({"error": str(exc)}) + except Exception: + logger.exception("%s %s failed", _LOG_PREFIX, fn.__name__) + return json.dumps({"error": f"Failed to run {fn.__name__}."}) + + return wrapper + + +@contextmanager +def _cursor(user_id: str): + """Yield ``(cursor, org_id)`` on a pooled connection with RLS scoped to the org.""" + with db_pool.get_connection() as conn: + with conn.cursor() as cur: + org_id = set_rls_context(cur, conn, user_id, log_prefix=_LOG_PREFIX) + if not org_id: + raise IntrospectionError("No organization context.") + yield cur, org_id + + +def _require_uuid(value: Optional[str], field: str = "id") -> str: + """Validate that ``value`` is a UUID, or fail with a tool-facing error. + + Surfaces as an ``IntrospectionError`` (which the decorator turns into a clean + payload) rather than the bare bool the shared predicate returns. + """ + if not is_valid_uuid(value): + raise IntrospectionError(f"A valid {field} UUID is required.") + return value + + +# --------------------------------------------------------------------------- +# Incidents +# --------------------------------------------------------------------------- + +class ListIncidentsArgs(BaseModel): + status: Optional[str] = Field( + default=None, + description="Filter by incident status: investigating, analyzed, merged, or resolved.", + ) + limit: int = Field(default=20, description="Max incidents to return (1–100, default 20).") + offset: int = Field(default=0, description="Paging offset (>= 0, default 0).") + + +@introspection_tool +def list_incidents(status=None, limit=20, offset=0, *, user_id, **_) -> dict: + """List Aurora incidents with optional status filter and pagination.""" + limit, offset = clamp(limit, 1, 100), max(0, int(offset)) + + with _cursor(user_id) as (cur, org_id): + where = "i.org_id = %s" + params = [org_id] + if status: + where += " AND i.status = %s" + params.append(status) + else: + # Hide merged by default unless caller explicitly filters for them + where += " AND i.status != 'merged'" + + cur.execute(f"SELECT COUNT(*) FROM incidents i WHERE {where}", tuple(params)) + total = cur.fetchone()[0] + cur.execute( + f"""SELECT i.id, i.status, i.severity, i.alert_title, i.alert_service, + i.alert_environment, i.aurora_status, i.aurora_summary, + i.started_at, i.analyzed_at, i.resolved_at, i.source_type + FROM incidents i WHERE {where} + ORDER BY i.started_at DESC LIMIT %s OFFSET %s""", + tuple(params + [limit, offset]), + ) + rows = cur.fetchall() + + incidents = [ + { + "id": str(r[0]), "status": r[1], "severity": r[2], "title": r[3], + "service": r[4], "environment": r[5], "auroraStatus": r[6], "summary": r[7], + "startedAt": iso_utc(r[8]), "analyzedAt": iso_utc(r[9]), "resolvedAt": iso_utc(r[10]), + "sourceType": r[11], + } + for r in rows + ] + return {"incidents": incidents, "total": total} + + +class GetIncidentArgs(BaseModel): + incident_id: str = Field(description="The UUID of the incident to retrieve.") + + +@introspection_tool +def get_incident(incident_id, *, user_id, **_) -> dict: + """Get full incident details: summary, suggestions, and correlated alerts.""" + _require_uuid(incident_id, "incident_id") + + with _cursor(user_id) as (cur, org_id): + cur.execute( + """SELECT id, status, severity, alert_title, alert_service, alert_environment, + aurora_status, aurora_summary, started_at, analyzed_at, resolved_at, + source_type, affected_services, correlated_alert_count + FROM incidents WHERE id = %s AND org_id = %s""", + (incident_id, org_id), + ) + row = cur.fetchone() + if not row: + raise IntrospectionError("Incident not found.") + + incident = { + "id": str(row[0]), "status": row[1], "severity": row[2], "title": row[3], + "service": row[4], "environment": row[5], "auroraStatus": row[6], "summary": row[7], + "startedAt": iso_utc(row[8]), "analyzedAt": iso_utc(row[9]), "resolvedAt": iso_utc(row[10]), + "sourceType": row[11], "affectedServices": row[12] or [], + "correlatedAlertCount": row[13] or 0, + } + + cur.execute( + """SELECT id, title, description, type, risk, command, created_at + FROM incident_suggestions WHERE incident_id = %s ORDER BY created_at ASC""", + (incident_id,), + ) + incident["suggestions"] = [ + { + "id": str(s[0]), "title": s[1], "description": s[2], + "type": s[3] or "diagnostic", "risk": s[4] or "safe", + "command": s[5], "createdAt": iso_utc(s[6]), + } + for s in cur.fetchall() + ] + + cur.execute( + """SELECT id, source_type, alert_title, alert_service, alert_severity, + correlation_score, received_at + FROM incident_alerts WHERE incident_id = %s ORDER BY received_at ASC""", + (incident_id,), + ) + incident["correlatedAlerts"] = [ + { + "id": str(a[0]), "sourceType": a[1], "title": a[2], "service": a[3], + "severity": a[4], "correlationScore": a[5], "receivedAt": iso_utc(a[6]), + } + for a in cur.fetchall() + ] + + return {"incident": incident} + + +class IncidentListAlertsArgs(BaseModel): + incident_id: str = Field( + description="The UUID of the incident whose correlated alerts to list.", + ) + + +@introspection_tool +def incident_list_alerts(incident_id, *, user_id, **_) -> dict: + """List the alerts correlated to an incident with correlation details.""" + _require_uuid(incident_id, "incident_id") + + with _cursor(user_id) as (cur, org_id): + cur.execute( + "SELECT 1 FROM incidents WHERE id = %s AND org_id = %s", (incident_id, org_id) + ) + if not cur.fetchone(): + raise IntrospectionError("Incident not found.") + + cur.execute( + """SELECT id, source_type, alert_title, alert_service, alert_severity, + correlation_strategy, correlation_score, correlation_details, received_at + FROM incident_alerts WHERE incident_id = %s ORDER BY received_at ASC""", + (incident_id,), + ) + alerts = [ + { + "id": str(r[0]), "sourceType": r[1], "title": r[2], "service": r[3], + "severity": r[4], "correlationStrategy": r[5], "correlationScore": r[6], + "correlationDetails": r[7] if isinstance(r[7], dict) else {}, + "receivedAt": iso_utc(r[8]), + } + for r in cur.fetchall() + ] + + return {"alerts": alerts, "total": len(alerts)} + + +# --------------------------------------------------------------------------- +# Service dependency graph +# --------------------------------------------------------------------------- + +class ListServicesArgs(BaseModel): + resource_type: Optional[str] = Field( + default=None, description="Filter by resource type (e.g. 'compute', 'database')." + ) + provider: Optional[str] = Field( + default=None, description="Filter by cloud provider (e.g. 'aws', 'gcp')." + ) + + +@introspection_tool +def list_services(resource_type=None, provider=None, *, user_id, **_) -> dict: + """List services in the infrastructure dependency graph.""" + from services.graph.memgraph_client import get_memgraph_client + + services = get_memgraph_client().list_services( + user_id, resource_type=resource_type, provider=provider + ) + return {"services": services, "total": len(services)} + + +class ServiceImpactArgs(BaseModel): + name: str = Field(description="The service name exactly as it appears in the dependency graph.") + + +@introspection_tool +def service_impact(name, *, user_id, **_) -> dict: + """Get a service's blast radius — the downstream services that depend on it.""" + if not name or not name.strip(): + raise IntrospectionError("Service name is required.") + from services.graph.memgraph_client import get_memgraph_client + + return get_memgraph_client().get_impact_radius(user_id, name.strip()) + + +class GraphGetServiceArgs(BaseModel): + name: str = Field(description="The service name exactly as it appears in the dependency graph.") + + +@introspection_tool +def graph_get_service(name, *, user_id, **_) -> dict: + """Get a service with its direct upstream and downstream dependencies.""" + if not name or not name.strip(): + raise IntrospectionError("Service name is required.") + from services.graph.memgraph_client import get_memgraph_client + + service = get_memgraph_client().get_service(user_id, name.strip()) + if not service: + raise IntrospectionError(f"Service '{name}' not found in graph.") + return service + + +# --------------------------------------------------------------------------- +# Actions (automations) +# --------------------------------------------------------------------------- + +class ListActionsArgs(BaseModel): + pass + + +@introspection_tool +def list_actions(*, user_id, **_) -> dict: + """List the org's Aurora actions (automations) with run counts and last status.""" + with _cursor(user_id) as (cur, org_id): + cur.execute( + """SELECT a.id, a.name, a.description, a.trigger_type, a.mode, a.enabled, + a.created_at, a.updated_at, COUNT(r.id) AS run_count, + MAX(r.started_at) AS last_run_at, + (SELECT r2.status FROM action_runs r2 WHERE r2.action_id = a.id + ORDER BY r2.started_at DESC LIMIT 1) AS last_run_status + FROM actions a + LEFT JOIN action_runs r ON r.action_id = a.id + WHERE a.org_id = %s + GROUP BY a.id + ORDER BY a.is_system DESC, a.created_at DESC""", + (org_id,), + ) + rows = fetch_dicts(cur) + + actions = [ + { + "id": str(a["id"]), "name": a["name"], "description": a["description"], + "triggerType": a["trigger_type"], "mode": a["mode"], "enabled": a["enabled"], + "runCount": a["run_count"] or 0, "lastRunAt": iso_utc(a["last_run_at"]), + "lastRunStatus": a["last_run_status"], "createdAt": iso_utc(a["created_at"]), + "updatedAt": iso_utc(a["updated_at"]), + } + for a in rows + ] + return {"actions": actions, "total": len(actions)} + + +class ListActionRunsArgs(BaseModel): + action_id: str = Field(description="The UUID of the action whose run history to list.") + limit: int = Field(default=50, description="Max runs to return (1–200, default 50).") + offset: int = Field(default=0, description="Paging offset (>= 0, default 0).") + + +@introspection_tool +def list_action_runs(action_id, limit=50, offset=0, *, user_id, **_) -> dict: + """List an action's run history: status, timing, linked incident, and errors.""" + _require_uuid(action_id, "action_id") + limit, offset = clamp(limit, 1, 200), max(0, int(offset)) + + with _cursor(user_id) as (cur, org_id): + # Verify the action belongs to this org before exposing run history + cur.execute("SELECT 1 FROM actions WHERE id = %s AND org_id = %s", (action_id, org_id)) + if not cur.fetchone(): + raise IntrospectionError("Action not found.") + + cur.execute( + """SELECT id, status, incident_id, chat_session_id, started_at, completed_at, error + FROM action_runs WHERE action_id = %s + ORDER BY started_at DESC LIMIT %s OFFSET %s""", + (action_id, limit, offset), + ) + rows = fetch_dicts(cur) + + runs = [ + { + "id": str(r["id"]), "status": r["status"], + "incidentId": str(r["incident_id"]) if r["incident_id"] else None, + "chatSessionId": str(r["chat_session_id"]) if r["chat_session_id"] else None, + "startedAt": iso_utc(r["started_at"]), "completedAt": iso_utc(r["completed_at"]), + "durationMs": duration_ms(r["started_at"], r["completed_at"]), "error": r["error"], + } + for r in rows + ] + return {"runs": runs, "total": len(runs)} + + +class GetActionArgs(BaseModel): + action_id: str = Field(description="The UUID of the action to retrieve.") + + +@introspection_tool +def get_action(action_id, *, user_id, **_) -> dict: + """Get an action's full config plus its 20 most recent runs.""" + _require_uuid(action_id, "action_id") + + with _cursor(user_id) as (cur, org_id): + cur.execute( + """SELECT id, name, description, instructions, trigger_type, trigger_config, + mode, enabled, is_system, system_key, created_at, updated_at + FROM actions WHERE id = %s AND org_id = %s""", + (action_id, org_id), + ) + cols = [d[0] for d in cur.description] + row = cur.fetchone() + if not row: + raise IntrospectionError("Action not found.") + action = dict(zip(cols, row)) + + cur.execute( + """SELECT id, status, incident_id, chat_session_id, trigger_context, + started_at, completed_at, error + FROM action_runs WHERE action_id = %s ORDER BY started_at DESC LIMIT 20""", + (action_id,), + ) + runs_raw = fetch_dicts(cur) + + action["id"] = str(action["id"]) + action["created_at"] = iso_utc(action.get("created_at")) + action["updated_at"] = iso_utc(action.get("updated_at")) + + runs = [ + { + "id": str(r["id"]), "status": r["status"], + "incident_id": str(r["incident_id"]) if r.get("incident_id") else None, + "started_at": iso_utc(r.get("started_at")), "completed_at": iso_utc(r.get("completed_at")), + "duration_ms": duration_ms(r.get("started_at"), r.get("completed_at")), + "error": r.get("error"), + } + for r in runs_raw + ] + return {"action": action, "recent_runs": runs} + + +# --------------------------------------------------------------------------- +# DORA / SRE metrics +# --------------------------------------------------------------------------- + +# Resolve/analysis latency in seconds, reused across the metrics queries. +_MTTR_EPOCH = "EXTRACT(EPOCH FROM (COALESCE(resolved_at, analyzed_at) - started_at))" + + +class GetMetricsSummaryArgs(BaseModel): + period: str = Field( + default="30d", description="Time period: 7d, 30d, 90d, 180d, or 365d (default 30d)." + ) + + +@introspection_tool +def get_metrics_summary(period="30d", *, user_id, **_) -> dict: + """Dashboard overview: incident counts, average MTTR/MTTS, and top services.""" + interval = period_to_interval(period) + + with _cursor(user_id) as (cur, _org): + cur.execute( + """SELECT + COUNT(*) FILTER (WHERE started_at >= NOW() - %s::interval) AS total, + COUNT(*) FILTER ( + WHERE status IN ('investigating', 'analyzed') + AND aurora_status NOT IN ('complete', 'resolved') + ) AS active, + COUNT(*) FILTER ( + WHERE status = 'resolved' AND resolved_at >= NOW() - %s::interval + ) AS resolved + FROM incidents""", + (interval, interval), + ) + total, active, resolved = cur.fetchone() + + cur.execute( + f"""SELECT AVG({_MTTR_EPOCH}) FROM incidents + WHERE resolved_at IS NOT NULL AND status = 'resolved' + AND resolved_at >= NOW() - %s::interval""", + (interval,), + ) + avg_mttr = cur.fetchone()[0] + + cur.execute( + """SELECT AVG(EXTRACT(EPOCH FROM (analyzed_at - started_at))) FROM incidents + WHERE analyzed_at IS NOT NULL AND analyzed_at >= NOW() - %s::interval""", + (interval,), + ) + avg_mtts = cur.fetchone()[0] + + cur.execute( + """SELECT alert_service, COUNT(*) AS cnt FROM incidents + WHERE started_at >= NOW() - %s::interval + AND alert_service IS NOT NULL AND status != 'merged' + GROUP BY alert_service ORDER BY cnt DESC LIMIT 10""", + (interval,), + ) + top_services = [{"service": r[0], "count": r[1]} for r in cur.fetchall()] + + return { + "period": period, + "totalIncidents": total or 0, + "activeIncidents": active or 0, + "resolvedIncidents": resolved or 0, + "avgMttrSeconds": round(avg_mttr, 1) if avg_mttr else None, + "avgMttsSeconds": round(avg_mtts, 1) if avg_mtts else None, + "topServices": top_services, + } + + +class GetMttrArgs(BaseModel): + period: str = Field(default="30d", description="Time period: 7d, 30d, 90d, 180d, or 365d.") + severity: Optional[str] = Field( + default=None, description="Filter by severity (e.g. critical, high, medium, low)." + ) + service: Optional[str] = Field(default=None, description="Filter by service name.") + + +@introspection_tool +def get_mttr(period="30d", severity=None, service=None, *, user_id, **_) -> dict: + """Mean Time to Resolve with p50/p95, broken down by severity and trended daily.""" + where = ["resolved_at IS NOT NULL", "status = 'resolved'", + "resolved_at >= NOW() - %s::interval"] + params = [period_to_interval(period)] + if severity: + where.append("severity = %s") + params.append(severity) + if service: + where.append("alert_service = %s") + params.append(service) + where_sql = " AND ".join(where) + + with _cursor(user_id) as (cur, _org): + cur.execute( + f"""SELECT COALESCE(severity, 'unknown') AS sev, COUNT(*) AS count, + AVG({_MTTR_EPOCH}) AS avg_mttr, + PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY {_MTTR_EPOCH}) AS p50, + PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY {_MTTR_EPOCH}) AS p95 + FROM incidents WHERE {where_sql} + GROUP BY sev ORDER BY count DESC""", + tuple(params), + ) + by_severity = [ + { + "severity": r[0], "count": r[1], + "avgSeconds": round(r[2], 1) if r[2] else None, + "p50Seconds": round(r[3], 1) if r[3] else None, + "p95Seconds": round(r[4], 1) if r[4] else None, + } + for r in cur.fetchall() + ] + + cur.execute( + f"""SELECT date_trunc('day', COALESCE(resolved_at, analyzed_at))::date AS day, + AVG({_MTTR_EPOCH}) AS avg_mttr, COUNT(*) AS count + FROM incidents WHERE {where_sql} + GROUP BY day ORDER BY day ASC""", + tuple(params), + ) + trend = [ + {"date": str(r[0]), "avgSeconds": round(r[1], 1) if r[1] else None, "count": r[2]} + for r in cur.fetchall() + ] + + return {"bySeverity": by_severity, "trend": trend, "period": period} + + +class GetIncidentFrequencyArgs(BaseModel): + period: str = Field(default="30d", description="Time period: 7d, 30d, 90d, 180d, or 365d.") + group_by: str = Field( + default="severity", description="Group results by: severity, service, or source_type." + ) + + +@introspection_tool +def get_incident_frequency(period="30d", group_by="severity", *, user_id, **_) -> dict: + """Incident count over time, grouped by severity, service, or source type.""" + if group_by not in ("severity", "service", "source_type"): + group_by = "severity" + # Whitelisted above, so this column name is safe to interpolate. + group_col = "alert_service" if group_by == "service" else group_by + + with _cursor(user_id) as (cur, _org): + cur.execute( + f"""SELECT date_trunc('day', started_at)::date AS day, + COALESCE({group_col}, 'unknown') AS group_value, COUNT(*) AS count + FROM incidents + WHERE started_at >= NOW() - %s::interval AND status != 'merged' + GROUP BY day, group_value ORDER BY day ASC, count DESC""", + (period_to_interval(period),), + ) + data = [{"date": str(r[0]), "group": r[1], "count": r[2]} for r in cur.fetchall()] + + return {"data": data, "groupBy": group_by, "period": period} + + +class GetChangeFailureRateArgs(BaseModel): + period: str = Field(default="30d", description="Time period: 7d, 30d, 90d, 180d, or 365d.") + + +@introspection_tool +def get_change_failure_rate(period="30d", *, user_id, **_) -> dict: + """Percentage of deployments that caused an incident within a 4-hour window.""" + with _cursor(user_id) as (cur, _org): + cur.execute( + """WITH deploys AS ( + SELECT id, service, received_at FROM jenkins_deployment_events + WHERE received_at >= NOW() - %s::interval + ), + deploy_failures AS ( + SELECT DISTINCT d.id FROM deploys d + JOIN incidents i ON ( + i.alert_service = d.service + AND i.started_at BETWEEN d.received_at + AND d.received_at + make_interval(hours => 4) + AND i.status != 'merged' + ) + ) + SELECT (SELECT COUNT(*) FROM deploys), + (SELECT COUNT(*) FROM deploy_failures)""", + (period_to_interval(period),), + ) + total, failures = cur.fetchone() + + total, failures = total or 0, failures or 0 + return { + "period": period, + "totalDeployments": total, + "failureLinked": failures, + "changeFailureRate": round(failures / total * 100, 2) if total else 0, + } + + +# --------------------------------------------------------------------------- +# LLM usage / cost +# --------------------------------------------------------------------------- + +class GetLlmUsageSummaryArgs(BaseModel): + period: str = Field(default="30d", description="Time period: 7d, 30d, 90d, 180d, or 365d.") + + +@introspection_tool +def get_llm_usage_summary(period="30d", *, user_id, **_) -> dict: + """Aggregate LLM token usage and estimated cost for the org.""" + with _cursor(user_id) as (cur, org_id): + cur.execute( + """SELECT COALESCE(SUM(estimated_cost), 0), COALESCE(SUM(total_tokens), 0), + COALESCE(SUM(input_tokens), 0), COALESCE(SUM(output_tokens), 0), + COUNT(*), COUNT(*) FILTER (WHERE error_message IS NOT NULL), + ROUND(AVG(response_time_ms) FILTER (WHERE response_time_ms IS NOT NULL)), + COUNT(DISTINCT model_name) + FROM llm_usage_tracking + WHERE org_id = %s AND timestamp >= NOW() - %s::interval""", + (org_id, period_to_interval(period)), + ) + cost, tokens, inp, outp, requests, errors, avg_ms, models = cur.fetchone() + + requests, errors = requests or 0, errors or 0 + return { + "period": period, + "totalCost": float(cost) if cost else 0.0, + "totalTokens": tokens or 0, + "inputTokens": inp or 0, + "outputTokens": outp or 0, + "totalRequests": requests, + "errorCount": errors, + "errorRate": round(errors / requests * 100, 1) if requests else 0, + "avgResponseMs": int(avg_ms) if avg_ms else None, + "modelsUsed": models or 0, + } + + +# --------------------------------------------------------------------------- +# RCA findings +# --------------------------------------------------------------------------- + +_AGENT_ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$") + + +class IncidentFindingsArgs(BaseModel): + incident_id: str = Field(description="The UUID of the incident whose RCA findings to list.") + + +@introspection_tool +def incident_findings(incident_id, *, user_id, **_) -> dict: + """List RCA sub-agent findings for an incident: role, status, tools, citations.""" + _require_uuid(incident_id, "incident_id") + + with _cursor(user_id) as (cur, _org): + cur.execute( + """SELECT agent_id, role_name, purpose, status, self_assessed_strength, + current_action, started_at, completed_at, tools_used, citations, + follow_ups_suggested, wave + FROM rca_findings WHERE incident_id = %s ORDER BY started_at ASC""", + (incident_id,), + ) + rows = fetch_dicts(cur) + + findings = [ + { + "agent_id": f["agent_id"], "role_name": f["role_name"], "purpose": f["purpose"], + "status": f["status"], "wave": f.get("wave"), + "self_assessed_strength": f.get("self_assessed_strength"), + "current_action": f.get("current_action"), + "started_at": iso_utc(f.get("started_at")), "completed_at": iso_utc(f.get("completed_at")), + "tools_used": f.get("tools_used") or [], "citations": f.get("citations") or [], + "follow_ups_suggested": f.get("follow_ups_suggested") or [], + } + for f in rows + ] + return {"findings": findings, "count": len(findings)} + + +class IncidentFindingDetailArgs(BaseModel): + incident_id: str = Field(description="The UUID of the incident.") + agent_id: str = Field( + description="The sub-agent ID (alphanumeric/dash/underscore, max 64 chars)." + ) + + +@introspection_tool +def incident_finding_detail(incident_id, agent_id, *, user_id, **_) -> dict: + """Get one sub-agent's full finding body plus its step-by-step tool history.""" + _require_uuid(incident_id, "incident_id") + if not _AGENT_ID_RE.match(agent_id or ""): + raise IntrospectionError("Invalid agent_id format.") + + with _cursor(user_id) as (cur, _org): + cur.execute( + "SELECT storage_uri, status, tool_call_history, user_id " + "FROM rca_findings WHERE incident_id = %s AND agent_id = %s", + (incident_id, agent_id), + ) + row = cur.fetchone() + if not row: + raise IntrospectionError("Finding not found.") + storage_uri, status, archived_history, originator_id = row + + # Prefer live steps; sub-agents in flight have no archived blob yet. + # Escape LIKE metacharacters — agent IDs may contain underscores + escaped_agent_id = agent_id.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + cur.execute( + """SELECT tool_name, tool_input, LEFT(tool_output, %s) AS tool_output, + status, started_at, completed_at + FROM execution_steps + WHERE incident_id = %s AND session_id LIKE %s AND tool_name <> 'write_findings' + ORDER BY step_index ASC LIMIT %s""", + (OUTPUT_EXCERPT_MAX_CHARS + 1, incident_id, f"%::{escaped_agent_id}", MAX_HISTORY_ENTRIES), + ) + step_rows = cur.fetchall() + + history = history_from_step_rows(step_rows) + # Fall back to the archived JSONB once terminal (live steps may be pruned). + if not history and status in TERMINAL_STATUSES: + history = archived_history or [] + + return { + "agent_id": agent_id, + "status": status, + "body": _load_finding_body(storage_uri, originator_id or user_id, agent_id), + "tool_call_history": history, + } + + +def _load_finding_body(storage_uri, storage_user_id, agent_id) -> Optional[str]: + """Download a finding's markdown body from object storage, if it exists.""" + if not storage_uri: + return None + try: + from utils.storage.storage import get_storage_manager + + data = get_storage_manager(storage_user_id).download_bytes(storage_uri, storage_user_id) + return data.decode("utf-8") if isinstance(data, bytes) else str(data) + except Exception: + logger.warning("%s Could not fetch finding body for agent=%s", _LOG_PREFIX, agent_id) + return None + + +# --------------------------------------------------------------------------- +# Postmortems +# --------------------------------------------------------------------------- + +class PostmortemListArgs(BaseModel): + limit: int = Field(default=50, description="Max postmortems to return (1–100, default 50).") + offset: int = Field(default=0, description="Paging offset (>= 0, default 0).") + + +@introspection_tool +def postmortem_list(limit=50, offset=0, *, user_id, **_) -> dict: + """List all postmortems for the org with incident titles and export URLs.""" + limit, offset = clamp(limit, 1, 100), max(0, int(offset)) + + with _cursor(user_id) as (cur, org_id): + cur.execute( + """SELECT p.id, p.incident_id, p.generated_at, p.updated_at, i.alert_title, + p.confluence_page_url, p.jira_issue_url, p.notion_page_url + FROM postmortems p + LEFT JOIN incidents i ON p.incident_id = i.id + WHERE p.org_id = %s + ORDER BY p.generated_at DESC LIMIT %s OFFSET %s""", + (org_id, limit, offset), + ) + rows = cur.fetchall() + + postmortems = [ + { + "id": str(r[0]), "incident_id": str(r[1]), "incident_title": r[4], + "generated_at": iso_utc(r[2]), "updated_at": iso_utc(r[3]), + "confluence_url": r[5], "jira_url": r[6], "notion_url": r[7], + } + for r in rows + ] + return {"postmortems": postmortems, "count": len(postmortems)} + + +# --------------------------------------------------------------------------- +# Knowledge base memory +# --------------------------------------------------------------------------- + +class KbGetMemoryArgs(BaseModel): + pass + + +@introspection_tool +def kb_get_memory(*, user_id, **_) -> dict: + """Read the org's persistent knowledge base memory.""" + with _cursor(user_id) as (cur, org_id): + cur.execute( + """SELECT content, updated_at FROM knowledge_base_memory + WHERE org_id = %s ORDER BY updated_at DESC LIMIT 1""", + (org_id,), + ) + row = cur.fetchone() + + if row and row[0]: + return {"content": row[0], "updated_at": iso_utc(row[1])} + return {"content": "", "updated_at": None} + + +# --------------------------------------------------------------------------- +# Grafana alerts (webhook-ingested) +# --------------------------------------------------------------------------- + +class GrafanaListAlertsArgs(BaseModel): + state: Optional[str] = Field( + default=None, description="Filter by alert state (e.g. 'alerting', 'ok', 'pending')." + ) + limit: int = Field(default=50, description="Max alerts to return (1–100, default 50).") + + +@introspection_tool +def grafana_list_alerts(state=None, limit=50, *, user_id, **_) -> dict: + """List Grafana alerts ingested via webhook, optionally filtered by state.""" + limit = clamp(limit, 1, 100) + + with _cursor(user_id) as (cur, org_id): + where = "org_id = %s" + params = [org_id] + if state: + where += " AND alert_state = %s" + params.append(state) + + cur.execute( + f"""SELECT id, alert_uid, alert_title, alert_state, rule_name, + rule_url, dashboard_url, received_at + FROM grafana_alerts WHERE {where} + ORDER BY received_at DESC LIMIT %s""", + tuple(params + [limit]), + ) + rows = cur.fetchall() + + alerts = [ + { + "id": r[0], "alert_uid": r[1], "title": r[2], "state": r[3], + "rule_name": r[4], "rule_url": r[5], "dashboard_url": r[6], + "received_at": iso_utc(r[7]), + } + for r in rows + ] + return {"alerts": alerts, "count": len(alerts)} diff --git a/server/chat/backend/agent/tools/notion/postmortem.py b/server/chat/backend/agent/tools/notion/postmortem.py index c9fcf0d3b..583979be0 100644 --- a/server/chat/backend/agent/tools/notion/postmortem.py +++ b/server/chat/backend/agent/tools/notion/postmortem.py @@ -5,7 +5,6 @@ import logging import re from typing import Any, Dict, Optional -from uuid import UUID from pydantic import BaseModel, Field @@ -14,6 +13,7 @@ from routes.audit_routes import record_audit_event from utils.auth.stateless_auth import resolve_org_id from utils.db.connection_pool import db_pool +from utils.validation import is_valid_uuid from .common import build_rich_text, notion_tool_error, notion_tool_success @@ -35,14 +35,6 @@ def _looks_like_iso_date(value: Optional[str]) -> bool: return bool(value and _ISO_DATE_RE.match(value.strip())) -def _validate_uuid(value: str) -> bool: - try: - UUID(value) - return True - except (ValueError, TypeError): - return False - - def _find_property_by_type( db_schema: Dict[str, Any], target_type: str ) -> Optional[str]: @@ -259,14 +251,14 @@ def _create_action_items( client: NotionClient, postmortem_md: str, action_items_database_id: str, -) -> int: +) -> Dict[str, int]: """Create one Notion page per unchecked action item in the target DB. - Returns the number of pages successfully created. + Returns counts of created/failed pages. """ items = parse_action_items(postmortem_md or "") if not items: - return 0 + return {"created": 0, "failed": 0} try: db_schema = client.get_database(action_items_database_id) @@ -281,7 +273,7 @@ def _create_action_items( "[NOTION] Action-items DB %s has no title property", action_items_database_id, ) - return 0 + return {"created": 0, "failed": 0} people_key = _find_property_by_type(db_schema, "people") date_key = _find_property_by_type(db_schema, "date") @@ -357,7 +349,7 @@ def _export_postmortem_to_notion( ValueError: on invalid input or missing postmortem content. NotionAuthExpiredError: when stored Notion credentials cannot refresh. """ - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): raise ValueError("Invalid incident ID") if not database_id: raise ValueError("database_id is required") @@ -594,7 +586,7 @@ def notion_create_action_items( _ = session_id if not user_id: return notion_tool_error("user_id is required", code="missing_user") - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return notion_tool_error("Invalid incident ID", code="bad_input") if not action_items_database_id: return notion_tool_error( diff --git a/server/chat/backend/agent/tools/postmortem_tool.py b/server/chat/backend/agent/tools/postmortem_tool.py index 4895d83c3..b5ab4bc2e 100644 --- a/server/chat/backend/agent/tools/postmortem_tool.py +++ b/server/chat/backend/agent/tools/postmortem_tool.py @@ -8,20 +8,12 @@ import json import logging -import uuid as _uuid_mod from pydantic import BaseModel, Field -logger = logging.getLogger(__name__) - +from utils.validation import is_valid_uuid -def _is_valid_uuid(value: str) -> bool: - """Return True only if value is a well-formed UUID string.""" - try: - _uuid_mod.UUID(value) - return True - except (ValueError, AttributeError): - return False +logger = logging.getLogger(__name__) class GetPostmortemArgs(BaseModel): @@ -46,7 +38,7 @@ def get_postmortem( if not incident_id: return json.dumps({"error": "incident_id is required."}) - if not _is_valid_uuid(incident_id): + if not is_valid_uuid(incident_id): logger.warning( "[PostmortemTool] get_postmortem called with non-UUID incident_id=%r — rejecting", incident_id, @@ -107,7 +99,7 @@ def save_postmortem( if not incident_id: return json.dumps({"error": "incident_id is required."}) - if not _is_valid_uuid(incident_id): + if not is_valid_uuid(incident_id): logger.warning( "[PostmortemTool] save_postmortem called with non-UUID incident_id=%r — rejecting", incident_id, diff --git a/server/chat/backend/agent/tools/trigger_rca_tool.py b/server/chat/backend/agent/tools/trigger_rca_tool.py index a4af5310c..c70b9834c 100644 --- a/server/chat/backend/agent/tools/trigger_rca_tool.py +++ b/server/chat/backend/agent/tools/trigger_rca_tool.py @@ -1,10 +1,9 @@ """ Trigger RCA Tool -LLM-callable tool that bridges interactive chat with the background RCA pipeline. -When the user describes an operational incident in chat, the LLM calls this tool -to create an incident and dispatch a full automated RCA investigation using all -connected integrations. +LLM-callable tool that creates an incident and dispatches a full automated RCA +investigation using all connected integrations. Can be invoked from any agent +session (interactive chat, Slack, MCP, etc.). """ import json @@ -17,7 +16,7 @@ class TriggerRCAArgs(BaseModel): - """Arguments for triggering an RCA investigation from chat.""" + """Arguments for triggering an RCA investigation.""" issue_description: str = Field( description="What the user described — the operational issue or symptoms they're seeing" @@ -46,7 +45,7 @@ def trigger_rca( **kwargs, ) -> str: """ - Trigger a full background RCA investigation from an interactive chat session. + Trigger a full background RCA investigation. Creates an incident, broadcasts an SSE update, and dispatches the background RCA pipeline (same one used by webhook-triggered alerts) with all connected @@ -73,10 +72,16 @@ def trigger_rca( try: from chat.backend.agent.tools.cloud_tools import get_state_context state = get_state_context() - if state and getattr(state, "is_background", False): + # Block nested RCA only — background Slack/Celery sessions without an + # active incident investigation may still trigger a new RCA. + if ( + state + and getattr(state, "is_background", False) + and getattr(state, "incident_id", None) + ): return json.dumps({ "error": "Cannot trigger RCA from within a background RCA session. " - "This tool is only available in interactive chat." + "Nested RCA is not supported — finish the current investigation first." }) except Exception as e: logger.warning(f"[TriggerRCA] Could not check background state, allowing: {e}") diff --git a/server/chat/backend/agent/utils/tool_call_history.py b/server/chat/backend/agent/utils/tool_call_history.py new file mode 100644 index 000000000..d18d28371 --- /dev/null +++ b/server/chat/backend/agent/utils/tool_call_history.py @@ -0,0 +1,81 @@ +"""Shared rendering of RCA sub-agent tool-call history. + +Tool calls surface in several places that must render identically: the live +sub-agent capture (``sub_agent.py`` / ``tool_context_capture.py``), the findings +API (``incidents_findings.py``), and the internal-agent introspection tool +(``introspection_tools.py``). This module owns the single definition of the +provider-CLI prefixing, the field-size caps, and the ``execution_steps`` → entry +shaping so those copies can't drift out of sync. +""" + +from utils.query_helpers import iso_utc +from utils.text.text_utils import truncate + +# Field-size caps for persisted/rendered history entries. Centralized so the +# live capture and the archived JSONB blob always render with the same limits. +OUTPUT_EXCERPT_MAX_CHARS = 1000 +COMMAND_MAX_CHARS = 1024 +MAX_HISTORY_ENTRIES = 30 + +# Sub-agent lifecycle states that mean "done" — once terminal we can trust the +# archived history blob instead of re-reading live execution_steps. +TERMINAL_STATUSES = frozenset( + {"succeeded", "failed", "timeout", "cancelled", "inconclusive"} +) + +# Map a cloud_exec provider to its CLI prefix, mirroring the frontend's +# getProviderCli so the captured command already includes the prefix +# (e.g. "aws cloudwatch ...") and the UI can pick the right icon from +# command.startswith without needing a separate provider field. +PROVIDER_CLI = { + "aws": "aws", "gcp": "gcloud", "gcloud": "gcloud", "azure": "az", "az": "az", + "ovh": "ovhcloud", "ovhcloud": "ovhcloud", "scaleway": "scw", "scw": "scw", +} +RECOGNIZED_CLI_PREFIXES = ( + "aws ", "gcloud ", "gsutil ", "bq ", "az ", "ovhcloud ", "scw ", + "kubectl ", "helm ", "docker ", +) + + +def derive_command(tool_input, limit: int = COMMAND_MAX_CHARS) -> str: + """Render a readable CLI command from a tool's structured input. + + Prefixes a raw shell command with the implied provider CLI; otherwise falls + back to the first query-like field present. + """ + if not isinstance(tool_input, dict): + return "" + cmd = tool_input.get("command") + # A raw shell command — prefix with the provider CLI when one is implied. + if isinstance(cmd, str) and cmd.strip(): + provider = tool_input.get("provider") + cli = PROVIDER_CLI.get(str(provider).lower()) if provider else None + if cli and not cmd.lstrip().startswith(RECOGNIZED_CLI_PREFIXES): + cmd = f"{cli} {cmd.lstrip()}" + return truncate(cmd, limit) + # Otherwise fall back to whichever query-like field is present. + for key in ("query", "path", "promql"): + if tool_input.get(key): + return truncate(tool_input[key], limit) + return "" + + +def history_from_step_rows(rows) -> list[dict]: + """Shape ``execution_steps`` rows into tool-call history entries. + + Each row must be ``(tool_name, tool_input, tool_output, status, started_at, + completed_at)`` — matching the column order in the queries that feed this. + """ + return [ + { + "tool_name": tool_name or "unknown", + "args": tool_input, + "command": derive_command(tool_input), + "output_excerpt": truncate(tool_output, OUTPUT_EXCERPT_MAX_CHARS), + "is_error": status == "error", + "status": status, + "started_at": iso_utc(started_at), + "completed_at": iso_utc(completed_at), + } + for tool_name, tool_input, tool_output, status, started_at, completed_at in rows + ] diff --git a/server/chat/backend/agent/utils/tool_context_capture.py b/server/chat/backend/agent/utils/tool_context_capture.py index c4dc41a05..0bec680b1 100644 --- a/server/chat/backend/agent/utils/tool_context_capture.py +++ b/server/chat/backend/agent/utils/tool_context_capture.py @@ -35,6 +35,8 @@ from .llm_usage_tracker import LLMUsageTracker from utils.db.connection_pool import db_pool from utils.auth.stateless_auth import set_rls_context +from chat.backend.agent.utils.tool_call_history import OUTPUT_EXCERPT_MAX_CHARS +from utils.text.text_utils import truncate # Import langchain components - direct imports for LangChain 1.2.6+ from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.callbacks import BaseCallbackHandler @@ -51,18 +53,6 @@ def __init__(self, content: str, tool_call_id: str = "", **kwargs): logger = logging.getLogger(__name__) -# Bound tool output excerpts retained for rca_findings.tool_call_history JSONB. -_OUTPUT_EXCERPT_MAX_CHARS = 1000 - - -def _build_excerpt(output: Optional[str]) -> str: - if not output: - return "" - s = str(output) - if len(s) > _OUTPUT_EXCERPT_MAX_CHARS: - return s[:_OUTPUT_EXCERPT_MAX_CHARS] + "...[truncated]" - return s - # -------------------------------------------------------------------------------------- # Token counting utility (delegates to LLMUsageTracker for context management only) # -------------------------------------------------------------------------------------- @@ -247,7 +237,7 @@ def capture_tool_end(self, tool_call_id: str, output: str, is_error: bool = Fals tool_info = self.current_tool_calls[tool_call_id] # Compute excerpt outside the lock — pure string processing, no shared state. - output_excerpt = _build_excerpt(output) + output_excerpt = truncate(output, OUTPUT_EXCERPT_MAX_CHARS) # Mirror _record_step_end's payload-shape classification so the history # entry (and downstream rca_findings.tool_call_history) matches the # execution_steps row's status. diff --git a/server/chat/background/task.py b/server/chat/background/task.py index 215f57db3..62d839aeb 100644 --- a/server/chat/background/task.py +++ b/server/chat/background/task.py @@ -319,6 +319,13 @@ def _ensure_llm_context_history( _RCA_SOURCES = {'grafana', 'datadog', 'netdata', 'splunk', 'slack', 'google_chat', 'pagerduty', 'dynatrace', 'jenkins', 'cloudbees', 'spinnaker', 'newrelic', 'chat', 'opsgenie', 'incidentio', 'action'} _GUARDRAIL_BLOCKED_MSG = 'Action blocked by safety guardrails' +_GUARDRAIL_USER_MSG = ( + "Your message was blocked by our safety system. Please rephrase your request." +) +_ACTION_GUARDRAIL_USER_MSG = ( + "This action was blocked by safety guardrails. " + "The instructions may need to be rephrased to pass input validation." +) # Initialize Redis client at module load time - fails if Redis is unavailable _redis_client = get_redis_client() @@ -829,14 +836,20 @@ def run_background_chat( # Skip if already sent inside _execute_background_chat (early send for lower latency) if trigger_metadata and trigger_metadata.get('source') in ['slack', 'slack_button'] and not result.get('slack_sent_early'): try: - _send_response_to_slack(user_id, session_id, trigger_metadata) + slack_fallback = _GUARDRAIL_USER_MSG if result.get("guardrail_blocked") else None + _send_response_to_slack( + user_id, session_id, trigger_metadata, fallback_text=slack_fallback, + ) except Exception as e: logger.error(f"[BackgroundChat] Failed to send response to Slack: {e}", exc_info=True) # Send response back to Google Chat if this was triggered from Google Chat if trigger_metadata and trigger_metadata.get('source') in ['google_chat', 'google_chat_button']: try: - _send_response_to_google_chat(user_id, session_id, trigger_metadata) + gchat_fallback = _GUARDRAIL_USER_MSG if result.get("guardrail_blocked") else None + _send_response_to_google_chat( + user_id, session_id, trigger_metadata, fallback_text=gchat_fallback, + ) except Exception as e: logger.error(f"[BackgroundChat] Failed to send response to Google Chat: {e}", exc_info=True) @@ -1538,6 +1551,17 @@ async def _execute_background_chat( # never return to the caller (worker process exits during event loop teardown # due to MCP subprocess cleanup), so this must happen before we return. guardrail_blocked = getattr(state, "guardrail_blocked", False) + if guardrail_blocked: + # Streaming copies are stripped at end of workflow; persist a real bot + # message so Slack/Google Chat can replace the "Thinking..." placeholder. + source = (trigger_metadata or {}).get("source", "") + block_text = ( + _ACTION_GUARDRAIL_USER_MSG + if source == "action" + else _GUARDRAIL_USER_MSG + ) + _append_block_message(session_id, user_id, block_text) + action_notification_sent = False if trigger_metadata and trigger_metadata.get('source') == 'action': action_status = None @@ -1545,7 +1569,6 @@ async def _execute_background_chat( try: from services.actions.executor import update_action_run_status if guardrail_blocked: - _append_block_message(session_id, user_id, "This action was blocked by safety guardrails. The instructions may need to be rephrased to pass input validation.") update_action_run_status( run_id=trigger_metadata['run_id'], status='error', user_id=user_id, error_message=_GUARDRAIL_BLOCKED_MSG, @@ -1967,9 +1990,14 @@ def _update_incident_status(incident_id: str, status: str, user_id: str) -> None logger.error(f"[BackgroundChat] Failed to update incident {incident_id} status to '{status}': {e}") -def _send_response_to_slack(user_id: str, session_id: str, trigger_metadata: Dict[str, Any]) -> bool: +def _send_response_to_slack( + user_id: str, + session_id: str, + trigger_metadata: Dict[str, Any], + fallback_text: Optional[str] = None, +) -> bool: """Send Aurora's response back to the Slack channel after background chat completes. - + Returns True if a message was actually posted, False otherwise. """ try: @@ -1995,25 +2023,29 @@ def _send_response_to_slack(user_id: str, session_id: str, trigger_metadata: Dic ) row = cursor.fetchone() - if not row or not row[0]: - logger.warning(f"[BackgroundChat] No messages found in session {session_id}") - return False - - messages = row[0] - if isinstance(messages, str): - import json - messages = json.loads(messages) - - # Find the last assistant/bot message last_assistant_message = None - for msg in reversed(messages): - if msg.get('sender') in ('bot', 'assistant'): - last_assistant_message = msg.get('text') or msg.get('content') - break - + if row and row[0]: + messages = row[0] + if isinstance(messages, str): + messages = json.loads(messages) + + # Find the last assistant/bot message + for msg in reversed(messages): + if not isinstance(msg, dict): + continue + if msg.get('sender') in ('bot', 'assistant'): + last_assistant_message = msg.get('text') or msg.get('content') + break + if not last_assistant_message: - logger.warning(f"[BackgroundChat] No assistant message found in session {session_id}") - return False + # Guardrail block may leave no assistant message — use caller-provided fallback + if fallback_text: + last_assistant_message = fallback_text + else: + logger.warning( + f"[BackgroundChat] No assistant message found in session {session_id}" + ) + return False # Format the response for Slack (markdown conversion, length limits, etc.) formatted_message = format_response_for_slack(last_assistant_message) @@ -2102,7 +2134,12 @@ def _send_response_to_slack(user_id: str, session_id: str, trigger_metadata: Dic raise -def _send_response_to_google_chat(user_id: str, session_id: str, trigger_metadata: Dict[str, Any]) -> None: +def _send_response_to_google_chat( + user_id: str, + session_id: str, + trigger_metadata: Dict[str, Any], + fallback_text: Optional[str] = None, +) -> None: """Send Aurora's response back to Google Chat after background chat completes.""" try: from routes.google_chat.google_chat_events_helpers import format_response_for_google_chat @@ -2125,23 +2162,27 @@ def _send_response_to_google_chat(user_id: str, session_id: str, trigger_metadat ) row = cursor.fetchone() - if not row or not row[0]: - logger.warning(f"[BackgroundChat] No messages found in session {session_id}") - return - - messages = row[0] - if isinstance(messages, str): - messages = json.loads(messages) - last_assistant_message = None - for msg in reversed(messages): - if msg.get('sender') in ('bot', 'assistant'): - last_assistant_message = msg.get('text') or msg.get('content') - break + if row and row[0]: + messages = row[0] + if isinstance(messages, str): + messages = json.loads(messages) + + for msg in reversed(messages): + if not isinstance(msg, dict): + continue + if msg.get('sender') in ('bot', 'assistant'): + last_assistant_message = msg.get('text') or msg.get('content') + break if not last_assistant_message: - logger.warning(f"[BackgroundChat] No assistant message found in session {session_id}") - return + if fallback_text: + last_assistant_message = fallback_text + else: + logger.warning( + f"[BackgroundChat] No assistant message found in session {session_id}" + ) + return formatted_message = format_response_for_google_chat(last_assistant_message) diff --git a/server/routes/actions/actions_routes.py b/server/routes/actions/actions_routes.py index 3d6cd4e3c..1f93504bf 100644 --- a/server/routes/actions/actions_routes.py +++ b/server/routes/actions/actions_routes.py @@ -1,24 +1,17 @@ """CRUD + trigger routes for Aurora Actions.""" import json import logging -import re from datetime import datetime, timezone from flask import jsonify, request from utils.db.connection_pool import db_pool from utils.auth.rbac_decorators import require_permission from utils.auth.stateless_auth import set_rls_context +from utils.validation import is_valid_uuid from services.actions.system_actions import seed_system_actions, SYSTEM_ACTIONS -_UUID_RE = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.I) - _ERR_INTERNAL = "Internal server error" - -def _validate_uuid(value: str) -> bool: - return bool(_UUID_RE.match(value)) - - _ERR_NOT_FOUND = "Action not found" from . import actions_bp @@ -260,7 +253,7 @@ def create_action(user_id): @actions_bp.route("/", methods=["GET"]) @require_permission("actions", "read") def get_action(user_id, action_id): - if not _validate_uuid(action_id): + if not is_valid_uuid(action_id): return jsonify({"error": _ERR_NOT_FOUND}), 404 return _get_action_response(action_id) @@ -318,7 +311,7 @@ def _get_action_response(action_id): @actions_bp.route("/", methods=["PUT"]) @require_permission("actions", "write") def update_action(user_id, action_id): - if not _validate_uuid(action_id): + if not is_valid_uuid(action_id): return jsonify({"error": _ERR_NOT_FOUND}), 404 body = request.get_json(silent=True) or {} @@ -353,7 +346,7 @@ def update_action(user_id, action_id): @actions_bp.route("/", methods=["DELETE"]) @require_permission("actions", "write") def delete_action(user_id, action_id): - if not _validate_uuid(action_id): + if not is_valid_uuid(action_id): return jsonify({"error": _ERR_NOT_FOUND}), 404 try: with db_pool.get_connection() as conn: @@ -376,7 +369,7 @@ def delete_action(user_id, action_id): @require_permission("actions", "write") def restore_default(user_id, action_id): """Restore a system action's instructions to the built-in default.""" - if not _validate_uuid(action_id): + if not is_valid_uuid(action_id): return jsonify({"error": _ERR_NOT_FOUND}), 404 try: with db_pool.get_connection() as conn: @@ -404,12 +397,12 @@ def restore_default(user_id, action_id): @actions_bp.route("//trigger", methods=["POST"]) @require_permission("actions", "write") def trigger_action(user_id, action_id): - if not _validate_uuid(action_id): + if not is_valid_uuid(action_id): return jsonify({"error": _ERR_NOT_FOUND}), 404 body = request.get_json(silent=True) or {} trigger_context = {} if body.get("incident_id"): - if not _validate_uuid(body["incident_id"]): + if not is_valid_uuid(body["incident_id"]): return jsonify({"error": "Invalid incident_id format"}), 400 trigger_context["incident_id"] = body["incident_id"] if body.get("trigger_label"): @@ -448,7 +441,7 @@ def trigger_action(user_id, action_id): @actions_bp.route("//runs", methods=["GET"]) @require_permission("actions", "read") def list_runs(user_id, action_id): - if not _validate_uuid(action_id): + if not is_valid_uuid(action_id): return jsonify({"error": _ERR_NOT_FOUND}), 404 try: limit = max(min(int(request.args.get("limit", 50)), 200), 1) diff --git a/server/routes/artifact_routes.py b/server/routes/artifact_routes.py index 23f2b0e21..6cbc69ca6 100644 --- a/server/routes/artifact_routes.py +++ b/server/routes/artifact_routes.py @@ -8,10 +8,7 @@ """ import logging -from datetime import timezone from functools import wraps -from typing import Optional -from uuid import UUID import psycopg2.errors from flask import Blueprint, jsonify, request @@ -21,6 +18,8 @@ from utils.auth.rbac_decorators import require_permission from utils.auth.stateless_auth import get_org_id_from_request, set_rls_context from utils.db.connection_pool import db_pool +from utils.query_helpers import iso_utc +from utils.validation import is_valid_uuid logger = logging.getLogger(__name__) @@ -31,22 +30,6 @@ _ARTIFACT_NOT_FOUND = "Artifact not found" -def _validate_uuid(value: str) -> bool: - try: - UUID(value) - return True - except (ValueError, TypeError): - return False - - -def _format_timestamp(ts) -> Optional[str]: - if not ts: - return None - if ts.tzinfo is None: - ts = ts.replace(tzinfo=timezone.utc) - return ts.isoformat() - - def _serialize_artifact(row, *, include_content: bool) -> dict: """Build a camelCase artifact dict from a row of (id, title, content, last_edited_by, created_at, updated_at, version_number). @@ -55,8 +38,8 @@ def _serialize_artifact(row, *, include_content: bool) -> dict: "id": str(row[0]), "title": row[1], "lastEditedBy": row[3], - "createdAt": _format_timestamp(row[4]), - "updatedAt": _format_timestamp(row[5]), + "createdAt": iso_utc(row[4]), + "updatedAt": iso_utc(row[5]), "version": row[6] or 0, } if include_content: @@ -70,7 +53,7 @@ def with_artifact(fn): """ @wraps(fn) def wrapper(user_id, artifact_id, *args, **kwargs): - if not _validate_uuid(artifact_id): + if not is_valid_uuid(artifact_id): return jsonify({"error": "Invalid artifact ID"}), 400 org_id = get_org_id_from_request() @@ -277,7 +260,7 @@ def list_artifact_versions(user_id, artifact_id, *, org_id, conn, cursor, **kwar "id": str(row[0]), "versionNumber": row[1], "source": row[2], - "createdAt": _format_timestamp(row[3]), + "createdAt": iso_utc(row[3]), "generationSessionId": str(row[4]) if row[4] else None, } for row in rows @@ -289,7 +272,7 @@ def list_artifact_versions(user_id, artifact_id, *, org_id, conn, cursor, **kwar @require_permission("artifacts", "read") @with_artifact def get_artifact_version(user_id, artifact_id, version_id, *, org_id, conn, cursor, **kwargs): - if not _validate_uuid(version_id): + if not is_valid_uuid(version_id): return jsonify({"error": "Invalid version ID"}), 400 cursor.execute( @@ -309,7 +292,7 @@ def get_artifact_version(user_id, artifact_id, version_id, *, org_id, conn, curs "versionNumber": row[1], "source": row[2], "content": row[3], - "createdAt": _format_timestamp(row[4]), + "createdAt": iso_utc(row[4]), } }) @@ -318,7 +301,7 @@ def get_artifact_version(user_id, artifact_id, version_id, *, org_id, conn, curs @require_permission("artifacts", "write") @with_artifact def restore_artifact_version(user_id, artifact_id, version_id, *, org_id, conn, cursor, **kwargs): - if not _validate_uuid(version_id): + if not is_valid_uuid(version_id): return jsonify({"error": "Invalid version ID"}), 400 cursor.execute( diff --git a/server/routes/audit_routes.py b/server/routes/audit_routes.py index 74bec2cb4..0f5215325 100644 --- a/server/routes/audit_routes.py +++ b/server/routes/audit_routes.py @@ -5,6 +5,7 @@ from utils.auth.rbac_decorators import require_permission from utils.auth.stateless_auth import get_org_id_from_request from utils.db.connection_pool import db_pool +from utils.query_helpers import iso_utc logger = logging.getLogger(__name__) @@ -44,10 +45,6 @@ def record_audit_event(org_id, user_id, action, resource_type, logger.exception("[AUDIT] Failed to record audit event: %s/%s", action, resource_type) -def _iso(val): - return val.isoformat() if hasattr(val, "isoformat") else val - - def _synthetic_events(cur, org_id, pg_interval): """Pull activity from core tables to fill gaps in the audit_log.""" events = [] @@ -63,7 +60,7 @@ def _synthetic_events(cur, org_id, pg_interval): "action": "member_joined", "resource_type": "user", "resource_id": row[0], "detail": {"email": row[1], "name": row[2], "role": row[3] or "viewer"}, - "ip_address": None, "created_at": _iso(row[4]), + "ip_address": None, "created_at": iso_utc(row[4]), }) cur.execute(""" @@ -78,7 +75,7 @@ def _synthetic_events(cur, org_id, pg_interval): "action": "incident_created", "resource_type": "incident", "resource_id": str(row[0]), "detail": {"source": row[1], "title": row[2], "severity": row[3], "status": row[4]}, - "ip_address": None, "created_at": _iso(row[5]), + "ip_address": None, "created_at": iso_utc(row[5]), }) cur.execute(""" @@ -96,7 +93,7 @@ def _synthetic_events(cur, org_id, pg_interval): "action": "connector_added", "resource_type": "connector", "resource_id": row[0], "detail": {"provider": row[0], "user_name": row[3] or row[4]}, - "ip_address": None, "created_at": _iso(row[1]), + "ip_address": None, "created_at": iso_utc(row[1]), }) cur.execute(""" @@ -116,7 +113,7 @@ def _synthetic_events(cur, org_id, pg_interval): "action": "connector_added", "resource_type": "connector", "resource_id": row[0], "detail": {"provider": row[0], "user_name": row[3] or row[4]}, - "ip_address": None, "created_at": _iso(row[1]), + "ip_address": None, "created_at": iso_utc(row[1]), }) return events @@ -178,7 +175,7 @@ def get_audit_log(user_id): for row in audit_rows: for k, v in row.items(): if hasattr(v, "isoformat"): - row[k] = v.isoformat() + row[k] = iso_utc(v) all_events = audit_rows + synthetic diff --git a/server/routes/incident_feedback/routes.py b/server/routes/incident_feedback/routes.py index 1abe8b3b8..d982616fe 100644 --- a/server/routes/incident_feedback/routes.py +++ b/server/routes/incident_feedback/routes.py @@ -12,7 +12,7 @@ ) from utils.auth.rbac_decorators import require_permission, require_auth_only from routes.incident_feedback.weaviate_client import store_good_rca -from uuid import UUID +from utils.validation import is_valid_uuid logger = logging.getLogger(__name__) @@ -25,15 +25,6 @@ VALID_FEEDBACK_TYPES = {"helpful", "not_helpful"} -def _validate_uuid(value: str) -> bool: - """Validate that a string is a valid UUID.""" - try: - UUID(value) - return True - except (ValueError, TypeError): - return False - - def _is_aurora_learn_enabled(user_id: str) -> bool: """Check if Aurora Learn is enabled for a user. Defaults to True.""" setting = get_user_preference(user_id, AURORA_LEARN_PREFERENCE_KEY, default=True) @@ -49,7 +40,7 @@ def _is_aurora_learn_enabled(user_id: str) -> bool: @require_permission("incidents", "write") def submit_feedback(user_id, incident_id: str): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 # Check if Aurora Learn is enabled @@ -232,7 +223,7 @@ def submit_feedback(user_id, incident_id: str): @require_permission("incidents", "read") def get_feedback(user_id, incident_id: str): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 try: diff --git a/server/routes/incidents_findings.py b/server/routes/incidents_findings.py index b04946da3..0749ae1ce 100644 --- a/server/routes/incidents_findings.py +++ b/server/routes/incidents_findings.py @@ -5,15 +5,21 @@ import logging import re -from uuid import UUID from flask import Blueprint, jsonify +from chat.backend.agent.utils.tool_call_history import ( + MAX_HISTORY_ENTRIES, + OUTPUT_EXCERPT_MAX_CHARS, + TERMINAL_STATUSES, + history_from_step_rows, +) from utils.auth.rbac_decorators import require_permission from utils.auth.stateless_auth import set_rls_context from utils.db.connection_pool import db_pool from utils.log_sanitizer import hash_for_log, sanitize from utils.storage.storage import get_storage_manager +from utils.validation import is_valid_uuid logger = logging.getLogger(__name__) @@ -21,88 +27,11 @@ _LOG_PREFIX = "[Findings]" _AGENT_ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$") -# Keep in sync with _OUTPUT_EXCERPT_MAX_CHARS in tool_context_capture.py and -# _MAX_HISTORY_ENTRIES / _entry_command's default limit in sub_agent.py — live -# and archived tool_call_history entries must render identically. -_OUTPUT_EXCERPT_MAX_CHARS = 1000 -_COMMAND_MAX_CHARS = 1024 -_MAX_HISTORY_ENTRIES = 30 - -_TERMINAL_STATUSES = frozenset({"succeeded", "failed", "timeout", "cancelled", "inconclusive"}) - -_PROVIDER_CLI = { - "aws": "aws", - "gcp": "gcloud", "gcloud": "gcloud", - "azure": "az", "az": "az", - "ovh": "ovhcloud", "ovhcloud": "ovhcloud", - "scaleway": "scw", "scw": "scw", -} -_RECOGNIZED_CLI_PREFIXES = ( - "aws ", "gcloud ", "gsutil ", "bq ", "az ", - "ovhcloud ", "scw ", - "kubectl ", "helm ", "docker ", -) - - -def _validate_uuid(value: str) -> bool: - try: - UUID(value) - return True - except (ValueError, AttributeError, TypeError): - return False - - -def _excerpt(output) -> str: - if not output: - return "" - s = output if isinstance(output, str) else str(output) - if len(s) > _OUTPUT_EXCERPT_MAX_CHARS: - return s[:_OUTPUT_EXCERPT_MAX_CHARS] + "...[truncated]" - return s - - -def _truncate(s: str, limit: int) -> str: - return s if len(s) <= limit else s[:limit] + "...[truncated]" - - -def _derive_command(tool_input) -> str: - if not isinstance(tool_input, dict): - return "" - cmd = tool_input.get("command") - if isinstance(cmd, str) and cmd.strip(): - provider = tool_input.get("provider") - if provider: - cli = _PROVIDER_CLI.get(str(provider).lower()) - if cli and not cmd.lstrip().startswith(_RECOGNIZED_CLI_PREFIXES): - cmd = f"{cli} {cmd.lstrip()}" - return _truncate(cmd, _COMMAND_MAX_CHARS) - for key in ("query", "path", "promql"): - v = tool_input.get(key) - if v: - return _truncate(str(v), _COMMAND_MAX_CHARS) - return "" - - -def _build_history_from_steps(rows) -> list: - return [ - { - "tool_name": tool_name or "unknown", - "args": tool_input, - "command": _derive_command(tool_input), - "output_excerpt": _excerpt(tool_output), - "is_error": status == "error", - "status": status, - "started_at": started_at.isoformat() if started_at else None, - "completed_at": completed_at.isoformat() if completed_at else None, - } - for tool_name, tool_input, tool_output, status, started_at, completed_at in rows - ] - @findings_bp.route("/api/incidents//findings", methods=["GET"]) @require_permission("incidents", "read") def list_findings(user_id, incident_id: str): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 try: @@ -159,7 +88,7 @@ def list_findings(user_id, incident_id: str): @findings_bp.route("/api/incidents//findings/", methods=["GET"]) @require_permission("incidents", "read") def get_finding_body(user_id, incident_id: str, agent_id: str): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 if not _AGENT_ID_RE.match(agent_id): return jsonify({"error": "Invalid agent ID format"}), 400 @@ -183,7 +112,7 @@ def get_finding_body(user_id, incident_id: str, agent_id: str): # before the sub-agent terminates and persists the JSONB blob. # Suffix-match: child_session_id on rca_findings is NULL for # in-flight sub-agents, so we can't use exact session_id =. - # +1 char so _excerpt still detects overflow and appends "...[truncated]". + # +1 char so truncate() still detects overflow and appends "...[truncated]". cursor.execute( """ SELECT tool_name, tool_input, @@ -197,18 +126,18 @@ def get_finding_body(user_id, incident_id: str, agent_id: str): LIMIT %s """, ( - _OUTPUT_EXCERPT_MAX_CHARS + 1, + OUTPUT_EXCERPT_MAX_CHARS + 1, incident_id, f"%::{agent_id}", - _MAX_HISTORY_ENTRIES, + MAX_HISTORY_ENTRIES, ), ) step_rows = cursor.fetchall() - history = _build_history_from_steps(step_rows) + history = history_from_step_rows(step_rows) # Fall back to the archived JSONB only when terminal — covers old # incidents whose execution_steps rows have been pruned. - if not history and status in _TERMINAL_STATUSES: + if not history and status in TERMINAL_STATUSES: history = tool_call_history or [] if not storage_uri: # Body not yet written. Return 200 with status so the client can keep diff --git a/server/routes/incidents_routes.py b/server/routes/incidents_routes.py index f781077e7..08f686b02 100644 --- a/server/routes/incidents_routes.py +++ b/server/routes/incidents_routes.py @@ -2,17 +2,17 @@ import json import logging -from datetime import timezone from routes.audit_routes import record_audit_event as _record_audit_event from flask import Blueprint, jsonify, request from utils.db.connection_pool import db_pool +from utils.query_helpers import iso_utc from utils.auth.token_management import get_token_data from utils.auth.rbac_decorators import require_permission from utils.auth.stateless_auth import get_org_id_from_request, set_rls_context from utils.log_sanitizer import hash_for_log, sanitize from chat.background.task import run_background_chat from typing import List, Dict, Any, Optional -from uuid import UUID +from utils.validation import is_valid_uuid from chat.background.task import create_background_chat_session, run_background_chat logger = logging.getLogger(__name__) @@ -26,25 +26,6 @@ TITLE_MAX_LENGTH = 50 -def _format_timestamp(ts) -> Optional[str]: - """Format timestamp ensuring UTC timezone.""" - if not ts: - return None - # If naive datetime, assume it's UTC (PostgreSQL TIMESTAMP without timezone) - if ts.tzinfo is None: - ts = ts.replace(tzinfo=timezone.utc) - return ts.isoformat() - - -def _validate_uuid(value: str) -> bool: - """Validate that a string is a valid UUID.""" - try: - UUID(value) - return True - except (ValueError, TypeError): - return False - - def _parse_suggestion_id(suggestion_id: str) -> Optional[int]: """Parse and validate a suggestion ID string to int.""" try: @@ -247,12 +228,12 @@ def _format_incident_response( if aurora_chat_session_id else None, "activeTab": active_tab or "thoughts", - "startedAt": _format_timestamp(started_at), - "analyzedAt": _format_timestamp(analyzed_at), - "resolvedAt": _format_timestamp(resolved_at), - "alertFiredAt": _format_timestamp(alert_fired_at), - "createdAt": _format_timestamp(created_at), - "updatedAt": _format_timestamp(updated_at), + "startedAt": iso_utc(started_at), + "analyzedAt": iso_utc(analyzed_at), + "resolvedAt": iso_utc(resolved_at), + "alertFiredAt": iso_utc(alert_fired_at), + "createdAt": iso_utc(created_at), + "updatedAt": iso_utc(updated_at), } # Add metadata fields to alert object if available @@ -357,7 +338,7 @@ def get_incidents(user_id): def get_incident(user_id, incident_id: str): # Validate incident_id is a valid UUID - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 org_id = get_org_id_from_request() @@ -650,7 +631,7 @@ def get_incident(user_id, incident_id: str): "correlationDetails": arow[7] if isinstance(arow[7], dict) else {}, - "receivedAt": _format_timestamp(arow[8]), + "receivedAt": iso_utc(arow[8]), } ) incident["correlatedAlerts"] = correlated_alerts @@ -686,8 +667,8 @@ def get_incident(user_id, incident_id: str): "type": srow[S_TYPE] or "diagnostic", "risk": srow[S_RISK] or "safe", "command": srow[S_CMD], - "createdAt": _format_timestamp(srow[S_CREATED]), - "executedAt": _format_timestamp(srow[S_EXECUTED_AT]), + "createdAt": iso_utc(srow[S_CREATED]), + "executedAt": iso_utc(srow[S_EXECUTED_AT]), "executionSessionId": str(srow[S_EXEC_SESSION]) if srow[S_EXEC_SESSION] else None, "executionStatus": srow[S_EXEC_STATUS], } @@ -703,7 +684,7 @@ def get_incident(user_id, incident_id: str): "prUrl": srow[S_PR_URL], "prNumber": srow[S_PR_NUM], "createdBranch": srow[S_BRANCH], - "appliedAt": _format_timestamp(srow[S_APPLIED]), + "appliedAt": iso_utc(srow[S_APPLIED]), } ) suggestions.append(suggestion) @@ -725,10 +706,10 @@ def get_incident(user_id, incident_id: str): thoughts.append( { "id": str(trow[0]), - "timestamp": _format_timestamp(trow[2]), + "timestamp": iso_utc(trow[2]), "content": trow[3], "type": trow[4] or "analysis", - "createdAt": _format_timestamp(trow[5]), + "createdAt": iso_utc(trow[5]), } ) @@ -762,8 +743,8 @@ def get_incident(user_id, incident_id: str): "toolName": crow[2] or "Unknown", "command": crow[3] or "", "output": crow[4] or "", - "executedAt": _format_timestamp(crow[5]), - "createdAt": _format_timestamp(crow[6]), + "executedAt": iso_utc(crow[5]), + "createdAt": iso_utc(crow[6]), } ) @@ -795,8 +776,8 @@ def get_incident(user_id, incident_id: str): "title": csrow[1], "messages": csrow[2] if csrow[2] else [], "status": csrow[3] or "active", - "createdAt": _format_timestamp(csrow[4]), - "updatedAt": _format_timestamp(csrow[5]), + "createdAt": iso_utc(csrow[4]), + "updatedAt": iso_utc(csrow[5]), } ) @@ -945,7 +926,7 @@ def get_incident(user_id, incident_id: str): @require_permission("incidents", "read") def get_incident_alerts(user_id, incident_id: str): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 org_id = get_org_id_from_request() @@ -986,7 +967,7 @@ def get_incident_alerts(user_id, incident_id: str): "correlationDetails": arow[7] if isinstance(arow[7], dict) else {}, - "receivedAt": _format_timestamp(arow[8]), + "receivedAt": iso_utc(arow[8]), } ) @@ -1014,7 +995,7 @@ def get_incident_alerts(user_id, incident_id: str): def update_incident(user_id, incident_id: str): # Validate incident_id is a valid UUID - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 org_id = get_org_id_from_request() @@ -1154,7 +1135,7 @@ def update_incident(user_id, incident_id: str): @require_permission("incidents", "write") def incident_chat(user_id, incident_id: str): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID format"}), 400 org_id = get_org_id_from_request() @@ -1184,7 +1165,7 @@ def incident_chat(user_id, incident_id: str): sanitize(existing_session_id), ) - if existing_session_id and not _validate_uuid(existing_session_id): + if existing_session_id and not is_valid_uuid(existing_session_id): return jsonify({"error": "Invalid session ID format"}), 400 try: @@ -1705,7 +1686,7 @@ def apply_fix_suggestion(user_id, suggestion_id: str): @require_permission("incidents", "write") def merge_alert_to_incident(user_id, target_incident_id: str): - if not _validate_uuid(target_incident_id): + if not is_valid_uuid(target_incident_id): return jsonify({"error": "Invalid target incident ID format"}), 400 org_id = get_org_id_from_request() @@ -1715,7 +1696,7 @@ def merge_alert_to_incident(user_id, target_incident_id: str): return jsonify({"error": "Missing request body"}), 400 source_incident_id = data.get("sourceIncidentId") - if not source_incident_id or not _validate_uuid(source_incident_id): + if not source_incident_id or not is_valid_uuid(source_incident_id): return jsonify({"error": "Invalid or missing sourceIncidentId"}), 400 if source_incident_id == target_incident_id: @@ -1995,7 +1976,7 @@ def get_recent_unlinked_incidents(user_id): """ params = [org_id] - if exclude_id and _validate_uuid(exclude_id): + if exclude_id and is_valid_uuid(exclude_id): query += " AND id != %s" params.append(exclude_id) @@ -2014,7 +1995,7 @@ def get_recent_unlinked_incidents(user_id): "sourceType": row[4], "status": row[5], "auroraStatus": row[6], - "createdAt": _format_timestamp(row[7]), + "createdAt": iso_utc(row[7]), }) return jsonify({"incidents": incidents}), 200 @@ -2031,7 +2012,7 @@ def get_recent_unlinked_incidents(user_id): @require_permission("incidents", "read") def get_incident_action_runs(user_id, incident_id: str): """Return action runs linked to a specific incident.""" - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID"}), 400 org_id = get_org_id_from_request() @@ -2058,8 +2039,8 @@ def get_incident_action_runs(user_id, incident_id: str): r["chat_session_id"] = str(r["chat_session_id"]) if r["chat_session_id"] else None if r["started_at"] and r["completed_at"]: r["duration_ms"] = max(0, int((r["completed_at"] - r["started_at"]).total_seconds() * 1000)) - r["started_at"] = _format_timestamp(r["started_at"]) - r["completed_at"] = _format_timestamp(r["completed_at"]) + r["started_at"] = iso_utc(r["started_at"]) + r["completed_at"] = iso_utc(r["completed_at"]) return jsonify({"runs": rows}) except Exception: diff --git a/server/routes/metrics_routes.py b/server/routes/metrics_routes.py index 9bf2908f4..cc98574f3 100644 --- a/server/routes/metrics_routes.py +++ b/server/routes/metrics_routes.py @@ -5,25 +5,13 @@ from utils.db.connection_pool import db_pool from utils.auth.rbac_decorators import require_permission from utils.auth.stateless_auth import set_rls_context +from utils.metrics_periods import period_to_interval as _get_period_interval logger = logging.getLogger(__name__) metrics_bp = Blueprint("metrics", __name__) _LOG_PREFIX = "[Metrics]" -# Valid period values and their PostgreSQL interval strings -_PERIOD_MAP = { - "7d": "7 days", - "30d": "30 days", - "90d": "90 days", - "180d": "180 days", - "365d": "365 days", -} - - -def _get_period_interval(period_str: str) -> str: - return _PERIOD_MAP.get(period_str, "30 days") - def _parse_window_hours(default: int = 4) -> tuple[int | None, tuple | None]: """Parse and validate the window_hours query parameter. diff --git a/server/routes/postmortem_routes.py b/server/routes/postmortem_routes.py index e2131f21b..b6b1eb097 100644 --- a/server/routes/postmortem_routes.py +++ b/server/routes/postmortem_routes.py @@ -2,10 +2,8 @@ import logging import time -from datetime import timezone from functools import wraps from typing import Any, Dict, Optional -from uuid import UUID import requests from flask import Blueprint, jsonify, request @@ -21,6 +19,8 @@ from connectors.confluence_connector.auth import refresh_access_token from utils.db.connection_pool import db_pool from utils.log_sanitizer import sanitize +from utils.query_helpers import iso_utc +from utils.validation import is_valid_uuid logger = logging.getLogger(__name__) @@ -28,24 +28,6 @@ _LOG_PREFIX = "[Postmortem]" -def _validate_uuid(value: str) -> bool: - """Validate that a string is a valid UUID.""" - try: - UUID(value) - return True - except (ValueError, TypeError): - return False - - -def _format_timestamp(ts) -> Optional[str]: - """Format timestamp ensuring UTC timezone.""" - if not ts: - return None - if ts.tzinfo is None: - ts = ts.replace(tzinfo=timezone.utc) - return ts.isoformat() - - def _create_version( cursor, postmortem_id: str, org_id: str, user_id: str, content: str, source: str = "manual", *, set_current: bool = True, @@ -83,7 +65,7 @@ def with_incident_postmortem(require_postmortem=False): def decorator(fn): @wraps(fn) def wrapper(user_id, incident_id, *args, **kwargs): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID"}), 400 org_id = get_org_id_from_request() @@ -184,18 +166,18 @@ def get_postmortem(user_id, incident_id, *, org_id, conn, cursor, postmortem_id, "incidentId": str(row[1]), "userId": row[2], "content": row[3], - "generatedAt": _format_timestamp(row[4]), - "updatedAt": _format_timestamp(row[5]), + "generatedAt": iso_utc(row[4]), + "updatedAt": iso_utc(row[5]), "confluencePageId": row[6], "confluencePageUrl": row[7], - "confluenceExportedAt": _format_timestamp(row[8]), + "confluenceExportedAt": iso_utc(row[8]), "jiraIssueId": row[9], "jiraIssueKey": row[10], "jiraIssueUrl": row[11], - "jiraExportedAt": _format_timestamp(row[12]), + "jiraExportedAt": iso_utc(row[12]), "notionPageId": row[13], "notionPageUrl": row[14], - "notionExportedAt": _format_timestamp(row[15]), + "notionExportedAt": iso_utc(row[15]), "notionDatabaseId": row[16], "generationSessionId": row[17], } @@ -263,7 +245,7 @@ def list_postmortem_versions(user_id, incident_id, *, org_id, conn, cursor, post "versionNumber": row[1], "source": row[2], "userId": row[3], - "createdAt": _format_timestamp(row[4]), + "createdAt": iso_utc(row[4]), "generationSessionId": str(row[5]) if row[5] else None, } for row in rows @@ -276,7 +258,7 @@ def list_postmortem_versions(user_id, incident_id, *, org_id, conn, cursor, post @with_incident_postmortem(require_postmortem=False) def get_postmortem_version(user_id, incident_id, version_id, *, org_id, conn, cursor, postmortem_id, **kwargs): """Get a specific postmortem version content.""" - if not _validate_uuid(version_id): + if not is_valid_uuid(version_id): return jsonify({"error": "Invalid version ID"}), 400 cursor.execute( @@ -298,7 +280,7 @@ def get_postmortem_version(user_id, incident_id, version_id, *, org_id, conn, cu "source": row[2], "userId": row[3], "content": row[4], - "createdAt": _format_timestamp(row[5]), + "createdAt": iso_utc(row[5]), } }) @@ -308,7 +290,7 @@ def get_postmortem_version(user_id, incident_id, version_id, *, org_id, conn, cu @with_incident_postmortem(require_postmortem=True) def restore_postmortem_version(user_id, incident_id, version_id, *, org_id, conn, cursor, postmortem_id, **kwargs): """Restore a previous postmortem version as the current content.""" - if not _validate_uuid(version_id): + if not is_valid_uuid(version_id): return jsonify({"error": "Invalid version ID"}), 400 cursor.execute( @@ -342,7 +324,7 @@ def restore_postmortem_version(user_id, incident_id, version_id, *, org_id, conn @require_permission("postmortems", "write") def regenerate_postmortem(user_id, incident_id): """Trigger postmortem generation (or regeneration) via the built-in action.""" - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID"}), 400 try: @@ -371,7 +353,7 @@ def regenerate_postmortem(user_id, incident_id): @require_permission("postmortems", "write") def export_to_confluence(user_id, incident_id): - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID"}), 400 try: @@ -532,7 +514,7 @@ def export_to_confluence(user_id, incident_id): @require_permission("postmortems", "write") def export_to_notion(user_id, incident_id): """Export postmortem to a Notion database.""" - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID"}), 400 try: data = request.get_json(force=True, silent=True) or {} @@ -632,18 +614,18 @@ def list_postmortems(user_id): "incidentId": str(row[1]), "incidentTitle": row[9], "content": row[3], - "generatedAt": _format_timestamp(row[4]), - "updatedAt": _format_timestamp(row[5]), + "generatedAt": iso_utc(row[4]), + "updatedAt": iso_utc(row[5]), "confluencePageId": row[6], "confluencePageUrl": row[7], - "confluenceExportedAt": _format_timestamp(row[8]), + "confluenceExportedAt": iso_utc(row[8]), "jiraIssueId": row[10], "jiraIssueKey": row[11], "jiraIssueUrl": row[12], - "jiraExportedAt": _format_timestamp(row[13]), + "jiraExportedAt": iso_utc(row[13]), "notionPageId": row[14], "notionPageUrl": row[15], - "notionExportedAt": _format_timestamp(row[16]), + "notionExportedAt": iso_utc(row[16]), "notionDatabaseId": row[17], } postmortems.append(postmortem) @@ -665,7 +647,7 @@ def list_postmortems(user_id): @require_permission("postmortems", "write") def export_to_jira(user_id, incident_id): """Export postmortem to Jira as a parent issue with subtasks for action items.""" - if not _validate_uuid(incident_id): + if not is_valid_uuid(incident_id): return jsonify({"error": "Invalid incident ID"}), 400 try: diff --git a/server/utils/metrics_periods.py b/server/utils/metrics_periods.py new file mode 100644 index 000000000..861555d5f --- /dev/null +++ b/server/utils/metrics_periods.py @@ -0,0 +1,22 @@ +"""Shared metric period helpers. + +A single source of truth for the ``7d/30d/90d/...`` period strings and their +PostgreSQL interval equivalents, used by the SRE metrics routes and the +internal agent's introspection tools so the two never drift apart. +""" + +# Accepted period values mapped to their PostgreSQL interval literal. +PERIOD_MAP = { + "7d": "7 days", + "30d": "30 days", + "90d": "90 days", + "180d": "180 days", + "365d": "365 days", +} + +DEFAULT_INTERVAL = "30 days" + + +def period_to_interval(period: str) -> str: + """Return the PostgreSQL interval literal for a period, defaulting to 30 days.""" + return PERIOD_MAP.get(period, DEFAULT_INTERVAL) diff --git a/server/utils/query_helpers.py b/server/utils/query_helpers.py new file mode 100644 index 000000000..80b38b444 --- /dev/null +++ b/server/utils/query_helpers.py @@ -0,0 +1,41 @@ +"""Small, generic helpers for turning SQL query results into JSON payloads. + +These were previously re-implemented in nearly every route and agent tool that +builds an API response from a cursor: cursor → list-of-dicts, UTC ISO-8601 +formatting (the duplicated ``_format_timestamp`` / ``_iso``), duration math, and +value clamping. Centralized here so callers import instead of copy-paste. +""" + +from datetime import timezone +from typing import Optional + + +def fetch_dicts(cursor) -> list[dict]: + """Fetch all remaining rows as dicts keyed by column name.""" + cols = [d[0] for d in cursor.description] + return [dict(zip(cols, row)) for row in cursor.fetchall()] + + +def iso_utc(ts) -> Optional[str]: + """ISO-8601 string for a datetime, treating naive values as UTC. + + PostgreSQL ``TIMESTAMP`` columns come back naive; tag them UTC so clients + don't misread them as local time. + """ + if ts is None: + return None + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + return ts.isoformat() + + +def duration_ms(start, end) -> Optional[int]: + """Elapsed milliseconds between two datetimes, or None if either is missing.""" + if not (start and end): + return None + return max(0, int((end - start).total_seconds() * 1000)) + + +def clamp(value, low: int, high: int) -> int: + """Clamp an int-coercible value into the inclusive ``[low, high]`` range.""" + return max(low, min(int(value), high)) diff --git a/server/utils/text/text_utils.py b/server/utils/text/text_utils.py index 0733d066e..a9594d818 100644 --- a/server/utils/text/text_utils.py +++ b/server/utils/text/text_utils.py @@ -2,6 +2,21 @@ import re +_TRUNCATION_SUFFIX = "...[truncated]" + + +def truncate(value, limit: int, suffix: str = _TRUNCATION_SUFFIX) -> str: + """Coerce ``value`` to text and cap it at ``limit`` characters. + + Returns "" for ``None``, leaves short values untouched, and appends + ``suffix`` only when content was actually cut. Consolidates the many + copy-pasted ``_truncate``/``_excerpt`` helpers across routes and agent tools. + """ + if value is None: + return "" + s = value if isinstance(value, str) else str(value) + return s if len(s) <= limit else s[:limit] + suffix + def clean_markdown(text: str) -> str: """Strip markdown formatting from text for clean thought display. diff --git a/server/utils/validation.py b/server/utils/validation.py new file mode 100644 index 000000000..4c4c2ab08 --- /dev/null +++ b/server/utils/validation.py @@ -0,0 +1,12 @@ +"""Generic input validators shared across routes, tools, and tasks.""" + +import uuid + + +def is_valid_uuid(value) -> bool: + """Return True if ``value`` is a well-formed UUID string.""" + try: + uuid.UUID(str(value)) + return True + except (ValueError, AttributeError, TypeError): + return False