{a.service}
- {a.confidence}
+ {a.status === 'failed' ? (
+
+ FAILED
+
+ ) : (
+ {a.confidence}
+ )}
{a.error}
@@ -192,7 +207,7 @@ export default function MonitoringPage() {
onClick={e => { e.stopPropagation(); investigate(a); }}
className="hidden group-hover:flex items-center gap-1 text-xs font-medium text-indigo-600 dark:text-indigo-400 bg-indigo-50 dark:bg-indigo-500/10 px-2 py-1 rounded-md hover:bg-indigo-100 dark:hover:bg-indigo-500/20 transition-colors"
>
-
Investigate
+
{a.session_id ? 'View investigation' : 'Investigate'}
diff --git a/frontend/src/pages/SettingsPage.tsx b/frontend/src/pages/SettingsPage.tsx
index 1385c50..c9ec16f 100644
--- a/frontend/src/pages/SettingsPage.tsx
+++ b/frontend/src/pages/SettingsPage.tsx
@@ -77,7 +77,7 @@ export default function SettingsPage() {
setSnsArn(d.sns_topic_arn || '');
setSqsUrl(d.sqs_queue_url || '');
setAwsRegion(d.aws_region || '');
- setInfraEnabled(!!d.initialized);
+ setInfraEnabled(!!d.event_infra_enabled);
setInfraRules(d.eventbridge_rule_arns || {});
})
.catch(() => {});
@@ -88,11 +88,12 @@ export default function SettingsPage() {
const saveAwsConfig = async () => {
setAwsSaving(true);
try {
- await apiFetch('/api/init/setup', {
+ const r = await apiFetch('/api/init/setup', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ sns_topic_arn: snsArn, sqs_queue_url: sqsUrl, aws_region: awsRegion }),
});
+ if (!r.ok) throw new Error('save failed');
toast.success('AWS configuration saved');
} catch {
toast.error('Failed to save configuration');
@@ -118,8 +119,8 @@ export default function SettingsPage() {
setInfraBusy(true);
try {
const r = await apiFetch('/api/init/complete', { method: 'POST' });
- const d = await r.json() as { initialized: boolean; error: string | null };
- if (!d.initialized) { toast.error(d.error || 'Infrastructure setup failed'); return; }
+ const d = await r.json() as { initialized: boolean; event_infra_enabled?: boolean; error?: string | null; detail?: string };
+ if (!r.ok || !d.event_infra_enabled) { toast.error(d.error || d.detail || 'Infrastructure setup failed'); return; }
toast.success('Event monitoring enabled');
// Refresh status
const s = await apiFetch('/api/init/status').then(res => res.json());
@@ -137,13 +138,17 @@ export default function SettingsPage() {
if (!window.confirm('Remove SQS queue and EventBridge rules? This will stop all automatic incident detection.')) return;
setInfraBusy(true);
try {
- await apiFetch('/api/init/infra', { method: 'DELETE' });
+ const r = await apiFetch('/api/init/infra', { method: 'DELETE' });
+ if (!r.ok) {
+ const d = await r.json().catch(() => ({}));
+ throw new Error(d.detail?.errors?.join('\n') || d.detail || 'Teardown failed');
+ }
toast.success('Event monitoring disabled');
setInfraEnabled(false);
setInfraRules({});
setSqsUrl('');
- } catch {
- toast.error('Teardown failed');
+ } catch (err) {
+ toast.error((err as Error).message || 'Teardown failed');
} finally {
setInfraBusy(false);
}
diff --git a/frontend/src/types.ts b/frontend/src/types.ts
index 1714cfe..202ca3e 100644
--- a/frontend/src/types.ts
+++ b/frontend/src/types.ts
@@ -104,8 +104,10 @@ export interface Alert {
error: string;
resolution: string;
confidence: 'HIGH' | 'MEDIUM' | 'LOW';
+ status: 'completed' | 'failed';
sns_sent: boolean;
timestamp: string;
+ session_id?: string;
}
export interface ServiceStatus {
diff --git a/migrations/006_app_config.sql b/migrations/006_app_config.sql
new file mode 100644
index 0000000..8a37637
--- /dev/null
+++ b/migrations/006_app_config.sql
@@ -0,0 +1,6 @@
+-- Stores application-level configuration such as init wizard and AWS event infra state
+CREATE TABLE IF NOT EXISTS app_config (
+ key TEXT PRIMARY KEY,
+ value JSONB NOT NULL DEFAULT '{}'::jsonb,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
diff --git a/migrations/007_alerts_dedup_key.sql b/migrations/007_alerts_dedup_key.sql
new file mode 100644
index 0000000..b9c4d42
--- /dev/null
+++ b/migrations/007_alerts_dedup_key.sql
@@ -0,0 +1,3 @@
+-- Add dedup_key to alerts for cross-path deduplication (poller + event consumer)
+ALTER TABLE alerts ADD COLUMN IF NOT EXISTS dedup_key TEXT;
+CREATE INDEX IF NOT EXISTS idx_alerts_dedup_key_created ON alerts (dedup_key, created_at DESC);
diff --git a/migrations/008_alerts_status.sql b/migrations/008_alerts_status.sql
new file mode 100644
index 0000000..088861c
--- /dev/null
+++ b/migrations/008_alerts_status.sql
@@ -0,0 +1,3 @@
+-- Add status to alerts to distinguish completed vs failed investigations
+ALTER TABLE alerts ADD COLUMN IF NOT EXISTS status TEXT NOT NULL DEFAULT 'completed';
+CREATE INDEX IF NOT EXISTS idx_alerts_status ON alerts (status);
diff --git a/migrations/009_session_source_alert_session.sql b/migrations/009_session_source_alert_session.sql
new file mode 100644
index 0000000..919e187
--- /dev/null
+++ b/migrations/009_session_source_alert_session.sql
@@ -0,0 +1,6 @@
+-- Distinguish event-triggered sessions from user-initiated chats
+ALTER TABLE sessions ADD COLUMN IF NOT EXISTS source TEXT NOT NULL DEFAULT 'chat';
+CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions (source);
+
+-- Link alerts back to the session that produced them
+ALTER TABLE alerts ADD COLUMN IF NOT EXISTS session_id UUID REFERENCES sessions(id) ON DELETE SET NULL;
diff --git a/migrations/010_session_user_interacted.sql b/migrations/010_session_user_interacted.sql
new file mode 100644
index 0000000..e14bab5
--- /dev/null
+++ b/migrations/010_session_user_interacted.sql
@@ -0,0 +1,3 @@
+-- Track whether a user has ever sent a message in an event-triggered session.
+-- FALSE = auto-run only (hidden from sidebar); TRUE = user engaged (show in sidebar).
+ALTER TABLE sessions ADD COLUMN IF NOT EXISTS user_interacted BOOLEAN NOT NULL DEFAULT FALSE;
diff --git a/scripts/test_pipeline.py b/scripts/test_pipeline.py
index eed3fe1..8ef619e 100644
--- a/scripts/test_pipeline.py
+++ b/scripts/test_pipeline.py
@@ -12,7 +12,26 @@
uv run python scripts/test_pipeline.py --region eu-west-1
uv run python scripts/test_pipeline.py --invocations 10
uv run python scripts/test_pipeline.py --list # list available Lambdas
- uv run python scripts/test_pipeline.py --lambda-only # invoke + wait for alarm (no manual SQS push)
+ uv run python scripts/test_pipeline.py --lambda-only # real alarm path (~2-4 min)
+
+─────────────────────────────────────────────────────────────────────────────────
+MANUAL SQS TEST COMMANDS (bypass script — push directly to the queue)
+Replace QUEUE_URL and ACCOUNT_ID with your values from Settings → AWS Configuration.
+─────────────────────────────────────────────────────────────────────────────────
+
+# Syntax error (Runtime.UserCodeSyntaxError) — fast, focused investigation:
+aws sqs send-message --profile devops-agent-readonly --queue-url "https://sqs.us-east-1.amazonaws.com/ACCOUNT_ID/opendevops-agent-events" --message-body "{\"source\":\"aws.lambda\",\"detail-type\":\"Lambda Function Invocation Result - Failure\",\"time\":\"2026-05-17T12:00:00Z\",\"detail\":{\"requestContext\":{\"functionArn\":\"arn:aws:lambda:us-east-1:ACCOUNT_ID:function:opendevops-test-failure\",\"condition\":\"RetriesExhausted\",\"approximateInvokeCount\":3},\"responsePayload\":{\"errorMessage\":\"Syntax error in module lambda_function: unexpected indent (lambda_function.py, line 9)\",\"errorType\":\"Runtime.UserCodeSyntaxError\",\"stackTrace\":[\"File /var/task/lambda_function.py line 9: def lambda_handler(event, context):\"]}}}
+
+# CloudWatch alarm event — slower path (~1 min CloudWatch delay in real scenario):
+aws sqs send-message --profile devops-agent-readonly \
+ --queue-url "https://sqs.us-east-1.amazonaws.com/ACCOUNT_ID/opendevops-agent-events" \
+ --message-body "{\"source\":\"aws.cloudwatch\",\"detail-type\":\"CloudWatch Alarm State Change\",\"time\":\"2026-05-17T12:00:00Z\",\"detail\":{\"alarmName\":\"opendevops-test-lambda-errors\",\"state\":{\"value\":\"ALARM\",\"reason\":\"TEST: Lambda error rate exceeded threshold\"},\"configuration\":{\"metrics\":[{\"metricStat\":{\"metric\":{\"namespace\":\"AWS/Lambda\",\"name\":\"Errors\",\"dimensions\":{\"FunctionName\":\"opendevops-test-failure\"}},\"period\":300,\"stat\":\"Sum\"}}]}}}"
+
+Notes:
+ - Event consumer picks up the message within 20s (SQS long-poll interval)
+ - Use aws.lambda source for faster, more focused investigations
+ - MAX_TOOL_CALLS=30 recommended to avoid recursion errors with weaker models
+─────────────────────────────────────────────────────────────────────────────────
"""
from __future__ import annotations
@@ -26,17 +45,26 @@
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
from dotenv import load_dotenv
+
load_dotenv()
-GREEN = "\033[0;32m"
+GREEN = "\033[0;32m"
YELLOW = "\033[1;33m"
-RED = "\033[0;31m"
-NC = "\033[0m"
+RED = "\033[0;31m"
+NC = "\033[0m"
+
+
+def log(msg: str) -> None:
+ print(f"{GREEN}[+]{NC} {msg}")
+
+
+def warn(msg: str) -> None:
+ print(f"{YELLOW}[!]{NC} {msg}")
-def log(msg: str) -> None: print(f"{GREEN}[+]{NC} {msg}")
-def warn(msg: str) -> None: print(f"{YELLOW}[!]{NC} {msg}")
-def die(msg: str) -> None: print(f"{RED}[✗]{NC} {msg}", file=sys.stderr); sys.exit(1)
+def die(msg: str) -> None:
+ print(f"{RED}[✗]{NC} {msg}", file=sys.stderr)
+ sys.exit(1)
def get_queue_url() -> str:
@@ -61,11 +89,18 @@ def list_functions(lam) -> list[str]:
return sorted(names)
-def run(region: str, invocations: int, function_name: str | None, list_only: bool, lambda_only: bool = False) -> None:
+def run(
+ region: str,
+ invocations: int,
+ function_name: str | None,
+ list_only: bool,
+ lambda_only: bool = False,
+) -> None:
try:
session = boto3.Session()
lam = session.client("lambda", region_name=region)
sqs = session.client("sqs", region_name=region)
+ cw = session.client("cloudwatch", region_name=region)
except NoCredentialsError:
die("No AWS credentials found. Configure via environment variables or ~/.aws/credentials.")
@@ -75,7 +110,10 @@ def run(region: str, invocations: int, function_name: str | None, list_only: boo
identity = sts.get_caller_identity()
log(f"AWS identity: {identity.get('Arn', 'unknown')}")
except ClientError as e:
- die(f"AWS credentials error: {e}\nRun 'aws sso login' or set AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY.")
+ die(
+ f"AWS credentials error: {e}\n"
+ "Run 'aws sso login' or set AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY."
+ )
# ── list mode ─────────────────────────────────────────────────────────────
functions = list_functions(lam)
@@ -103,7 +141,9 @@ def run(region: str, invocations: int, function_name: str | None, list_only: boo
queue_url = get_queue_url()
if not queue_url:
- die("No SQS queue URL found. Run 'Create Infrastructure' in Settings → AWS Configuration first.")
+ die(
+ "No SQS queue URL found. Run 'Create Infrastructure' in Settings first."
+ )
log(f"Region : {region}")
log(f"Queue : {queue_url}")
@@ -114,6 +154,7 @@ def run(region: str, invocations: int, function_name: str | None, list_only: boo
log(f"Invoking '{function_name}' {invocations}× with a bad payload to generate real errors...")
bad_payload = b'{"__test_force_error": true, "trigger": "opendevops-pipeline-test"}'
succeeded = 0
+ errors_observed = 0
for i in range(1, invocations + 1):
try:
resp = lam.invoke(
@@ -125,27 +166,50 @@ def run(region: str, invocations: int, function_name: str | None, list_only: boo
fn_error = resp.get("FunctionError", "")
if fn_error or status >= 400:
print(f" invocation {i}/{invocations} → error (good)")
+ errors_observed += 1
else:
- print(f" invocation {i}/{invocations} → success (function handled bad input gracefully)")
+ print(
+ f" invocation {i}/{invocations} → success "
+ "(function handled bad input gracefully)"
+ )
succeeded += 1
except ClientError as e:
warn(f" invocation {i}/{invocations} → invoke failed: {e}")
if succeeded == 0:
warn("All invocations failed to reach the function — no logs may exist.")
+ elif errors_observed == 0:
+ warn(
+ "The function handled the test payload successfully — Lambda Errors may stay at 0."
+ )
else:
log(f"Real invocation logs at: CloudWatch → /aws/lambda/{function_name}")
print()
# ── lambda-only: skip manual SQS push, let alarm fire naturally ──────────
if lambda_only:
+ if errors_observed == 0:
+ die(
+ "No Lambda function errors were observed; the alarm is unlikely to fire."
+ )
+ try:
+ alarm = cw.describe_alarms(AlarmNames=["opendevops-lambda-errors-aggregate"]).get(
+ "MetricAlarms", []
+ )
+ if alarm and alarm[0].get("StateValue") == "ALARM":
+ warn(
+ "Aggregate alarm is already in ALARM; EventBridge may not emit a transition."
+ )
+ except ClientError as e:
+ warn(f"Could not inspect aggregate alarm state: {e}")
+
sep = "━" * 60
print(f"\n{GREEN}{sep}{NC}")
print(f"{GREEN}✓ Lambda invoked — waiting for alarm to fire automatically{NC}")
print(f" Function : {function_name}")
- print(f" Alarm : opendevops-lambda-errors-aggregate")
+ print(" Alarm : opendevops-lambda-errors-aggregate")
print(f" Real logs : CloudWatch → /aws/lambda/{function_name}")
- print(f" Monitoring: http://localhost/monitoring")
+ print(" Monitoring: http://localhost/monitoring")
print(f"{GREEN}{sep}{NC}\n")
warn("CloudWatch evaluates every 60s — alarm should trip in ~1-2 minutes.")
warn("EventBridge fires → SQS → agent investigates automatically.")
@@ -169,17 +233,19 @@ def run(region: str, invocations: int, function_name: str | None, list_only: boo
),
},
"configuration": {
- "metrics": [{
- "metricStat": {
- "metric": {
- "namespace": "AWS/Lambda",
- "name": "Errors",
- "dimensions": {"FunctionName": function_name},
+ "metrics": [
+ {
+ "metricStat": {
+ "metric": {
+ "namespace": "AWS/Lambda",
+ "name": "Errors",
+ "dimensions": {"FunctionName": function_name},
+ },
+ "period": 60,
+ "stat": "Sum",
},
- "period": 60,
- "stat": "Sum",
- },
- }],
+ }
+ ],
},
},
}
@@ -199,7 +265,7 @@ def run(region: str, invocations: int, function_name: str | None, list_only: boo
print(f" MessageId : {msg_id}")
print(f" Function : {function_name}")
print(f" Real logs : CloudWatch → /aws/lambda/{function_name}")
- print(f" Monitoring: http://localhost/monitoring")
+ print(" Monitoring: http://localhost/monitoring")
print(f"{GREEN}{sep}{NC}\n")
warn("The agent will pull REAL CloudWatch logs from that function.")
warn("Investigation takes ~30–90s. Watch the Monitoring page.")
@@ -209,12 +275,20 @@ def main() -> None:
parser = argparse.ArgumentParser(
description="Trigger a real Lambda failure through the monitoring pipeline"
)
- parser.add_argument("--function", help="Lambda function name to use (auto-picks first if omitted)")
- parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-east-1"))
+ parser.add_argument(
+ "--function", help="Lambda function name to use (auto-picks first if omitted)"
+ )
+ parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-east-1"))
parser.add_argument("--invocations", type=int, default=5)
- parser.add_argument("--list", action="store_true", help="List available Lambda functions and exit")
- parser.add_argument("--lambda-only", action="store_true", dest="lambda_only",
- help="Invoke Lambda and let the CloudWatch alarm fire automatically (no manual SQS push)")
+ parser.add_argument(
+ "--list", action="store_true", help="List available Lambda functions and exit"
+ )
+ parser.add_argument(
+ "--lambda-only",
+ action="store_true",
+ dest="lambda_only",
+ help="Invoke Lambda and let the CloudWatch alarm fire automatically",
+ )
args = parser.parse_args()
run(args.region, args.invocations, args.function, args.list, args.lambda_only)
diff --git a/src/agent/context_collectors.py b/src/agent/context_collectors.py
index 0106e9c..01b7099 100644
--- a/src/agent/context_collectors.py
+++ b/src/agent/context_collectors.py
@@ -7,11 +7,16 @@
import boto3
from loguru import logger
+from agent.init_store import get_runtime_aws_region
from config import settings
def _session() -> boto3.Session:
- return boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session()
+ return (
+ boto3.Session(profile_name=settings.aws_profile)
+ if settings.aws_profile
+ else boto3.Session()
+ )
def collect_context(event: dict) -> dict[str, Any]:
@@ -38,7 +43,12 @@ def collect_context(event: dict) -> dict[str, Any]:
elif source == "aws.guardduty":
return {"type": "guardduty", "detail": detail}
else:
- return {"type": "unknown", "source": source, "detail_type": detail_type, "detail": detail}
+ return {
+ "type": "unknown",
+ "source": source,
+ "detail_type": detail_type,
+ "detail": detail,
+ }
except Exception as e:
logger.error("Context collection failed for {}/{}: {}", source, detail_type, e)
return {"type": source, "error": str(e), "detail": detail}
@@ -47,8 +57,9 @@ def collect_context(event: dict) -> dict[str, Any]:
def _collect_alarm(detail: dict) -> dict:
"""Collect context for a CloudWatch alarm state change."""
import datetime
+
s = _session()
- region = settings.aws_region
+ region = get_runtime_aws_region()
cw = s.client("cloudwatch", region_name=region)
alarm_name = detail.get("alarmName", "")
@@ -67,19 +78,22 @@ def _collect_alarm(detail: dict) -> dict:
start = end - datetime.timedelta(hours=1)
try:
metric_resp = cw.get_metric_data(
- MetricDataQueries=[{
- "Id": "m1",
- "MetricStat": {
- "Metric": {
- "Namespace": alarm.get("Namespace", ""),
- "MetricName": alarm.get("MetricName", ""),
- "Dimensions": alarm.get("Dimensions", []),
+ MetricDataQueries=[
+ {
+ "Id": "m1",
+ "MetricStat": {
+ "Metric": {
+ "Namespace": alarm.get("Namespace", ""),
+ "MetricName": alarm.get("MetricName", ""),
+ "Dimensions": alarm.get("Dimensions", []),
+ },
+ "Period": 60,
+ "Stat": alarm.get("Statistic", "Average"),
},
- "Period": 60,
- "Stat": alarm.get("Statistic", "Average"),
- },
- }],
- StartTime=start, EndTime=end,
+ }
+ ],
+ StartTime=start,
+ EndTime=end,
)
values = metric_resp.get("MetricDataResults", [{}])[0].get("Values", [])
ctx["recent_values"] = values[:10]
@@ -92,13 +106,12 @@ def _collect_alarm(detail: dict) -> dict:
def _collect_lambda(detail: dict) -> dict:
"""Collect context for a Lambda failure."""
s = _session()
- region = settings.aws_region
+ region = get_runtime_aws_region()
lam = s.client("lambda", region_name=region)
logs = s.client("logs", region_name=region)
- fn_name = (
- detail.get("requestContext", {}).get("functionArn", "").split(":")[-1]
- or detail.get("functionName", "")
+ fn_name = detail.get("requestContext", {}).get("functionArn", "").split(":")[-1] or detail.get(
+ "functionName", ""
)
ctx: dict[str, Any] = {"type": "lambda", "function_name": fn_name, "detail": detail}
@@ -131,7 +144,7 @@ def _collect_lambda(detail: dict) -> dict:
def _collect_ecs(detail: dict) -> dict:
"""Collect context for an ECS task stopped event."""
s = _session()
- region = settings.aws_region
+ region = get_runtime_aws_region()
ecs = s.client("ecs", region_name=region)
logs = s.client("logs", region_name=region)
@@ -148,11 +161,13 @@ def _collect_ecs(detail: dict) -> dict:
}
for container in detail.get("containers", []):
- ctx["containers"].append({
- "name": container.get("name", ""),
- "exit_code": container.get("exitCode"),
- "reason": container.get("reason", ""),
- })
+ ctx["containers"].append(
+ {
+ "name": container.get("name", ""),
+ "exit_code": container.get("exitCode"),
+ "reason": container.get("reason", ""),
+ }
+ )
task_def_arn = detail.get("taskDefinitionArn", "")
if task_def_arn:
@@ -168,7 +183,10 @@ def _collect_ecs(detail: dict) -> dict:
task_id = task_arn.split("/")[-1]
stream_name = f"{prefix}/{containers[0]['name']}/{task_id}"
log_events = logs.get_log_events(
- logGroupName=log_group, logStreamName=stream_name, limit=30, startFromHead=False
+ logGroupName=log_group,
+ logStreamName=stream_name,
+ limit=30,
+ startFromHead=False,
)
ctx["recent_logs"] = [e["message"] for e in log_events.get("events", [])]
except Exception:
@@ -180,7 +198,7 @@ def _collect_ecs(detail: dict) -> dict:
def _collect_ec2(detail: dict) -> dict:
"""Collect context for an EC2 state change."""
s = _session()
- region = settings.aws_region
+ region = get_runtime_aws_region()
ec2 = s.client("ec2", region_name=region)
instance_id = detail.get("instance-id", "")
@@ -211,7 +229,7 @@ def _collect_ec2(detail: dict) -> dict:
def _collect_rds(detail: dict) -> dict:
"""Collect context for an RDS event."""
s = _session()
- region = settings.aws_region
+ region = get_runtime_aws_region()
rds = s.client("rds", region_name=region)
source_id = detail.get("SourceIdentifier", "")
diff --git a/src/agent/db/base.py b/src/agent/db/base.py
index e781ef3..40620ef 100644
--- a/src/agent/db/base.py
+++ b/src/agent/db/base.py
@@ -31,6 +31,7 @@ async def upsert_session(
model: str,
aws_region: str,
title: str | None = None,
+ source: str = "chat",
) -> None: ...
@abstractmethod
@@ -102,8 +103,15 @@ async def get_user_by_email(self, email: str) -> dict | None:
async def get_user_by_id(self, user_id: str) -> dict | None:
return None
+ async def create_org(self, name: str, slug: str) -> dict | None:
+ return None
+
+ async def get_first_org(self) -> dict | None:
+ return None
+
async def create_user(
- self, email: str, name: str, password_hash: str, role: str
+ self, email: str, name: str, password_hash: str, role: str,
+ org_id: str | None = None,
) -> dict | None:
return None
@@ -113,6 +121,9 @@ async def list_users(self) -> list[dict]:
async def update_user(self, user_id: str, **fields: Any) -> dict | None:
return None
+ async def assign_org_to_users_without_org(self, org_id: str) -> None:
+ pass
+
async def delete_user(self, user_id: str) -> None:
pass
@@ -126,11 +137,25 @@ async def add_alert(
resolution: str,
confidence: str,
sns_sent: bool,
+ dedup_key: str | None = None,
+ status: str = "completed",
+ session_id: str | None = None,
) -> str:
return ""
+ async def is_recent_alert(self, dedup_key: str, within_minutes: int = 3) -> bool:
+ return False
+
async def get_alerts(self, limit: int = 50) -> list[dict]:
return []
async def get_alert(self, alert_id: str) -> dict | None:
return None
+
+ # ── App config (init wizard / infrastructure state) ──────────────────────
+
+ async def get_app_config(self, key: str) -> dict | None:
+ return None
+
+ async def set_app_config(self, key: str, value: dict) -> None:
+ pass
diff --git a/src/agent/db/memory.py b/src/agent/db/memory.py
index ebefe0a..ffd8b1f 100644
--- a/src/agent/db/memory.py
+++ b/src/agent/db/memory.py
@@ -4,7 +4,7 @@
import uuid
from collections import defaultdict
-from datetime import datetime, timezone
+from datetime import UTC, datetime
from typing import Any
from loguru import logger
@@ -22,6 +22,7 @@ def __init__(self) -> None:
self._tool_calls: dict[str, list[dict]] = defaultdict(list)
self._usage: dict[str, list[dict]] = defaultdict(list)
self._alerts: list[dict] = []
+ self._app_config: dict[str, dict] = {}
async def init(self) -> Any:
from langgraph.checkpoint.memory import MemorySaver
@@ -42,7 +43,7 @@ def checkpointer(self) -> Any:
@staticmethod
def _now() -> str:
- return datetime.now(timezone.utc).isoformat()
+ return datetime.now(UTC).isoformat()
# ── App helpers ───────────────────────────────────────────────────────────
@@ -52,16 +53,21 @@ async def upsert_session(
model: str,
aws_region: str,
title: str | None = None,
+ source: str = "chat",
) -> None:
if session_id in self._sessions:
self._sessions[session_id]["last_active_at"] = self._now()
self._sessions[session_id]["model"] = model
+ if source == "chat":
+ self._sessions[session_id]["user_interacted"] = True
else:
self._sessions[session_id] = {
"id": session_id,
"title": title,
"model": model,
"aws_region": aws_region,
+ "source": source,
+ "user_interacted": source == "chat",
"created_at": self._now(),
"last_active_at": self._now(),
"is_deleted": False,
@@ -132,7 +138,11 @@ async def save_usage_event(
})
async def list_sessions(self, limit: int = 15, offset: int = 0) -> list[dict]:
- active = [s for s in self._sessions.values() if not s.get("is_deleted")]
+ active = [
+ s for s in self._sessions.values()
+ if not s.get("is_deleted")
+ and (s.get("source", "chat") == "chat" or s.get("user_interacted", False))
+ ]
sorted_sessions = sorted(active, key=lambda s: s["last_active_at"], reverse=True)
return sorted_sessions[offset: offset + limit]
@@ -240,6 +250,9 @@ async def add_alert(
resolution: str,
confidence: str,
sns_sent: bool,
+ dedup_key: str | None = None,
+ status: str = "completed",
+ session_id: str | None = None,
) -> str:
alert_id = str(uuid.uuid4())
self._alerts.append({
@@ -250,6 +263,9 @@ async def add_alert(
"confidence": confidence,
"sns_sent": sns_sent,
"timestamp": self._now(),
+ "dedup_key": dedup_key,
+ "status": status,
+ "session_id": session_id,
})
return alert_id
@@ -262,6 +278,12 @@ async def get_alert(self, alert_id: str) -> dict | None:
return a
return None
+ async def get_app_config(self, key: str) -> dict | None:
+ return self._app_config.get(key)
+
+ async def set_app_config(self, key: str, value: dict) -> None:
+ self._app_config[key] = value
+
async def search_sessions(self, query: str, limit: int = 10) -> list[dict]:
if not query.strip():
return []
diff --git a/src/agent/db/postgres.py b/src/agent/db/postgres.py
index 4896791..bc0c8ca 100644
--- a/src/agent/db/postgres.py
+++ b/src/agent/db/postgres.py
@@ -27,8 +27,8 @@ async def init(self) -> Any:
try:
import psycopg # type: ignore
- from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
+ from psycopg_pool import AsyncConnectionPool
self._pool = AsyncConnectionPool(
conninfo=settings.database_url,
@@ -123,16 +123,21 @@ async def upsert_session(
model: str,
aws_region: str,
title: str | None = None,
+ source: str = "chat",
) -> None:
await self._exec(
"""
- INSERT INTO sessions (id, title, model, aws_region)
- VALUES (%s, %s, %s, %s)
+ INSERT INTO sessions (id, title, model, aws_region, source)
+ VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (id) DO UPDATE SET
last_active_at = NOW(),
- model = EXCLUDED.model
+ model = EXCLUDED.model,
+ user_interacted = CASE
+ WHEN EXCLUDED.source = 'chat' THEN TRUE
+ ELSE sessions.user_interacted
+ END
""",
- uuid.UUID(session_id), title, model, aws_region,
+ uuid.UUID(session_id), title, model, aws_region, source,
)
async def save_message(
@@ -198,7 +203,9 @@ async def save_usage_event(
async def list_sessions(self, limit: int = 15, offset: int = 0) -> list[dict]:
rows = await self._fetchall(
- "SELECT id, title, last_active_at, model, aws_region FROM sessions WHERE is_deleted = FALSE ORDER BY last_active_at DESC LIMIT %s OFFSET %s",
+ "SELECT id, title, last_active_at, model, aws_region FROM sessions"
+ " WHERE is_deleted = FALSE AND (source = 'chat' OR user_interacted = TRUE)"
+ " ORDER BY last_active_at DESC LIMIT %s OFFSET %s",
limit, offset,
)
return [
@@ -474,21 +481,54 @@ async def get_user_by_email(self, email: str) -> dict | None:
async def get_user_by_id(self, user_id: str) -> dict | None:
return await self._fetchrow("SELECT * FROM users WHERE id = %s", uuid.UUID(user_id))
- async def create_user(
- self, email: str, name: str, password_hash: str, role: str
- ) -> dict | None:
+ async def create_org(self, name: str, slug: str) -> dict | None:
row = await self._fetchrow(
"""
- INSERT INTO users (email, name, password_hash, role)
- VALUES (%s, %s, %s, %s)
- RETURNING id, email, name, role, created_at
+ INSERT INTO organizations (name, slug)
+ VALUES (%s, %s)
+ ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name
+ RETURNING id, name, slug
""",
- email, name, password_hash, role,
+ name, slug,
+ )
+ if row:
+ row["id"] = str(row["id"])
+ return row
+
+ async def get_first_org(self) -> dict | None:
+ row = await self._fetchrow(
+ "SELECT id, name, slug FROM organizations ORDER BY created_at ASC LIMIT 1"
)
if row:
row["id"] = str(row["id"])
return row
+ async def create_user(
+ self, email: str, name: str, password_hash: str, role: str,
+ org_id: str | None = None,
+ ) -> dict | None:
+ if org_id:
+ row = await self._fetchrow(
+ """
+ INSERT INTO users (email, name, password_hash, role, org_id)
+ VALUES (%s, %s, %s, %s, %s)
+ RETURNING id, email, name, role, created_at
+ """,
+ email, name, password_hash, role, uuid.UUID(org_id),
+ )
+ else:
+ row = await self._fetchrow(
+ """
+ INSERT INTO users (email, name, password_hash, role)
+ VALUES (%s, %s, %s, %s)
+ RETURNING id, email, name, role, created_at
+ """,
+ email, name, password_hash, role,
+ )
+ if row:
+ row["id"] = str(row["id"])
+ return row
+
async def list_users(self) -> list[dict]:
rows = await self._fetchall(
"SELECT id, email, name, role, created_at FROM users ORDER BY created_at ASC"
@@ -516,6 +556,12 @@ async def update_user(self, user_id: str, **fields: Any) -> dict | None:
row["id"] = str(row["id"])
return row
+ async def assign_org_to_users_without_org(self, org_id: str) -> None:
+ await self._exec(
+ "UPDATE users SET org_id = %s WHERE org_id IS NULL",
+ uuid.UUID(org_id),
+ )
+
async def delete_user(self, user_id: str) -> None:
await self._exec("DELETE FROM users WHERE id = %s", uuid.UUID(user_id))
@@ -528,17 +574,30 @@ async def add_alert(
resolution: str,
confidence: str,
sns_sent: bool,
+ dedup_key: str | None = None,
+ status: str = "completed",
+ session_id: str | None = None,
) -> str:
row = await self._fetchrow(
- "INSERT INTO alerts (service, error, resolution, confidence, sns_sent) "
- "VALUES (%s, %s, %s, %s, %s) RETURNING id",
- service, error, resolution, confidence, sns_sent,
+ "INSERT INTO alerts"
+ " (service, error, resolution, confidence, sns_sent, dedup_key, status, session_id)"
+ " VALUES (%s, %s, %s, %s, %s, %s, %s, %s) RETURNING id",
+ service, error, resolution, confidence, sns_sent, dedup_key, status,
+ uuid.UUID(session_id) if session_id else None,
)
return str(row["id"]) if row else ""
+ async def is_recent_alert(self, dedup_key: str, within_minutes: int = 3) -> bool:
+ row = await self._fetchrow(
+ "SELECT 1 FROM alerts WHERE dedup_key = %s"
+ " AND created_at > NOW() - (%s * INTERVAL '1 minute') LIMIT 1",
+ dedup_key, within_minutes,
+ )
+ return row is not None
+
async def get_alerts(self, limit: int = 50) -> list[dict]:
rows = await self._fetchall(
- "SELECT id, service, error, resolution, confidence, sns_sent, created_at "
+ "SELECT id, service, error, resolution, confidence, sns_sent, status, created_at, session_id "
"FROM alerts ORDER BY created_at DESC LIMIT %s",
min(limit, 200),
)
@@ -550,14 +609,16 @@ async def get_alerts(self, limit: int = 50) -> list[dict]:
"resolution": r["resolution"],
"confidence": r["confidence"],
"sns_sent": bool(r["sns_sent"]),
+ "status": r["status"],
"timestamp": r["created_at"].isoformat() if r["created_at"] else None,
+ "session_id": str(r["session_id"]) if r["session_id"] else None,
}
for r in rows
]
async def get_alert(self, alert_id: str) -> dict | None:
row = await self._fetchrow(
- "SELECT id, service, error, resolution, confidence, sns_sent, created_at "
+ "SELECT id, service, error, resolution, confidence, sns_sent, status, created_at, session_id "
"FROM alerts WHERE id = %s",
uuid.UUID(alert_id),
)
@@ -570,9 +631,23 @@ async def get_alert(self, alert_id: str) -> dict | None:
"resolution": row["resolution"],
"confidence": row["confidence"],
"sns_sent": bool(row["sns_sent"]),
+ "status": row["status"],
"timestamp": row["created_at"].isoformat() if row["created_at"] else None,
+ "session_id": str(row["session_id"]) if row["session_id"] else None,
}
+ async def get_app_config(self, key: str) -> dict | None:
+ row = await self._fetchrow("SELECT value FROM app_config WHERE key = %s", key)
+ return row["value"] if row else None
+
+ async def set_app_config(self, key: str, value: dict) -> None:
+ await self._exec(
+ "INSERT INTO app_config (key, value) VALUES (%s, %s) "
+ "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()",
+ key,
+ self._jsonb(value),
+ )
+
async def search_sessions(self, query: str, limit: int = 10) -> list[dict]:
if not query.strip():
return []
diff --git a/src/agent/db/sqlite.py b/src/agent/db/sqlite.py
index 76c8f30..e30ad76 100644
--- a/src/agent/db/sqlite.py
+++ b/src/agent/db/sqlite.py
@@ -12,17 +12,18 @@
from agent.db.base import DatabaseBackend
from config import settings
-
_DDL = """
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
title TEXT,
model TEXT NOT NULL DEFAULT '',
aws_region TEXT NOT NULL DEFAULT 'us-east-1',
- created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')),
- last_active_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')),
- is_deleted INTEGER NOT NULL DEFAULT 0,
- deleted_at TEXT
+ source TEXT NOT NULL DEFAULT 'chat',
+ user_interacted INTEGER NOT NULL DEFAULT 0,
+ created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')),
+ last_active_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')),
+ is_deleted INTEGER NOT NULL DEFAULT 0,
+ deleted_at TEXT
);
CREATE TABLE IF NOT EXISTS messages (
@@ -71,6 +72,12 @@
);
CREATE INDEX IF NOT EXISTS alerts_created_at_idx ON alerts(created_at DESC);
CREATE INDEX IF NOT EXISTS alerts_service_idx ON alerts(service);
+
+CREATE TABLE IF NOT EXISTS app_config (
+ key TEXT PRIMARY KEY,
+ value TEXT NOT NULL DEFAULT '{}',
+ updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now'))
+);
"""
_SERVICE_MAP: dict[str, str] = {
@@ -126,6 +133,17 @@ async def init(self) -> Any:
except Exception:
pass # column already exists
+ # Migrate existing sessions tables: add source / user_interacted columns if absent
+ for col_sql in (
+ "ALTER TABLE sessions ADD COLUMN source TEXT NOT NULL DEFAULT 'chat'",
+ "ALTER TABLE sessions ADD COLUMN user_interacted INTEGER NOT NULL DEFAULT 0",
+ ):
+ try:
+ await self._conn.execute(col_sql)
+ await self._conn.commit()
+ except Exception:
+ pass # column already exists
+
# Migrate existing databases: create alerts table if absent
try:
await self._conn.execute(
@@ -133,12 +151,36 @@ async def init(self) -> Any:
"id TEXT PRIMARY KEY, service TEXT NOT NULL, error TEXT NOT NULL, "
"resolution TEXT NOT NULL DEFAULT '', confidence TEXT NOT NULL DEFAULT 'LOW', "
"sns_sent INTEGER NOT NULL DEFAULT 0, "
+ "dedup_key TEXT, status TEXT NOT NULL DEFAULT 'completed', "
+ "session_id TEXT, "
"created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')))"
)
await self._conn.commit()
except Exception:
pass
+ # Migrate existing alerts tables: add dedup_key / status / session_id if absent
+ for col_sql in (
+ "ALTER TABLE alerts ADD COLUMN dedup_key TEXT",
+ "ALTER TABLE alerts ADD COLUMN status TEXT NOT NULL DEFAULT 'completed'",
+ "ALTER TABLE alerts ADD COLUMN session_id TEXT",
+ ):
+ try:
+ await self._conn.execute(col_sql)
+ await self._conn.commit()
+ except Exception:
+ pass # column already exists
+
+ try:
+ await self._conn.execute(
+ "CREATE TABLE IF NOT EXISTS app_config ("
+ "key TEXT PRIMARY KEY, value TEXT NOT NULL DEFAULT '{}', "
+ "updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')))"
+ )
+ await self._conn.commit()
+ except Exception:
+ pass
+
# LangGraph checkpointer uses its own connection to the same file
cp_conn = await aiosqlite.connect(self._path)
self._checkpointer = AsyncSqliteSaver(cp_conn)
@@ -203,16 +245,21 @@ async def upsert_session(
model: str,
aws_region: str,
title: str | None = None,
+ source: str = "chat",
) -> None:
await self._exec(
"""
- INSERT INTO sessions (id, title, model, aws_region)
- VALUES (?, ?, ?, ?)
+ INSERT INTO sessions (id, title, model, aws_region, source)
+ VALUES (?, ?, ?, ?, ?)
ON CONFLICT (id) DO UPDATE SET
last_active_at = strftime('%Y-%m-%dT%H:%M:%S', 'now'),
- model = excluded.model
+ model = excluded.model,
+ user_interacted = CASE
+ WHEN excluded.source = 'chat' THEN 1
+ ELSE sessions.user_interacted
+ END
""",
- session_id, title, model, aws_region,
+ session_id, title, model, aws_region, source,
)
async def save_message(
@@ -275,7 +322,9 @@ async def save_usage_event(
async def list_sessions(self, limit: int = 15, offset: int = 0) -> list[dict]:
rows = await self._fetchall(
- "SELECT id, title, last_active_at, model, aws_region FROM sessions WHERE is_deleted = 0 ORDER BY last_active_at DESC LIMIT ? OFFSET ?",
+ "SELECT id, title, last_active_at, model, aws_region FROM sessions"
+ " WHERE is_deleted = 0 AND (source = 'chat' OR user_interacted = 1)"
+ " ORDER BY last_active_at DESC LIMIT ? OFFSET ?",
limit, offset,
)
return [
@@ -608,11 +657,17 @@ async def add_alert(
resolution: str,
confidence: str,
sns_sent: bool,
+ dedup_key: str | None = None,
+ status: str = "completed",
+ session_id: str | None = None,
) -> str:
alert_id = str(uuid.uuid4())
await self._exec(
- "INSERT INTO alerts (id, service, error, resolution, confidence, sns_sent) VALUES (?, ?, ?, ?, ?, ?)",
- alert_id, service, error, resolution, confidence, 1 if sns_sent else 0,
+ "INSERT INTO alerts"
+ " (id, service, error, resolution, confidence, sns_sent, dedup_key, status, session_id)"
+ " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ alert_id, service, error, resolution, confidence,
+ 1 if sns_sent else 0, dedup_key, status, session_id,
)
return alert_id
@@ -653,6 +708,25 @@ async def get_alert(self, alert_id: str) -> dict | None:
"timestamp": row["created_at"],
}
+ async def get_app_config(self, key: str) -> dict | None:
+ row = await self._fetchone("SELECT value FROM app_config WHERE key = ?", key)
+ if not row:
+ return None
+ try:
+ return json.loads(row["value"])
+ except Exception:
+ return None
+
+ async def set_app_config(self, key: str, value: dict) -> None:
+ await self._exec(
+ "INSERT INTO app_config (key, value, updated_at) "
+ "VALUES (?, ?, strftime('%Y-%m-%dT%H:%M:%S', 'now')) "
+ "ON CONFLICT(key) DO UPDATE SET "
+ "value = excluded.value, updated_at = excluded.updated_at",
+ key,
+ json.dumps(value),
+ )
+
async def search_sessions(self, query: str, limit: int = 10) -> list[dict]:
if not query.strip():
return []
diff --git a/src/agent/event_consumer.py b/src/agent/event_consumer.py
index fc6894a..a00c243 100644
--- a/src/agent/event_consumer.py
+++ b/src/agent/event_consumer.py
@@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
+import hashlib
import json
import uuid
from typing import Any
@@ -10,9 +11,18 @@
import boto3
from loguru import logger
+from agent.init_store import (
+ get_runtime_aws_region,
+ get_runtime_sns_topic_arn,
+ get_runtime_sqs_queue_url,
+)
+from agent.monitor_store import add_alert, is_recent_alert, update_service
from config import settings
-from agent.init_store import load_init
-from agent.monitor_store import add_alert
+
+# In-memory set of dedup keys currently being investigated.
+# Prevents a second identical event from starting a parallel agent run
+# before the first one finishes and writes its key to the DB.
+_in_progress: set[str] = set()
def _build_investigation_prompt(event: dict) -> str:
@@ -25,21 +35,31 @@ def _build_investigation_prompt(event: dict) -> str:
if source == "aws.cloudwatch" and "alarmName" in detail:
alarm_name = detail["alarmName"]
reason = detail.get("state", {}).get("reason", "")
- metric = detail.get("configuration", {}).get("metrics", [{}])[0].get("metricStat", {}).get("metric", {})
+ metric = (
+ detail.get("configuration", {})
+ .get("metrics", [{}])[0]
+ .get("metricStat", {})
+ .get("metric", {})
+ )
namespace = metric.get("namespace", "")
metric_name = metric.get("name", "")
dimensions = metric.get("dimensions", {})
if not dimensions:
dim_hint = (
- "IMPORTANT: This alarm has NO dimensions — it monitors the AGGREGATE metric across all resources. "
+ "IMPORTANT: This alarm has NO dimensions — it monitors the AGGREGATE "
+ "metric across all resources. "
"You MUST first identify which specific resource is causing the errors. "
- "For Lambda errors: use get_metric_data with a SEARCH expression like "
- "SEARCH('{AWS/Lambda,FunctionName} MetricName=\"Errors\"', 'Sum', 300) to find which function(s) have errors. "
+ "For Lambda errors: call list_lambda_functions, then use "
+ "get_lambda_error_rate on likely functions "
+ "to find which function(s) have errors. "
"Then pull that function's CloudWatch logs to find the actual error message."
)
else:
- dim_hint = f"Dimensions: {json.dumps(dimensions)} — use these to identify the specific resource."
+ dim_hint = (
+ f"Dimensions: {json.dumps(dimensions)} — use these to identify the "
+ "specific resource."
+ )
return (
f'CloudWatch alarm "{alarm_name}" triggered at {time}.\n'
@@ -48,7 +68,7 @@ def _build_investigation_prompt(event: dict) -> str:
f"{dim_hint}\n"
f"Reason: {reason}\n\n"
"Investigate step by step:\n"
- "1. Identify the SPECIFIC resource (function name, instance ID, etc.) causing the issue\n"
+ "1. Identify the SPECIFIC resource causing the issue\n"
"2. Pull its recent CloudWatch logs\n"
"3. Find the actual error message or stack trace\n"
"4. Provide actionable fix steps based on the real error"
@@ -86,56 +106,190 @@ def _build_investigation_prompt(event: dict) -> str:
)
-async def _run_investigation(prompt: str) -> dict[str, Any] | None:
- """Run a full agent investigation and return the submit_investigation result."""
+async def _run_investigation(prompt: str, session_id: str) -> dict[str, Any] | None:
+ """Run a full agent investigation using the same streaming path as the chat endpoint."""
from agent.core import get_agent
+ from agent.turns import save_turn
+
+ def _f(obj: Any, key: str, default: Any = None) -> Any:
+ return obj.get(key, default) if isinstance(obj, dict) else getattr(obj, key, default)
- thread_id = str(uuid.uuid4())
config = {
- "configurable": {"thread_id": thread_id},
+ "configurable": {"thread_id": session_id},
"recursion_limit": settings.max_tool_calls * 3 + 15,
}
+ tc_accum: dict[int, dict[str, Any]] = {}
+ pending_calls: dict[str, dict[str, Any]] = {}
+ tool_calls_log: list[dict[str, Any]] = []
+ investigation_result: dict[str, Any] | None = None
+ response_text = ""
+ usage_meta: Any = None
+
try:
- result = await get_agent().ainvoke(
+ async for chunk, _meta in get_agent().astream(
{"messages": [{"role": "user", "content": prompt}]},
config=config,
- )
- for msg in reversed(result.get("messages", [])):
- for tc in getattr(msg, "tool_calls", []) or []:
- name = tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
- args = tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", {})
- if name == "submit_investigation":
- return args
- messages = result.get("messages", [])
- if messages:
- last = messages[-1]
- content = last.content if hasattr(last, "content") else str(last)
- if content:
- return {"root_cause_summary": content, "confidence": "MEDIUM", "mitigation_steps": []}
+ stream_mode="messages",
+ ):
+ um = getattr(chunk, "usage_metadata", None)
+ if um:
+ usage_meta = um
+
+ for tcc in getattr(chunk, "tool_call_chunks", []) or []:
+ idx = _f(tcc, "index", 0)
+ if idx not in tc_accum:
+ tc_accum[idx] = {"id": "", "name": "", "args_str": ""}
+ if tc_id := _f(tcc, "id"):
+ tc_accum[idx]["id"] = tc_id
+ if name := (_f(tcc, "name") or ""):
+ tc_accum[idx]["name"] += name
+ if args := (_f(tcc, "args") or ""):
+ tc_accum[idx]["args_str"] += args
+
+ for tc in getattr(chunk, "tool_calls", []) or []:
+ tc_id = _f(tc, "id") or ""
+ name = _f(tc, "name") or ""
+ args = _f(tc, "args") or {}
+ if tc_id and name:
+ pending_calls[tc_id] = {"tool": name, "args": args if isinstance(args, dict) else {}}
+
+ content = getattr(chunk, "content", "")
+ tc_id = getattr(chunk, "tool_call_id", None)
+
+ if content and isinstance(content, str) and not tc_id:
+ response_text += content
+
+ if tc_id:
+ for entry in tc_accum.values():
+ if eid := entry["id"]:
+ try:
+ eargs: Any = json.loads(entry["args_str"]) if entry["args_str"] else {}
+ except json.JSONDecodeError:
+ eargs = {}
+ pending_calls[eid] = {"tool": entry["name"], "args": eargs}
+ tc_accum.clear()
+
+ call_info = pending_calls.pop(
+ tc_id,
+ {"tool": getattr(chunk, "name", None) or "unknown", "args": {}},
+ )
+ try:
+ result = json.loads(content)
+ except (json.JSONDecodeError, TypeError):
+ result = {"raw": str(content)[:500]}
+
+ tool_calls_log.append({"tool": call_info["tool"], "args": call_info["args"], "result": result})
+ if call_info["tool"] == "submit_investigation":
+ investigation_result = call_info["args"]
+
except Exception as e:
logger.error("Agent investigation failed: {}", e)
- return None
+ investigation_result = {
+ "_status": "failed",
+ "root_cause_summary": f"Investigation failed: {e}",
+ "confidence": "LOW",
+ "mitigation_steps": [
+ "Re-investigate manually via the chat — use a higher MAX_TOOL_CALLS if needed.",
+ ],
+ }
+
+ # Build readable assistant summary
+ if investigation_result and investigation_result.get("root_cause_summary") and "_status" not in investigation_result:
+ steps = investigation_result.get("mitigation_steps", [])
+ steps_text = "\n".join(f"- {s}" for s in steps) if steps else "_No steps provided._"
+ confidence = investigation_result.get("confidence", "MEDIUM")
+ assistant_text = (
+ f"**Root Cause ({confidence} confidence):** "
+ f"{investigation_result['root_cause_summary']}\n\n"
+ f"**Mitigation Steps:**\n{steps_text}"
+ )
+ elif response_text.strip():
+ assistant_text = response_text.strip()
+ else:
+ assistant_text = "Investigation completed. See tool calls for details."
+
+ await save_turn(
+ session_id=session_id,
+ user_message=prompt,
+ assistant_text=assistant_text,
+ tool_calls_log=tool_calls_log,
+ usage={
+ "model": settings.llm_model,
+ "input_tokens": _f(usage_meta, "input_tokens", 0) or 0,
+ "output_tokens": _f(usage_meta, "output_tokens", 0) or 0,
+ "latency_ms": 0,
+ },
+ source="event",
+ )
+
+ return investigation_result
+
+_VOLATILE_KEYS = {
+ "time", "timestamp", "eventTime", "requestId", "eventID",
+ "approximateInvokeCount", "startTime", "endTime", "updatedAt",
+ "stateTransitionedAt", "lastUpdatedTime",
+}
-async def _deliver(result: dict[str, Any], event: dict) -> None:
+
+def _strip_volatile(obj: Any) -> Any:
+ if isinstance(obj, dict):
+ return {k: _strip_volatile(v) for k, v in obj.items() if k not in _VOLATILE_KEYS}
+ if isinstance(obj, list):
+ return [_strip_volatile(i) for i in obj]
+ return obj
+
+
+def _event_dedup_key(event: dict) -> str:
+ """MD5 fingerprint of the event's stable fields — works for any AWS service."""
+ source = event.get("source", "unknown")
+ fingerprint = {
+ "source": source,
+ "detail-type": event.get("detail-type", ""),
+ "detail": _strip_volatile(event.get("detail", {})),
+ }
+ digest = hashlib.md5(
+ json.dumps(fingerprint, sort_keys=True, default=str).encode()
+ ).hexdigest()[:12]
+ return f"{source}:{digest}"
+
+
+def _extract_aws_error(event: dict) -> str:
+ """Pull the original AWS error message out of the event payload."""
+ source = event.get("source", "")
+ detail = event.get("detail", {})
+ if source == "aws.lambda":
+ return detail.get("responsePayload", {}).get("errorMessage", "")
+ if source == "aws.cloudwatch":
+ return detail.get("state", {}).get("reason", "")
+ if source == "aws.ecs":
+ return detail.get("stoppedReason", "")
+ if source == "aws.rds":
+ return detail.get("Message", "")
+ return ""
+
+
+async def _deliver(result: dict[str, Any], event: dict, dedup_key: str, session_id: str) -> None:
"""Send investigation result to SNS + Slack and persist to the monitor store."""
+ status = result.pop("_status", "completed")
+ is_test = bool(event.get("_opendevops_test", False))
services_affected = result.get("services_affected", [])
- service = result.get("service_name", result.get("service",
- services_affected[0] if services_affected else "unknown"
- ))
+ fallback = services_affected[0] if services_affected else event.get("source", "unknown")
+ service = result.get("service_name", result.get("service", fallback))
root_cause = result.get("root_cause_summary", "")
mitigation = result.get("mitigation_steps", [])
confidence = result.get("confidence", "MEDIUM")
resolution = "\n".join(mitigation) if isinstance(mitigation, list) else str(mitigation)
+ aws_error = _extract_aws_error(event)
- init_data = load_init()
- sns_arn = settings.sns_topic_arn or init_data.get("sns_topic_arn")
+ sns_arn = get_runtime_sns_topic_arn()
sns_sent = False
if sns_arn:
try:
from tools.sns import publish_sns_alert
+
services_affected = result.get("services_affected", [])
evidence = result.get("evidence", [])
message = (
@@ -143,12 +297,18 @@ async def _deliver(result: dict[str, Any], event: dict) -> None:
f"Root Cause: {root_cause}\n"
f"Evidence: {chr(10).join(evidence[:5]) if evidence else 'N/A'}\n"
f"Mitigation Steps:\n{resolution}\n"
- f"Services Affected: {', '.join(services_affected) if services_affected else service}\n"
+ "Services Affected: "
+ f"{', '.join(services_affected) if services_affected else service}\n"
f"Confidence: {confidence}\n"
f"Event Time: {event.get('time', '')}"
)
+ subject_prefix = "[FAILED]" if status == "failed" else f"[{confidence}]"
await asyncio.get_event_loop().run_in_executor(
- None, publish_sns_alert, sns_arn, f"[{confidence}] {service} — {root_cause[:80]}", message
+ None,
+ publish_sns_alert,
+ sns_arn,
+ f"{subject_prefix} {service} — {root_cause[:80]}",
+ message,
)
sns_sent = True
except Exception as e:
@@ -157,13 +317,31 @@ async def _deliver(result: dict[str, Any], event: dict) -> None:
# Also notify Slack (keeps parity with the poller)
if settings.slack_webhook_url:
try:
- from integrations.slack_webhook import post_investigation
- session_id = str(uuid.uuid4())
- await post_investigation(settings.slack_webhook_url, result, session_id)
+ from integrations.slack_webhook import post_failed_investigation, post_investigation
+
+ if status == "failed":
+ await post_failed_investigation(
+ settings.slack_webhook_url, service, root_cause,
+ session_id, aws_error=aws_error, is_test=is_test,
+ )
+ else:
+ await post_investigation(
+ settings.slack_webhook_url, result, session_id, is_test=is_test
+ )
except Exception as e:
logger.error("Slack delivery failed from event consumer: {}", e)
- add_alert(service=service, error=root_cause, resolution=resolution, confidence=confidence, sns_sent=sns_sent)
+ update_service(service, status, root_cause)
+ await add_alert(
+ service=service,
+ error=root_cause,
+ resolution=resolution,
+ confidence=confidence,
+ sns_sent=sns_sent,
+ dedup_key=dedup_key,
+ status=status,
+ session_id=session_id,
+ )
logger.info("Alert delivered: service={} confidence={} sns={}", service, confidence, sns_sent)
@@ -186,7 +364,7 @@ def _is_real_failure(source: str, detail: dict) -> bool:
async def _process_event(event: dict) -> None:
- """Filter → enrich prompt → investigate → deliver."""
+ """Filter → dedup check → enrich prompt → investigate → deliver."""
source = event.get("source", "unknown")
detail_type = event.get("detail-type", "")
detail = event.get("detail", {})
@@ -195,6 +373,17 @@ async def _process_event(event: dict) -> None:
logger.debug("Skipping non-failure event: {} / {}", source, detail_type)
return
+ dedup_key = _event_dedup_key(event)
+ if dedup_key in _in_progress or await is_recent_alert(dedup_key):
+ logger.info(
+ "Dedup: skipping {} / {} (key={}) — {}",
+ source, detail_type, dedup_key,
+ "investigation already in progress" if dedup_key in _in_progress
+ else "same event investigated within last 3 min",
+ )
+ return
+
+ _in_progress.add(dedup_key)
logger.info("Processing event: {} / {}", source, detail_type)
prompt = _build_investigation_prompt(event)
@@ -202,31 +391,42 @@ async def _process_event(event: dict) -> None:
# Enrich prompt with deterministic boto3 context (no LLM cost)
try:
from agent.context_collectors import collect_context
+
context = collect_context(event)
if context and not context.get("error"):
context_str = json.dumps(context, default=str)
- prompt = prompt + f"\n\nPre-collected context (use as starting facts, verify with tools):\n{context_str[:3000]}"
+ prompt = (
+ prompt
+ + "\n\nPre-collected context (use as starting facts, verify with tools):\n"
+ + context_str[:3000]
+ )
except Exception as e:
logger.debug("Context collection skipped: {}", e)
- result = await _run_investigation(prompt)
- if result:
- await _deliver(result, event)
- else:
- logger.warning("Investigation produced no result for {} / {}", source, detail_type)
+ session_id = str(uuid.uuid4())
+ try:
+ result = await _run_investigation(prompt, session_id)
+ if not result:
+ logger.warning("Investigation produced no result for {} / {}", source, detail_type)
+ return
+ await _deliver(result, event, dedup_key, session_id)
+ finally:
+ _in_progress.discard(dedup_key)
async def event_consumer_loop() -> None:
"""Main loop — long-polls SQS, processes events."""
- init_data = load_init()
- # Prefer env var over init.json so users can configure via .env without the wizard
- queue_url = settings.sqs_queue_url or init_data.get("sqs_queue_url")
+ queue_url = get_runtime_sqs_queue_url()
if not queue_url:
logger.warning("No SQS queue URL configured — event consumer not starting")
return
- s = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session()
- sqs = s.client("sqs", region_name=settings.aws_region)
+ s = (
+ boto3.Session(profile_name=settings.aws_profile)
+ if settings.aws_profile
+ else boto3.Session()
+ )
+ sqs = s.client("sqs", region_name=get_runtime_aws_region())
logger.info("Event consumer started — polling {}", queue_url)
diff --git a/src/agent/event_infra.py b/src/agent/event_infra.py
index 7d41185..f956f37 100644
--- a/src/agent/event_infra.py
+++ b/src/agent/event_infra.py
@@ -5,6 +5,7 @@
import json
import boto3
+from botocore.exceptions import ClientError
from loguru import logger
from config import settings
@@ -24,7 +25,10 @@
},
"opendevops-lambda-throttle": {
"source": ["aws.lambda"],
- "detail-type": ["Lambda Function Invocation Result - Failure", "AWS API Call via CloudTrail"],
+ "detail-type": [
+ "Lambda Function Invocation Result - Failure",
+ "AWS API Call via CloudTrail",
+ ],
"detail": {"errorCode": ["TooManyRequestsException", "Throttling"]},
},
"opendevops-ecs-task-stopped": {
@@ -32,7 +36,11 @@
"detail-type": ["ECS Task State Change"],
"detail": {
"lastStatus": ["STOPPED"],
- "stopCode": ["TaskFailedToStart", "EssentialContainerExited", "ServiceSchedulerInitiated"],
+ "stopCode": [
+ "TaskFailedToStart",
+ "EssentialContainerExited",
+ "ServiceSchedulerInitiated",
+ ],
"containers": {"exitCode": [{"anything-but": 0}]},
},
},
@@ -65,108 +73,166 @@
def _session() -> boto3.Session:
- return boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session()
+ return (
+ boto3.Session(profile_name=settings.aws_profile)
+ if settings.aws_profile
+ else boto3.Session()
+ )
ALARM_NAME = "opendevops-lambda-errors-aggregate"
-def setup_event_infra() -> dict:
+def _missing_resource(exc: Exception) -> bool:
+ if not isinstance(exc, ClientError):
+ return False
+ code = exc.response.get("Error", {}).get("Code", "")
+ return code in {
+ "ResourceNotFoundException",
+ "AWS.SimpleQueueService.NonExistentQueue",
+ "QueueDoesNotExist",
+ }
+
+
+def setup_event_infra(region: str | None = None) -> dict:
"""Create SQS queue, EventBridge rules, and aggregate Lambda alarm.
Returns {queue_url, queue_arn, rule_arns}.
"""
s = _session()
- region = settings.aws_region
+ region = region or settings.aws_region
sqs = s.client("sqs", region_name=region)
events = s.client("events", region_name=region)
cw = s.client("cloudwatch", region_name=region)
sts = s.client("sts", region_name=region)
account_id = sts.get_caller_identity()["Account"]
-
- resp = sqs.create_queue(
- QueueName=QUEUE_NAME,
- Attributes={
- "MessageRetentionPeriod": "86400",
- "VisibilityTimeout": "120",
- "ReceiveMessageWaitTimeSeconds": "20",
- },
- )
- queue_url = resp["QueueUrl"]
- queue_attrs = sqs.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["QueueArn"])
- queue_arn = queue_attrs["Attributes"]["QueueArn"]
-
- policy = {
- "Version": "2012-10-17",
- "Statement": [{
- "Sid": "AllowEventBridge",
- "Effect": "Allow",
- "Principal": {"Service": "events.amazonaws.com"},
- "Action": "sqs:SendMessage",
- "Resource": queue_arn,
- "Condition": {"ArnLike": {"aws:SourceArn": f"arn:aws:events:{region}:{account_id}:rule/opendevops-*"}},
- }],
- }
- sqs.set_queue_attributes(QueueUrl=queue_url, Attributes={"Policy": json.dumps(policy)})
-
+ queue_url = ""
rule_arns: dict[str, str] = {}
- for rule_name, pattern in RULES.items():
- resp = events.put_rule(
- Name=rule_name,
- EventPattern=json.dumps(pattern),
- State="ENABLED",
- Description=f"OpenDevOps Agent — {rule_name}",
+
+ try:
+ resp = sqs.create_queue(
+ QueueName=QUEUE_NAME,
+ Attributes={
+ "MessageRetentionPeriod": "86400",
+ "VisibilityTimeout": "120",
+ "ReceiveMessageWaitTimeSeconds": "20",
+ },
)
- rule_arns[rule_name] = resp["RuleArn"]
- events.put_targets(
- Rule=rule_name,
- Targets=[{"Id": "opendevops-sqs", "Arn": queue_arn}],
+ queue_url = resp["QueueUrl"]
+ queue_attrs = sqs.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["QueueArn"])
+ queue_arn = queue_attrs["Attributes"]["QueueArn"]
+
+ policy = {
+ "Version": "2012-10-17",
+ "Statement": [
+ {
+ "Sid": "AllowEventBridge",
+ "Effect": "Allow",
+ "Principal": {"Service": "events.amazonaws.com"},
+ "Action": "sqs:SendMessage",
+ "Resource": queue_arn,
+ "Condition": {
+ "ArnLike": {
+ "aws:SourceArn": (
+ f"arn:aws:events:{region}:{account_id}:rule/opendevops-*"
+ )
+ }
+ },
+ }
+ ],
+ }
+ sqs.set_queue_attributes(QueueUrl=queue_url, Attributes={"Policy": json.dumps(policy)})
+
+ for rule_name, pattern in RULES.items():
+ resp = events.put_rule(
+ Name=rule_name,
+ EventPattern=json.dumps(pattern),
+ State="ENABLED",
+ Description=f"OpenDevOps Agent — {rule_name}",
+ )
+ rule_arns[rule_name] = resp["RuleArn"]
+ events.put_targets(
+ Rule=rule_name,
+ Targets=[{"Id": "opendevops-sqs", "Arn": queue_arn}],
+ )
+
+ # Aggregate CloudWatch alarm — fires when ANY Lambda in the account errors.
+ # No dimensions = account-wide. EventBridge alarm-state rule routes it to SQS.
+ cw.put_metric_alarm(
+ AlarmName=ALARM_NAME,
+ AlarmDescription="OpenDevOps Agent — triggers investigation on any Lambda error",
+ Namespace="AWS/Lambda",
+ MetricName="Errors",
+ Statistic="Sum",
+ Period=60,
+ EvaluationPeriods=1,
+ Threshold=1,
+ ComparisonOperator="GreaterThanOrEqualToThreshold",
+ TreatMissingData="notBreaching",
)
+ except Exception:
+ if queue_url or rule_arns:
+ teardown_event_infra(queue_url, rule_arns, region)
+ raise
- # Aggregate CloudWatch alarm — fires when ANY Lambda in the account errors.
- # No dimensions = account-wide. EventBridge alarm-state rule routes it to SQS.
- cw.put_metric_alarm(
- AlarmName=ALARM_NAME,
- AlarmDescription="OpenDevOps Agent — triggers investigation on any Lambda error",
- Namespace="AWS/Lambda",
- MetricName="Errors",
- Statistic="Sum",
- Period=60,
- EvaluationPeriods=1,
- Threshold=1,
- ComparisonOperator="GreaterThanOrEqualToThreshold",
- TreatMissingData="notBreaching",
- )
logger.info(
"Event infra created: queue={} rules={} alarm={}", queue_url, len(rule_arns), ALARM_NAME
)
return {"queue_url": queue_url, "queue_arn": queue_arn, "rule_arns": rule_arns}
-def teardown_event_infra(queue_url: str, rule_arns: dict[str, str]) -> None:
+def teardown_event_infra(
+ queue_url: str | None,
+ rule_arns: dict[str, str] | None,
+ region: str | None = None,
+) -> dict:
"""Remove EventBridge rules, aggregate CloudWatch alarm, and SQS queue."""
s = _session()
- region = settings.aws_region
+ region = region or settings.aws_region
sqs = s.client("sqs", region_name=region)
events = s.client("events", region_name=region)
cw = s.client("cloudwatch", region_name=region)
+ warnings: list[str] = []
+ errors: list[str] = []
- for rule_name in rule_arns:
+ rules_to_delete = list((rule_arns or {}).keys()) or list(RULES.keys())
+ for rule_name in rules_to_delete:
try:
- events.remove_targets(Rule=rule_name, Ids=["opendevops-sqs"])
+ resp = events.remove_targets(Rule=rule_name, Ids=["opendevops-sqs"])
+ if resp.get("FailedEntryCount"):
+ errors.append(f"Failed to remove target for rule {rule_name}: {resp}")
events.delete_rule(Name=rule_name)
except Exception as e:
- logger.warning("Failed to delete rule {}: {}", rule_name, e)
+ message = f"Failed to delete rule {rule_name}: {e}"
+ logger.warning(message)
+ if _missing_resource(e):
+ warnings.append(message)
+ else:
+ errors.append(message)
try:
cw.delete_alarms(AlarmNames=[ALARM_NAME])
except Exception as e:
- logger.warning("Failed to delete alarm {}: {}", ALARM_NAME, e)
-
- try:
- sqs.delete_queue(QueueUrl=queue_url)
- except Exception as e:
- logger.warning("Failed to delete queue: {}", e)
+ message = f"Failed to delete alarm {ALARM_NAME}: {e}"
+ logger.warning(message)
+ if _missing_resource(e):
+ warnings.append(message)
+ else:
+ errors.append(message)
+
+ if queue_url:
+ try:
+ sqs.delete_queue(QueueUrl=queue_url)
+ except Exception as e:
+ message = f"Failed to delete queue: {e}"
+ logger.warning(message)
+ if _missing_resource(e):
+ warnings.append(message)
+ else:
+ errors.append(message)
+ else:
+ warnings.append("No queue URL was configured")
logger.info("Event infra torn down")
+ return {"warnings": warnings, "errors": errors}
diff --git a/src/agent/init_store.py b/src/agent/init_store.py
index 2d01b2b..d837163 100644
--- a/src/agent/init_store.py
+++ b/src/agent/init_store.py
@@ -1,52 +1,166 @@
-"""Init config store — persists initialization state to