Skip to content
Merged
Show file tree
Hide file tree
Changes from 16 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 13 additions & 56 deletions server/chat/backend/agent/orchestrator/sub_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)."""
Expand Down Expand Up @@ -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"),
Expand All @@ -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",
Expand All @@ -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 []
Expand Down
203 changes: 198 additions & 5 deletions server/chat/backend/agent/tools/cloud_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
isiddharthsingh marked this conversation as resolved.
_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 (
Expand Down Expand Up @@ -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"))
Comment thread
coderabbitai[bot] marked this conversation as resolved.

# 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"))
Comment thread
coderabbitai[bot] marked this conversation as resolved.

Expand Down Expand Up @@ -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,
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

_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 <service> 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:
Expand Down
Loading
Loading