How Rescale Built an AI Triage Agent Using Strands Agents and AI-DLC

Traditional SDLC:  [Spec] ──> [Code] ──> [CI/CD Build] ──> [Deterministic Test] ──> [Manual Triage] ──> [Release]
AI-DLC Loop:       [Spec] ──> [AI Synthesis] ──> [Build] ──> [Test Execution]
                                 ▲                                 │
                                 └─── [AI Triage Agent] ◄──────────┘
                                      ├─ Log Parsing & Classification
                                      ├─ Root Cause Analysis (RCA)
                                      └─ Auto-remediation / PR Draft

 ┌────────────────────────────────────────────────────────────────────────┐
 │                       Continuous Integration (CI)                      │
 │                 (GitHub Actions / Jenkins / GitLab CI)                 │
 └───────────────────────────────────┬────────────────────────────────────┘
                                     │ Webhook / Event Stream
                                     ▼
 ┌────────────────────────────────────────────────────────────────────────┐
 │                     Strands Ingestion Pipeline                         │
 │  ├─ Telemetry Sanitizer (strips secrets, tokens, PII)                  │
 │  ├─ Noise Reducer (removes heartbeats, recurring info logs)            │
 │  └─ Chunking Engine (semantic code & stack-trace slicing)              │
 └───────────────────────────────────┬────────────────────────────────────┘
                                     │
                                     ▼
 ┌────────────────────────────────────────────────────────────────────────┐
 │                     Strands Agent Orchestration Core                   │
 │                                                                        │
 │  ┌───────────────────────┐          ┌───────────────────────────────┐  │
 │  │   Classifier Agent    │          │     Root Cause Analyzer       │  │
 │  │ (Infra vs Code vs UI) │ ───────> │  (Evaluates Diffs & Traces)   │  │
 │  └───────────────────────┘          └───────────────┬───────────────┘  │
 │             ▲                                       │                  │
 │             │ Vector Context                        │ Tool Calls       │
 │  ┌──────────┴────────────┐          ┌───────────────▼───────────────┐  │
 │  │  Historical Vector DB │          │      Action Executor          │  │
 │  │ (OpenSearch / pgvector)│         │ (Jira / Slack / GitHub API)   │  │
 │  └───────────────────────┘          └───────────────────────────────┘  │
 └───────────────────────────────────┬────────────────────────────────────┘
                                     │
                                     ▼
 ┌────────────────────────────────────────────────────────────────────────┐
 │               Developer Feedback Loop & Automated Actions              │
 │    - Auto-quarantine Flaky Test                                        │
 │    - Post Root Cause Analysis in PR Comments                            │
 │    - Rerun transient timeouts                                          │
 └────────────────────────────────────────────────────────────────────────┘

import os
import json
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field
import requests

# Pydantic Schema for Guaranteed LLM Output
class TriageAnalysis(BaseModel):
    failure_category: str = Field(description="Category: INFRASTRUCTURE, FLAKY, CODE_DEFECT, CONFIGURATION")
    root_cause_summary: str = Field(description="Concise technical summary of why the test failed")
    confidence_score: float = Field(description="Confidence score between 0.0 and 1.0")
    affected_files: List[str] = Field(description="Source files implicated by the trace")
    suggested_action: str = Field(description="Actionable next steps: RERUN, QUARANTINE, FILE_BUG, FIX_DIFF")

class TestTriageAgent:
    def __init__(self, api_key: str, base_url: str = "https://api.n1n.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def sanitize_log(self, raw_log: str, max_lines: int = 200) -> str:
        """Extract relevant failure window from raw console output."""
        lines = raw_log.splitlines()
        error_indices = [i for i, line in enumerate(lines) if "ERROR" in line or "FAIL" in line or "Traceback" in line]
        
        if not error_indices:
            # Return the tail if no explicit error tokens found
            return "\n".join(lines[-max_lines:])
        
        # Slices window starting slightly before first error until end
        start_line = max(0, error_indices[0] - 15)
        return "\n".join(lines[start_line:start_line + max_lines])

    def evaluate_failure(self, test_name: str, sanitized_trace: str, git_diff: str) -> TriageAnalysis:
        system_prompt = (
            "You are an automated CI/CD Diagnostic Agent. Analyze test logs and commit diffs.\n"
            "Classify the failure into: INFRASTRUCTURE, FLAKY, CODE_DEFECT, or CONFIGURATION.\n"
            "Output strictly valid JSON conforming to the schema."
        )

        user_prompt = f"""Test Case: {test_name}

=== RECENT GIT DIFF ===
{git_diff[:2500]}

=== SANITIZED TRACEBACK ===
{sanitized_trace}
"""

        payload = {
            "model": "claude-3-5-sonnet-20241022",
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            "response_format": {"type": "json_object"},
            "temperature": 0.1
        }

        response = requests.post(f"{self.base_url}/chat/completions", headers=self.headers, json=payload)
        response.raise_for_status()
        
        result_json = response.json()["choices"][0]["message"]["content"]
        data = json.loads(result_json)
        return TriageAnalysis(**data)

# Execution Example
if __name__ == "__main__":
    # Initialize with n1n.ai unified API gateway
    API_KEY = os.getenv("N1N_API_KEY", "sk-test-key")
    agent = TestTriageAgent(api_key=API_KEY, base_url="https://api.n1n.ai/v1")

    mock_log = """
    [INFO] Starting TestRun: test_hpc_cluster_provisioning
    [DEBUG] Allocating worker nodes AWS c5.4xlarge... OK
    [ERROR] ConnectionResetError: [Errno 104] Connection reset by peer
    Traceback (most recent call last):
      File "/opt/rescale/tests/test_hpc.py", line 142, in test_hpc_cluster_provisioning
        cluster.await_ready(timeout=60)
      File "/opt/rescale/core/cluster.py", line 89, in await_ready
        raise TimeoutError("Cluster orchestration timed out after 60s")
    TimeoutError: Cluster orchestration timed out after 60s
    """

    mock_diff = """
    diff --git a/rescale/core/cluster.py b/rescale/core/cluster.py
    - cluster.await_ready(timeout=180)
    + cluster.await_ready(timeout=60)
    """

    sanitized = agent.sanitize_log(mock_log)
    analysis = agent.evaluate_failure("test_hpc_cluster_provisioning", sanitized, mock_diff)
    print("Triage Verdict:", analysis.json(indent=2))