From cd3085ff8b02343694a7678d54d1be1c43ac9789 Mon Sep 17 00:00:00 2001 From: AhmadHammad21 Date: Thu, 14 May 2026 20:47:52 +0300 Subject: [PATCH 01/16] fix: redirect to /init after first registration and skip account step if already logged in --- frontend/src/App.tsx | 29 ++++++++++++++++++++++------- frontend/src/pages/InitPage.tsx | 5 ++++- 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index d5c650f..6cf394e 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -29,13 +29,28 @@ function LogoutPage() { } function RedirectToSession() { - const stored = localStorage.getItem('devops-session-id'); - const id = stored ?? (() => { - const newId = crypto.randomUUID(); - localStorage.setItem('devops-session-id', newId); - return newId; - })(); - return ; + const [ready, setReady] = useState<'init' | 'chat' | null>(null); + const { user } = useAuth(); + + useEffect(() => { + if (user?.role !== 'admin') { setReady('chat'); return; } + fetch('/api/init/status') + .then(r => r.json()) + .then(d => setReady(d.initialized ? 'chat' : 'init')) + .catch(() => setReady('chat')); + }, []); // eslint-disable-line react-hooks/exhaustive-deps + + if (ready === 'init') return ; + if (ready === 'chat') { + const stored = localStorage.getItem('devops-session-id'); + const id = stored ?? (() => { + const newId = crypto.randomUUID(); + localStorage.setItem('devops-session-id', newId); + return newId; + })(); + return ; + } + return null; } const PAGE_SIZE = 15; diff --git a/frontend/src/pages/InitPage.tsx b/frontend/src/pages/InitPage.tsx index 15a25ba..2f7d78a 100644 --- a/frontend/src/pages/InitPage.tsx +++ b/frontend/src/pages/InitPage.tsx @@ -2,6 +2,7 @@ import { useState, useEffect } from 'react'; import { useNavigate } from 'react-router-dom'; import { ArrowRight, Loader2, Terminal, Eye, EyeOff, CheckCircle, XCircle, AlertTriangle, ChevronRight, Check } from 'lucide-react'; import { cn } from '../lib/utils'; +import { useAuth } from '../context/AuthContext'; const inp = 'w-full text-base text-gray-900 dark:text-white bg-gray-50 dark:bg-[#18181B] border border-gray-200 dark:border-[#27272A] rounded-xl px-4 py-3.5 outline-none focus:border-indigo-500 dark:focus:border-indigo-400 focus:ring-2 focus:ring-indigo-500/10 transition-all placeholder:text-gray-400 dark:placeholder:text-[#52525B]'; @@ -43,8 +44,10 @@ type Step = 1 | 2 | 3 | 4; export default function InitPage() { const navigate = useNavigate(); + const { user } = useAuth(); - const [step, setStep] = useState(1); + // Skip account creation if already logged in (registered via Login page) + const [step, setStep] = useState(user ? 2 : 1); // Step 1 const [username, setUsername] = useState('admin'); From ab543869b7acbaf2f40c717647b988dfa37c835d Mon Sep 17 00:00:00 2001 From: AhmadHammad21 Date: Thu, 14 May 2026 20:49:25 +0300 Subject: [PATCH 02/16] fix: redirect to /init when no users exist even if init.json says initialized --- frontend/src/App.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 6cf394e..160b07f 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -36,7 +36,7 @@ function RedirectToSession() { if (user?.role !== 'admin') { setReady('chat'); return; } fetch('/api/init/status') .then(r => r.json()) - .then(d => setReady(d.initialized ? 'chat' : 'init')) + .then(d => setReady((d.initialized && d.has_user) ? 'chat' : 'init')) .catch(() => setReady('chat')); }, []); // eslint-disable-line react-hooks/exhaustive-deps From 53727d8688e7f392a82227dfb5e9d04640367472 Mon Sep 17 00:00:00 2001 From: AhmadHammad21 Date: Thu, 14 May 2026 20:58:57 +0300 Subject: [PATCH 03/16] =?UTF-8?q?fix:=20stop=20/init=20redirect=20loop=20?= =?UTF-8?q?=E2=80=94=20only=20redirect=20away=20when=20both=20has=5Fuser?= =?UTF-8?q?=20and=20initialized?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend/src/pages/InitPage.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frontend/src/pages/InitPage.tsx b/frontend/src/pages/InitPage.tsx index 2f7d78a..1109cb7 100644 --- a/frontend/src/pages/InitPage.tsx +++ b/frontend/src/pages/InitPage.tsx @@ -72,7 +72,7 @@ export default function InitPage() { useEffect(() => { fetch('/api/init/status') .then(r => r.json()) - .then(d => { if (d.has_user) navigate('/', { replace: true }); }) + .then(d => { if (d.has_user && d.initialized) navigate('/', { replace: true }); }) .catch(() => {}); }, [navigate]); From aa395b3ef1ef77ddd20ece2d4e3aa07d67242111 Mon Sep 17 00:00:00 2001 From: AhmadHammad21 Date: Sat, 16 May 2026 23:27:49 +0300 Subject: [PATCH 04/16] feat: init wizard robustness + organization onboarding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix redirect loop: RedirectToSession now waits for authLoading before routing, eliminating the /init ↔ / race condition - Add require_admin auth guard to all protected init endpoints - Add org_name field to wizard Step 1; creates organizations record and links admin user via org_id on first registration - Subsequent users registered via /auth/register inherit the existing org - init_store: add in-memory cache, async DB persistence via app_config table, legacy field migration, and helper accessors - migrations/006_app_config.sql: new app_config table for init state - stop_event_consumer() added; called on teardown and re-provision - teardown_event_infra returns warnings/errors, handles managed vs external queues, and rolls back on partial failure - Better HTTP status codes in init router (409 conflicts, 500 errors) - login() called after Step 1 account creation so Steps 2-4 have auth --- docs/databases.md | 6 + docs/event_detection.md | 27 ++-- docs/monitoring.md | 6 +- docs/schema.md | 24 +++ docs/ui.md | 6 +- frontend/src/App.tsx | 34 +++- frontend/src/pages/InitPage.tsx | 78 ++++++--- frontend/src/pages/MonitoringPage.tsx | 16 +- frontend/src/pages/SettingsPage.tsx | 19 ++- migrations/006_app_config.sql | 6 + scripts/test_pipeline.py | 113 +++++++++---- src/agent/context_collectors.py | 74 +++++---- src/agent/db/base.py | 17 +- src/agent/db/memory.py | 7 + src/agent/db/postgres.py | 59 ++++++- src/agent/db/sqlite.py | 35 ++++ src/agent/event_consumer.py | 85 +++++++--- src/agent/event_infra.py | 198 +++++++++++++++-------- src/agent/init_store.py | 160 +++++++++++++++---- src/agent/monitor_store.py | 23 +-- src/agent/permission_checker.py | 4 +- src/agent/poller.py | 25 +-- src/agent/turns.py | 36 +++-- src/api/app.py | 46 ++++-- src/api/routers/auth.py | 4 +- src/api/routers/init.py | 220 +++++++++++++++++++------- src/api/routers/settings.py | 2 + src/tools/_cache.py | 3 +- src/tools/cloudtrail.py | 21 ++- src/tools/cloudwatch.py | 34 ++-- src/tools/ec2.py | 13 +- src/tools/ecs.py | 19 ++- src/tools/iam.py | 19 ++- src/tools/lambda_.py | 28 +++- src/tools/rds.py | 11 +- src/tools/sns.py | 9 +- 36 files changed, 1088 insertions(+), 399 deletions(-) create mode 100644 migrations/006_app_config.sql diff --git a/docs/databases.md b/docs/databases.md index 6828118..0bc539e 100644 --- a/docs/databases.md +++ b/docs/databases.md @@ -44,6 +44,10 @@ Uses `aiosqlite` for the app tables and `langgraph-checkpoint-sqlite` for the LangGraph checkpointer. Both share the same `.db` file via separate connections with WAL mode enabled. +SQLite also stores app-level configuration such as init wizard completion and event +monitoring infrastructure state in `app_config`. `data/init.json` is only a local +cache/fallback. + The file and its parent directory are created automatically on first start. **When to use:** Single-server deployments, personal use, hobbyist setups where @@ -81,6 +85,8 @@ DATABASE_URL=postgresql://user:password@localhost:5432/opendevops Uses `psycopg3` + `AsyncConnectionPool` for the app tables and `langgraph-checkpoint-postgres` for the LangGraph checkpointer. The checkpointer schema is created automatically via `AsyncPostgresSaver.setup()`. +Init wizard and event-infrastructure state are stored in the `app_config` table so +all app instances share the same setup status. **When to use:** Production deployments, team environments, when you need full dashboard analytics, multi-instance horizontal scaling. diff --git a/docs/event_detection.md b/docs/event_detection.md index 817557b..b6e82bf 100644 --- a/docs/event_detection.md +++ b/docs/event_detection.md @@ -30,7 +30,7 @@ AWS Service (ECS, Lambda, RDS, EC2…) ### Option A — Init wizard (recommended) -Go to **Settings → AWS Configuration** after first login. Enter your SNS Topic ARN and AWS Region, save, then click **Run checks** to verify IAM permissions. The SQS queue and EventBridge rules are created automatically the first time the server starts with valid config. +Go to **Settings → AWS Configuration** after first login. Enter your AWS Region and optional SNS Topic ARN, save, then click **Run checks** to verify IAM permissions. Click **Create Infrastructure** to create the SQS queue, EventBridge rules, and aggregate Lambda CloudWatch alarm. ### Option B — Manual `.env` @@ -40,7 +40,7 @@ EVENT_CONSUMER_ENABLED=true # or leave unset — consumer also starts if SNS_TOPIC_ARN=arn:aws:sns:us-east-1:123456789012:opendevops-alerts ``` -The consumer starts automatically on server startup if either `EVENT_CONSUMER_ENABLED=true` or `SQS_QUEUE_URL` is set. +The consumer starts automatically on server startup if either `EVENT_CONSUMER_ENABLED=true`, `SQS_QUEUE_URL` is set, or the init wizard has enabled event infrastructure in persistent app config. --- @@ -50,15 +50,15 @@ The consumer starts automatically on server startup if either `EVENT_CONSUMER_EN | Rule | Events captured | |---|---| -| `opendevops-cloudwatch-alarms` | CloudWatch Alarm state changes | -| `opendevops-ecs-failures` | ECS task stopped with non-zero exit / OOM | -| `opendevops-lambda-errors` | Lambda async invocation failures | -| `opendevops-rds-events` | RDS instance failures and failovers | +| `opendevops-alarm-state` | CloudWatch Alarm state changes to `ALARM` | +| `opendevops-lambda-failure` | Lambda async invocation failures | +| `opendevops-lambda-throttle` | Lambda throttling errors | +| `opendevops-ecs-task-stopped` | ECS task stopped with non-zero exit / OOM | | `opendevops-ec2-state` | EC2 instance state changes | -| `opendevops-asg-events` | Auto Scaling launch / terminate failures | -| `opendevops-codepipeline` | CodePipeline stage failures | -| `opendevops-ssm-compliance` | SSM patch compliance failures | -| `opendevops-health-events` | AWS Health personal health events | +| `opendevops-rds-events` | RDS instance failures and failovers | +| `opendevops-health` | AWS Health personal health events | +| `opendevops-codedeploy-failure` | CodeDeploy deployment failures | +| `opendevops-guardduty` | GuardDuty findings | --- @@ -82,7 +82,7 @@ Context is appended to the investigation prompt, capped at 3 000 chars to stay w `event_consumer_loop()` in `src/agent/event_consumer.py`: -1. Receives up to 10 messages per poll with a 20-second long-poll wait +1. Receives up to 5 messages per poll with a 20-second long-poll wait 2. Filters noise: only processes events where `_is_real_failure()` returns true (e.g. skips EC2 `running` state changes, healthy RDS events) 3. Calls `collect_context(event)` to enrich the prompt 4. Runs a full agent investigation via `agent.ainvoke` @@ -120,9 +120,10 @@ The IAM role/user used by the agent needs these additional permissions for event ``` sns:Publish, sns:GetTopicAttributes sqs:CreateQueue, sqs:SetQueueAttributes, sqs:GetQueueAttributes -sqs:ReceiveMessage, sqs:DeleteMessage, sqs:ListQueues +sqs:ReceiveMessage, sqs:DeleteMessage, sqs:DeleteQueue, sqs:ListQueues events:PutRule, events:PutTargets, events:ListRules events:RemoveTargets, events:DeleteRule +cloudwatch:PutMetricAlarm, cloudwatch:DeleteAlarms, cloudwatch:DescribeAlarms ``` Use **Settings → AWS Configuration → Run checks** to validate all permissions before going live. @@ -151,7 +152,7 @@ For SQLite, the `alerts` table is created automatically on first start. | `src/agent/event_infra.py` | Create/teardown SQS queue + EventBridge rules | | `src/agent/context_collectors.py` | Per-event-type boto3 context enrichment | | `src/agent/monitor_store.py` | In-memory service status + DB-backed alert persistence | -| `src/agent/init_store.py` | Persist init config to `data/init.json` | +| `src/agent/init_store.py` | Persist init config to DB-backed `app_config` with `data/init.json` fallback/cache | | `src/agent/permission_checker.py` | IAM permission validation | | `src/tools/sns.py` | SNS publish wrapper | | `src/api/routers/init.py` | Setup wizard API endpoints | diff --git a/docs/monitoring.md b/docs/monitoring.md index 63e3563..42f42b3 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -13,7 +13,7 @@ OpenDevOps Agent has two complementary detection modes. Both write to the same ` | **How it works** | Agent polls CloudWatch alarms + Lambda error rates on a timer | AWS fires EventBridge rules → SQS → consumer → agent | | **Setup** | Set `POLL_INTERVAL_MINUTES` in `.env` | Create infrastructure via Settings → AWS Configuration | | **Detection latency** | Up to N minutes (your poll interval) | 2–5 min for alarm-based events; near-instant for direct events (ECS, RDS, GuardDuty) | -| **CloudWatch alarm required** | No — reads Lambda error rate directly via API | Yes — EventBridge only fires when an alarm changes state | +| **CloudWatch alarm required** | No — reads Lambda error rate directly via API | Only for metric/alarm detections; direct AWS service events do not need a CloudWatch alarm | | **Works without AWS setup** | Yes — no SQS/EventBridge needed | No — requires SQS queue + EventBridge rules | | **Event types covered** | CloudWatch alarms in ALARM state + Lambda error rates | CloudWatch alarms, ECS task failures, Lambda async errors, RDS events, EC2 state changes, CodePipeline failures, GuardDuty findings, AWS Health events | | **Appears in /monitoring** | Yes | Yes | @@ -131,7 +131,7 @@ When you create infrastructure (Settings → AWS Configuration → Create Infras - Trips when any Lambda has ≥ 1 error in a 60-second window - Triggers the `opendevops-alarm-state` EventBridge rule → SQS → consumer -This means you get event-driven Lambda coverage without creating per-function alarms. +This means you get event-driven Lambda coverage without creating per-function alarms. Because it is account-wide and trips on a single error, it is intentionally easy to test but can be noisy in busy accounts. For production, consider replacing it with scoped per-function or per-service alarms once you know which workloads matter. ### Setup @@ -140,6 +140,8 @@ Go to **Settings → AWS Configuration → Create Infrastructure**. This creates - 9 EventBridge rules (all prefixed `opendevops-`) - Aggregate CloudWatch alarm: `opendevops-lambda-errors-aggregate` +Wizard and infrastructure state is stored in the database for SQLite/PostgreSQL deployments (`app_config`) and mirrored to `data/init.json` as a local cache/fallback. Memory mode keeps only local process/file state. Teardown deletes only infrastructure created by the wizard; queues supplied through `.env` or pasted manually are disconnected from the app but not deleted. + To remove it, click **Teardown**. See [event_detection.md](event_detection.md) for full setup details and IAM permission requirements. ### Context collectors diff --git a/docs/schema.md b/docs/schema.md index 2b75a31..482e8dc 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -15,6 +15,8 @@ psql $DATABASE_URL -f migrations/001_initial.sql psql $DATABASE_URL -f migrations/002_soft_delete.sql psql $DATABASE_URL -f migrations/003_usage_events_metadata.sql psql $DATABASE_URL -f migrations/004_users_rbac.sql +psql $DATABASE_URL -f migrations/005_alerts.sql +psql $DATABASE_URL -f migrations/006_app_config.sql ``` All migrations are idempotent (`IF NOT EXISTS`, `IF column does not already exist`). @@ -134,6 +136,28 @@ Structured root-cause analysis rows extracted from agent final answers. Table ex ### `api_keys` *(future — Phase 3)* Hashed API keys for programmatic access. Table exists but not yet implemented. +### `alerts` +Persisted event-driven and proactive polling investigation results shown on `/monitoring`. + +| Column | Type | Notes | +|---|---|---| +| `id` | UUID PK | `gen_random_uuid()` | +| `service` | TEXT | Affected service/resource label | +| `error` | TEXT | Root cause summary | +| `resolution` | TEXT | Mitigation steps joined as text | +| `confidence` | TEXT | `HIGH`, `MEDIUM`, or `LOW` | +| `sns_sent` | BOOLEAN | Whether SNS publish succeeded | +| `created_at` | TIMESTAMPTZ | | + +### `app_config` +Application-level key/value configuration shared by all server instances. The init wizard stores its setup and event-infrastructure state under key `init`. `data/init.json` is kept as a local cache/fallback for memory mode and bootstrap recovery. + +| Column | Type | Notes | +|---|---|---| +| `key` | TEXT PK | e.g. `init` | +| `value` | JSONB | Setup state, AWS region, SQS URL, rule ARNs | +| `updated_at` | TIMESTAMPTZ | | + --- ## Entity Relationship diff --git a/docs/ui.md b/docs/ui.md index 0e36db2..66df26e 100644 --- a/docs/ui.md +++ b/docs/ui.md @@ -90,7 +90,7 @@ Accessible from the sidebar for all authenticated users. Each card shows the aff Shows active configuration. Four tabs (admins see all four; regular users see the last three): -**AWS Configuration** *(admin only, default tab for admins)* — editable fields for SNS Topic ARN, SQS Queue URL, and AWS Region. Values are saved to `data/init.json` server-side and shared across all admin users. Includes: +**AWS Configuration** *(admin only, default tab for admins)* — editable fields for SNS Topic ARN, SQS Queue URL, and AWS Region. Values are stored in database-backed app config for SQLite/PostgreSQL deployments and mirrored to `data/init.json` as a local cache/fallback. Includes: - **Event Monitoring** section — shows Enabled / Disabled status; **Create Infrastructure** button provisions the SQS queue and 9 EventBridge rules; **Teardown** button removes them (with confirmation). - **Run checks** button that validates IAM permissions for every AWS service the agent uses. @@ -109,9 +109,9 @@ Accessible at `/init` before any users exist. A 4-step wizard guides the admin t 1. **Create account** — email + password for the admin user 2. **AWS Configuration** — region selector and optional SNS Topic ARN for alert delivery 3. **Permission check** — verifies IAM permissions per service; required services (CloudWatch, Lambda, SQS, EventBridge) are highlighted; "Continue anyway" is always available -4. **Enable event monitoring** — creates the SQS queue and 9 EventBridge rules; shows a success summary with queue URL and rule list; includes a **Skip** option with a disclaimer listing what won't work without it +4. **Enable event monitoring** — creates the SQS queue, 9 EventBridge rules, and aggregate Lambda alarm; shows a success summary with queue URL and rule list; includes a **Skip** option that finishes onboarding without event infrastructure -The page redirects away automatically once an admin account exists. +The page redirects away automatically once onboarding is complete. Event monitoring can still be enabled or torn down later from Settings without forcing users back through onboarding. --- diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 160b07f..076c2c8 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -11,7 +11,7 @@ import MonitoringPage from './pages/MonitoringPage'; import AlertDetailPage from './pages/AlertDetailPage'; import InitPage from './pages/InitPage'; import ProtectedRoute from './components/ProtectedRoute'; -import { fetchSessions, deleteSession as apiDeleteSession, renameSession as apiRenameSession } from './lib/api'; +import { apiFetch, fetchSessions, deleteSession as apiDeleteSession, renameSession as apiRenameSession } from './lib/api'; import { useAuth } from './context/AuthContext'; import type { Session } from './types'; @@ -30,15 +30,33 @@ function LogoutPage() { function RedirectToSession() { const [ready, setReady] = useState<'init' | 'chat' | null>(null); - const { user } = useAuth(); + const { user, authRequired, loading } = useAuth(); useEffect(() => { - if (user?.role !== 'admin') { setReady('chat'); return; } - fetch('/api/init/status') - .then(r => r.json()) - .then(d => setReady((d.initialized && d.has_user) ? 'chat' : 'init')) - .catch(() => setReady('chat')); - }, []); // eslint-disable-line react-hooks/exhaustive-deps + if (loading) return; + + let cancelled = false; + const routeAfterInitCheck = async () => { + if (authRequired && user?.role !== 'admin') { + if (!cancelled) setReady('chat'); + return; + } + + try { + const r = await apiFetch('/api/init/status'); + const d = await r.json(); + const setupComplete = Boolean(d.setup_complete ?? d.initialized); + const accountReady = Boolean(d.has_user || !d.auth_enabled); + if (!cancelled) setReady(setupComplete && accountReady ? 'chat' : 'init'); + } catch { + if (!cancelled) setReady('chat'); + } + }; + + setReady(null); + void routeAfterInitCheck(); + return () => { cancelled = true; }; + }, [authRequired, loading, user?.role]); if (ready === 'init') return ; if (ready === 'chat') { diff --git a/frontend/src/pages/InitPage.tsx b/frontend/src/pages/InitPage.tsx index 1109cb7..0288c8d 100644 --- a/frontend/src/pages/InitPage.tsx +++ b/frontend/src/pages/InitPage.tsx @@ -3,6 +3,7 @@ import { useNavigate } from 'react-router-dom'; import { ArrowRight, Loader2, Terminal, Eye, EyeOff, CheckCircle, XCircle, AlertTriangle, ChevronRight, Check } from 'lucide-react'; import { cn } from '../lib/utils'; import { useAuth } from '../context/AuthContext'; +import { apiFetch } from '../lib/api'; const inp = 'w-full text-base text-gray-900 dark:text-white bg-gray-50 dark:bg-[#18181B] border border-gray-200 dark:border-[#27272A] rounded-xl px-4 py-3.5 outline-none focus:border-indigo-500 dark:focus:border-indigo-400 focus:ring-2 focus:ring-indigo-500/10 transition-all placeholder:text-gray-400 dark:placeholder:text-[#52525B]'; @@ -44,13 +45,13 @@ type Step = 1 | 2 | 3 | 4; export default function InitPage() { const navigate = useNavigate(); - const { user } = useAuth(); + const { user, login, authRequired, loading: authLoading } = useAuth(); - // Skip account creation if already logged in (registered via Login page) - const [step, setStep] = useState(user ? 2 : 1); + const [step, setStep] = useState(1); // Step 1 - const [username, setUsername] = useState('admin'); + const [orgName, setOrgName] = useState(''); + const [username, setUsername] = useState(''); const [password, setPassword] = useState(''); const [showPw, setShowPw] = useState(false); @@ -70,24 +71,36 @@ export default function InitPage() { const [error, setError] = useState(''); useEffect(() => { - fetch('/api/init/status') + if (authLoading) return; + if (user && user.role !== 'admin') { navigate('/', { replace: true }); return; } + apiFetch('/api/init/status') .then(r => r.json()) - .then(d => { if (d.has_user && d.initialized) navigate('/', { replace: true }); }) + .then(d => { + setRegion(d.aws_region || 'us-east-1'); + setSnsArn(d.sns_topic_arn || ''); + if (d.initialized && (d.has_user || !d.auth_enabled)) { + navigate('/', { replace: true }); + return; + } + setStep((user || !authRequired) ? 2 : 1); + }) .catch(() => {}); - }, [navigate]); + }, [authLoading, authRequired, navigate, user]); // ─── Step 1: create admin account ───────────────────────────────────────── const createUser = async () => { - if (!username.trim() || !password.trim()) { setError('Both fields are required'); return; } + if (!orgName.trim()) { setError('Organization name is required'); return; } + if (!username.trim() || !password.trim()) { setError('Email and password are required'); return; } setLoading(true); setError(''); try { - const r = await fetch('/api/init/create-user', { + const r = await apiFetch('/api/init/create-user', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ username: username.trim(), password }), + body: JSON.stringify({ username: username.trim(), password, org_name: orgName.trim() }), }); const data = await r.json(); - if (data.error) { setError(data.error); return; } + if (!r.ok || data.error) { setError(data.detail || data.error || 'Failed to create account'); return; } + if (authRequired) await login(username.trim(), password); setStep(2); } finally { setLoading(false); @@ -98,7 +111,7 @@ export default function InitPage() { const saveConfig = async () => { setLoading(true); setError(''); try { - const r = await fetch('/api/init/setup', { + const r = await apiFetch('/api/init/setup', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ aws_region: region, sns_topic_arn: snsArn, sqs_queue_url: '' }), @@ -114,7 +127,7 @@ export default function InitPage() { const checkPerms = async () => { setLoading(true); setError(''); try { - const r = await fetch('/api/init/check-permissions', { method: 'POST' }); + const r = await apiFetch('/api/init/check-permissions', { method: 'POST' }); const d = await r.json() as { permissions: Record }; setPerms(d.permissions); } catch { @@ -128,18 +141,35 @@ export default function InitPage() { const createInfra = async () => { setLoading(true); setError(''); try { - const r = await fetch('/api/init/complete', { method: 'POST' }); - const d = await r.json() as { initialized: boolean; error: string | null }; - if (!d.initialized) { setError(d.error || 'Infrastructure setup failed'); return; } + const r = await apiFetch('/api/init/complete', { method: 'POST' }); + const d = await r.json() as { initialized: boolean; error?: string | null; detail?: string }; + if (!r.ok || !d.initialized) { setError(d.error || d.detail || 'Infrastructure setup failed'); return; } setInfraDone(true); // Fetch the queue URL from init status to display in summary - const s = await fetch('/api/init/status').then(res => res.json()); + const s = await apiFetch('/api/init/status').then(res => res.json()); setQueueUrl(s.sqs_queue_url || 'opendevops-agent-events'); } finally { setLoading(false); } }; + const finishWithoutInfra = async () => { + setLoading(true); setError(''); + try { + const r = await apiFetch('/api/init/skip-services', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ services: ['event-monitoring'] }), + }); + if (!r.ok) throw new Error('finish failed'); + navigate('/', { replace: true }); + } catch { + setError('Failed to finish setup'); + } finally { + setLoading(false); + } + }; + const requiredPassed = Object.entries(perms) .filter(([svc]) => SVC[svc]?.required) .every(([, r]) => r.passed); @@ -191,8 +221,12 @@ export default function InitPage() {

- - setUsername(e.target.value)} placeholder="admin" className={inp} /> + + setOrgName(e.target.value)} placeholder="Acme Corp" className={inp} /> +
+
+ + setUsername(e.target.value)} placeholder="you@example.com" className={inp} />
@@ -202,7 +236,7 @@ export default function InitPage() { value={password} onChange={e => setPassword(e.target.value)} onKeyDown={e => e.key === 'Enter' && createUser()} - placeholder="At least 6 characters" + placeholder="At least 8 characters" className={inp + ' pr-11'} /> - @@ -400,7 +434,7 @@ export default function InitPage() {

You can enable this anytime from Settings → AWS Configuration.

- diff --git a/frontend/src/pages/MonitoringPage.tsx b/frontend/src/pages/MonitoringPage.tsx index 457a96b..c8c3995 100644 --- a/frontend/src/pages/MonitoringPage.tsx +++ b/frontend/src/pages/MonitoringPage.tsx @@ -32,9 +32,9 @@ export default function MonitoringPage() { useEffect(() => { if (!isAdmin) return; - fetch('/api/init/status') + apiFetch('/api/init/status') .then(r => r.json()) - .then(d => setInfraEnabled(!!d.initialized)) + .then(d => setInfraEnabled(!!d.event_infra_enabled)) .catch(() => setInfraEnabled(false)); }, [isAdmin]); @@ -75,6 +75,10 @@ export default function MonitoringPage() { const healthy = services.filter(s => s.status === 'healthy').length; const errored = services.filter(s => s.status === 'error').length; + const emptyTitle = infraEnabled === false ? 'Event monitoring is not enabled' : 'No activity yet'; + const emptyCopy = infraEnabled === false + ? 'Enable event monitoring in Settings, or turn on proactive polling, to populate this feed.' + : 'Listening for AWS events. Incidents will appear here when detected.'; const investigate = (a: Alert) => { const sessionId = crypto.randomUUID(); @@ -94,12 +98,12 @@ export default function MonitoringPage() {
- Live + {infraEnabled === false ? 'Feed' : 'Live'}
{isAdmin && (
)} 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/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/scripts/test_pipeline.py b/scripts/test_pipeline.py index eed3fe1..38f4aa6 100644 --- a/scripts/test_pipeline.py +++ b/scripts/test_pipeline.py @@ -12,7 +12,7 @@ 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 """ from __future__ import annotations @@ -26,17 +26,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 +70,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 +91,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 +122,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 +135,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 +147,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 +214,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 +246,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 +256,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..33395d0 100644 --- a/src/agent/db/base.py +++ b/src/agent/db/base.py @@ -102,8 +102,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 @@ -134,3 +141,11 @@ async def get_alerts(self, limit: int = 50) -> list[dict]: 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..b69ab12 100644 --- a/src/agent/db/memory.py +++ b/src/agent/db/memory.py @@ -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 @@ -262,6 +263,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..78e9a92 100644 --- a/src/agent/db/postgres.py +++ b/src/agent/db/postgres.py @@ -474,21 +474,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" @@ -573,6 +606,18 @@ async def get_alert(self, alert_id: str) -> dict | None: "timestamp": row["created_at"].isoformat() if row["created_at"] 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..6bd646c 100644 --- a/src/agent/db/sqlite.py +++ b/src/agent/db/sqlite.py @@ -71,6 +71,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] = { @@ -139,6 +145,16 @@ async def init(self) -> Any: except Exception: pass + 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) @@ -653,6 +669,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..f1578bb 100644 --- a/src/agent/event_consumer.py +++ b/src/agent/event_consumer.py @@ -10,9 +10,13 @@ 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, update_service from config import settings -from agent.init_store import load_init -from agent.monitor_store import add_alert def _build_investigation_prompt(event: dict) -> str: @@ -25,21 +29,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 +62,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" @@ -112,7 +126,11 @@ async def _run_investigation(prompt: str) -> dict[str, Any] | None: last = messages[-1] content = last.content if hasattr(last, "content") else str(last) if content: - return {"root_cause_summary": content, "confidence": "MEDIUM", "mitigation_steps": []} + return { + "root_cause_summary": content, + "confidence": "MEDIUM", + "mitigation_steps": [], + } except Exception as e: logger.error("Agent investigation failed: {}", e) return None @@ -121,21 +139,22 @@ async def _run_investigation(prompt: str) -> dict[str, Any] | None: async def _deliver(result: dict[str, Any], event: dict) -> None: """Send investigation result to SNS + Slack and persist to the monitor store.""" services_affected = result.get("services_affected", []) - service = result.get("service_name", result.get("service", - services_affected[0] if services_affected else "unknown" - )) + service = result.get( + "service_name", + result.get("service", services_affected[0] if services_affected else "unknown"), + ) 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) - 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 +162,17 @@ 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', '')}" ) 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"[{confidence}] {service} — {root_cause[:80]}", + message, ) sns_sent = True except Exception as e: @@ -158,12 +182,20 @@ async def _deliver(result: dict[str, Any], event: dict) -> None: 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) 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, "error", root_cause) + await add_alert( + service=service, + error=root_cause, + resolution=resolution, + confidence=confidence, + sns_sent=sns_sent, + ) logger.info("Alert delivered: service={} confidence={} sns={}", service, confidence, sns_sent) @@ -202,10 +234,15 @@ 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) @@ -218,15 +255,17 @@ async def _process_event(event: dict) -> None: 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..b88627b 100644 --- a/src/agent/init_store.py +++ b/src/agent/init_store.py @@ -1,52 +1,158 @@ -"""Init config store — persists initialization state to /init.json.""" +"""Init config store — persists setup and event-infra state. + +SQLite/Postgres deployments store the canonical value in the app_config table. +The JSON file remains a local cache and fallback for memory/dev mode and sync AWS tool code. +""" + +from __future__ import annotations import json from pathlib import Path +from typing import Any from loguru import logger from config import settings _INIT_FILE = Path(settings.data_dir) / "init.json" +_INIT_KEY = "init" +_CACHE: dict[str, Any] | None = None -_DEFAULT: dict = { - "initialized": False, - "sns_topic_arn": "", - "aws_region": "us-east-1", - "sqs_queue_url": "", - "sqs_queue_arn": "", - "eventbridge_rule_arns": {}, - "permissions": { - "cloudwatch": None, - "cloudtrail": None, - "ecs": None, - "lambda": None, - "ec2": None, - "rds": None, - "iam": None, - "sns": None, - "sqs": None, - "events": None, - }, - "skipped_services": [], -} +def _clone(data: dict) -> dict: + return json.loads(json.dumps(data)) -def load_init() -> dict: + +def _default() -> dict: + return { + "initialized": False, + "setup_complete": False, + "event_infra_enabled": False, + "event_infra_managed": False, + "sns_topic_arn": "", + "aws_region": settings.aws_region, + "sqs_queue_url": "", + "sqs_queue_arn": "", + "eventbridge_rule_arns": {}, + "permissions": { + "cloudwatch": None, + "cloudtrail": None, + "ecs": None, + "lambda": None, + "ec2": None, + "rds": None, + "iam": None, + "sns": None, + "sqs": None, + "events": None, + }, + "skipped_services": [], + } + + +def _with_defaults(data: dict) -> dict: + merged = _default() + permissions = {**merged["permissions"], **data.get("permissions", {})} + merged.update(data) + merged["permissions"] = permissions + + legacy_initialized = bool(data.get("initialized", False)) + merged["setup_complete"] = bool(data.get("setup_complete", legacy_initialized)) + merged["event_infra_enabled"] = bool( + data.get( + "event_infra_enabled", + legacy_initialized + and bool(data.get("sqs_queue_url") or data.get("eventbridge_rule_arns")), + ) + ) + merged["event_infra_managed"] = bool( + data.get( + "event_infra_managed", + bool(data.get("eventbridge_rule_arns") or data.get("sqs_queue_arn")), + ) + ) + merged["initialized"] = merged["setup_complete"] + return merged + + +def _read_file() -> dict | None: if not _INIT_FILE.exists(): - return json.loads(json.dumps(_DEFAULT)) + return None try: return json.loads(_INIT_FILE.read_text(encoding="utf-8")) except Exception as e: logger.warning("Failed to read init.json, returning defaults: {}", e) - return json.loads(json.dumps(_DEFAULT)) + return None -def save_init(data: dict) -> None: +def _write_file(data: dict) -> None: _INIT_FILE.parent.mkdir(parents=True, exist_ok=True) _INIT_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") logger.debug("Saved init.json") +def load_init() -> dict: + global _CACHE + if _CACHE is None: + _CACHE = _with_defaults(_read_file() or {}) + return _clone(_CACHE) + + +def save_init(data: dict) -> None: + global _CACHE + _CACHE = _with_defaults(data) + _write_file(_CACHE) + + +async def load_init_async() -> dict: + global _CACHE + try: + from agent.db import db + + data = await db.get_app_config(_INIT_KEY) + if data is not None: + _CACHE = _with_defaults(data) + _write_file(_CACHE) + return _clone(_CACHE) + file_data = load_init() + await db.set_app_config(_INIT_KEY, file_data) + return file_data + except Exception as e: + logger.debug("DB init config unavailable, using file fallback: {}", e) + return load_init() + + +async def save_init_async(data: dict) -> dict: + save_init(data) + try: + from agent.db import db + + await db.set_app_config(_INIT_KEY, load_init()) + except Exception as e: + logger.warning("Failed to persist init config to DB: {}", e) + return load_init() + + +async def refresh_init_cache_from_db() -> dict: + return await load_init_async() + + def is_initialized() -> bool: - return load_init().get("initialized", False) + return load_init().get("setup_complete", False) + + +def is_event_infra_enabled() -> bool: + data = load_init() + return bool(data.get("event_infra_enabled") and data.get("sqs_queue_url")) + + +def get_runtime_aws_region() -> str: + return str(load_init().get("aws_region") or settings.aws_region) + + +def get_runtime_sqs_queue_url() -> str: + return str(settings.sqs_queue_url or load_init().get("sqs_queue_url") or "") + + +def get_runtime_sns_topic_arn() -> str: + return str(settings.sns_topic_arn or load_init().get("sns_topic_arn") or "") diff --git a/src/agent/monitor_store.py b/src/agent/monitor_store.py index 30cd76e..ac0a233 100644 --- a/src/agent/monitor_store.py +++ b/src/agent/monitor_store.py @@ -2,7 +2,6 @@ from __future__ import annotations -import asyncio from datetime import UTC, datetime from loguru import logger @@ -11,21 +10,21 @@ _services: dict[str, dict] = {} -def add_alert(service: str, error: str, resolution: str, confidence: str, sns_sent: bool) -> None: - """Persist an alert to the DB backend. Called from async context.""" +async def add_alert( + service: str, + error: str, + resolution: str, + confidence: str, + sns_sent: bool, +) -> str: + """Persist an alert to the DB backend.""" from agent.db import db - async def _save() -> None: - await db.add_alert(service, error, resolution, confidence, sns_sent) - try: - loop = asyncio.get_event_loop() - if loop.is_running(): - asyncio.ensure_future(_save()) - else: - loop.run_until_complete(_save()) + return await db.add_alert(service, error, resolution, confidence, sns_sent) except Exception as e: logger.error("Failed to persist alert: {}", e) + return "" def update_service(name: str, status: str, error: str | None = None) -> None: @@ -39,11 +38,13 @@ def update_service(name: str, status: str, error: str | None = None) -> None: async def get_alerts_async(limit: int = 50) -> list[dict]: from agent.db import db + return await db.get_alerts(limit) async def get_alert_async(alert_id: str) -> dict | None: from agent.db import db + return await db.get_alert(alert_id) diff --git a/src/agent/permission_checker.py b/src/agent/permission_checker.py index b18b4c3..e5c85c7 100644 --- a/src/agent/permission_checker.py +++ b/src/agent/permission_checker.py @@ -20,10 +20,10 @@ def _check(fn) -> dict: return {"passed": False, "error": str(e)} -def check_permissions(sns_topic_arn: str = "") -> dict[str, dict]: +def check_permissions(sns_topic_arn: str = "", region: str | None = None) -> dict[str, dict]: """Run one lightweight read call per service. Returns {service: {passed, error}}.""" s = _session() - region = settings.aws_region + region = region or settings.aws_region results: dict[str, dict] = {} diff --git a/src/agent/poller.py b/src/agent/poller.py index e428af4..5c62865 100644 --- a/src/agent/poller.py +++ b/src/agent/poller.py @@ -64,7 +64,7 @@ async def _run_investigation(prompt: str, trigger_key: str) -> dict[str, Any] | async def _persist_and_notify(prompt: str, result: dict[str, Any], session_id: str) -> None: - from agent.monitor_store import add_alert + from agent.monitor_store import add_alert, update_service from agent.turns import notify_slack, save_turn tool_calls_log = [{"tool": "submit_investigation", "args": result, "result": {}}] @@ -78,9 +78,13 @@ async def _persist_and_notify(prompt: str, result: dict[str, Any], session_id: s mitigation = result.get("mitigation_steps", []) confidence = result.get("confidence", "MEDIUM") resolution = "\n".join(mitigation) if isinstance(mitigation, list) else str(mitigation) - add_alert( - service=service, error=root_cause, resolution=resolution, - confidence=confidence, sns_sent=False, + update_service(service, "error", root_cause) + await add_alert( + service=service, + error=root_cause, + resolution=resolution, + confidence=confidence, + sns_sent=False, ) @@ -92,10 +96,10 @@ async def _check_alarms() -> None: data = await asyncio.get_event_loop().run_in_executor(None, get_alarms, "ALARM") for alarm in data.get("alarms", []): - name = alarm.get("name", "unknown") - reason = alarm.get("reason", "") - metric = alarm.get("metric", "") - key = f"alarm:{name}" + name = alarm.get("name", "unknown") + reason = alarm.get("reason", "") + metric = alarm.get("metric", "") + key = f"alarm:{name}" if not _should_investigate(key): continue @@ -129,7 +133,10 @@ async def _check_lambda_errors() -> None: continue metrics = await asyncio.get_event_loop().run_in_executor( - None, get_lambda_error_rate, name, 1 # last 1 hour + None, + get_lambda_error_rate, + name, + 1, # last 1 hour ) error_rate = metrics.get("error_rate_pct", 0) or 0 if error_rate < settings.poll_error_threshold: diff --git a/src/agent/turns.py b/src/agent/turns.py index 4e212d2..d6bb5bb 100644 --- a/src/agent/turns.py +++ b/src/agent/turns.py @@ -6,26 +6,27 @@ from loguru import logger -from config import settings from agent.db import db +from agent.init_store import get_runtime_aws_region +from config import settings # Fallback pricing ($/M tokens) for models absent from LiteLLM's database. _FALLBACK_PRICING: dict[str, dict[str, float]] = { - "openrouter/google/gemma-4-26b-a4b-it": {"input": 0.07, "output": 0.35}, - "openrouter/google/gemma-2-9b-it": {"input": 0.06, "output": 0.06}, - "openrouter/meta-llama/llama-3.1-8b-instruct": {"input": 0.055, "output": 0.055}, - "openrouter/mistralai/mistral-7b-instruct": {"input": 0.055, "output": 0.055}, + "openrouter/google/gemma-4-26b-a4b-it": {"input": 0.07, "output": 0.35}, + "openrouter/google/gemma-2-9b-it": {"input": 0.06, "output": 0.06}, + "openrouter/meta-llama/llama-3.1-8b-instruct": {"input": 0.055, "output": 0.055}, + "openrouter/mistralai/mistral-7b-instruct": {"input": 0.055, "output": 0.055}, } def calc_cost(model: str, input_tok: int, output_tok: int) -> float | None: try: import litellm + info = litellm.model_cost.get(model) if info: - return ( - input_tok * info.get("input_cost_per_token", 0) - + output_tok * info.get("output_cost_per_token", 0) + return input_tok * info.get("input_cost_per_token", 0) + output_tok * info.get( + "output_cost_per_token", 0 ) fallback = _FALLBACK_PRICING.get(model) if fallback: @@ -45,18 +46,23 @@ async def save_turn( """Persist a completed turn to Postgres. Errors are logged, never raised.""" try: title = user_message[:80] if user_message else None - await db.upsert_session(session_id, usage.get("model", ""), settings.aws_region, title) + await db.upsert_session(session_id, usage.get("model", ""), get_runtime_aws_region(), title) - user_msg_id = await db.save_message(session_id, "user", user_message) + await db.save_message(session_id, "user", user_message) asst_msg_id = await db.save_message( - session_id, "assistant", assistant_text, + session_id, + "assistant", + assistant_text, metadata={"model": usage.get("model"), "latency_ms": usage.get("latency_ms")}, ) for tc in tool_calls_log: await db.save_tool_call( - session_id, asst_msg_id, - tc["tool"], tc["args"], tc["result"], + session_id, + asst_msg_id, + tc["tool"], + tc["args"], + tc["result"], ) cost = calc_cost( @@ -65,7 +71,8 @@ async def save_turn( usage.get("output_tokens", 0), ) await db.save_usage_event( - session_id, asst_msg_id, + session_id, + asst_msg_id, model=usage.get("model", ""), input_tokens=usage.get("input_tokens", 0), output_tokens=usage.get("output_tokens", 0), @@ -84,5 +91,6 @@ async def notify_slack(session_id: str, tool_calls_log: list[dict[str, Any]]) -> for tc in tool_calls_log: if tc.get("tool") == "submit_investigation": from integrations.slack_webhook import post_investigation + await post_investigation(settings.slack_webhook_url, tc["args"], session_id) return diff --git a/src/api/app.py b/src/api/app.py index 662842f..c245da3 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -13,10 +13,10 @@ # Suppress LiteLLM's "Provider List" info spam — we don't need provider discovery hints in logs litellm.suppress_debug_info = True -from agent.core import init_agent -from agent.db import db -from api.routers import auth, chat, dashboard, history, sessions, settings, users -from api.routers import init as init_router, monitoring +from agent.core import init_agent # noqa: E402 +from agent.db import db # noqa: E402 +from api.routers import init as init_router # noqa: E402 +from api.routers import auth, chat, dashboard, history, monitoring, sessions, settings, users # noqa: E402 class _InterceptHandler(logging.Handler): @@ -42,26 +42,50 @@ def emit(self, record: logging.LogRecord) -> None: def start_event_consumer(app_instance: "FastAPI | None" = None) -> None: """Start the SQS event consumer as a background task (idempotent).""" import asyncio + target = app_instance or app existing: asyncio.Task | None = getattr(target, "_consumer_task", None) if existing and not existing.done(): return from agent.event_consumer import event_consumer_loop + target._consumer_task = asyncio.create_task(event_consumer_loop()) # type: ignore[attr-defined] logger.info("Event consumer task started") +async def stop_event_consumer(app_instance: "FastAPI | None" = None) -> None: + """Stop the SQS event consumer if it is running.""" + import asyncio + + target = app_instance or app + existing: asyncio.Task | None = getattr(target, "_consumer_task", None) + if not existing or existing.done(): + return + existing.cancel() + try: + await existing + except asyncio.CancelledError: + pass + target._consumer_task = None # type: ignore[attr-defined] + logger.info("Event consumer task stopped") + + @asynccontextmanager async def lifespan(_app: FastAPI): import asyncio + from config import settings as _cfg checkpointer = await db.init() + from agent.init_store import refresh_init_cache_from_db + + await refresh_init_cache_from_db() init_agent(checkpointer) poller_task = None if _cfg.poll_interval_minutes > 0: from agent.poller import polling_loop + poller_task = asyncio.create_task(polling_loop()) logger.info("Proactive poller started (interval={}min)", _cfg.poll_interval_minutes) @@ -70,8 +94,9 @@ async def lifespan(_app: FastAPI): start_event_consumer(_app) else: try: - from agent.init_store import is_initialized - if is_initialized(): + from agent.init_store import is_event_infra_enabled + + if is_event_infra_enabled(): start_event_consumer(_app) except Exception: pass @@ -87,11 +112,7 @@ async def lifespan(_app: FastAPI): consumer_task: asyncio.Task | None = getattr(_app, "_consumer_task", None) if consumer_task: - consumer_task.cancel() - try: - await consumer_task - except asyncio.CancelledError: - pass + await stop_event_consumer(_app) await db.close() @@ -120,7 +141,8 @@ def _serve_index() -> HTMLResponse: if not path.exists(): return HTMLResponse( "

" - "Frontend not built yet. Run: cd frontend && npm install && npm run build" + "Frontend not built yet. Run: cd frontend && " + "npm install && npm run build" "

", status_code=503, ) diff --git a/src/api/routers/auth.py b/src/api/routers/auth.py index 646e0f9..7da5c0e 100644 --- a/src/api/routers/auth.py +++ b/src/api/routers/auth.py @@ -41,7 +41,9 @@ async def register(req: RegisterRequest) -> TokenResponse: raise HTTPException(status_code=400, detail="Email already registered") is_first = (await db.count_users()) == 0 role = "admin" if is_first else "user" - user = await db.create_user(req.email, req.name, hash_password(req.password), role) + org = await db.get_first_org() + org_id = org["id"] if org else None + user = await db.create_user(req.email, req.name, hash_password(req.password), role, org_id) if not user: raise HTTPException( status_code=501, diff --git a/src/api/routers/init.py b/src/api/routers/init.py index 445583c..8624d2a 100644 --- a/src/api/routers/init.py +++ b/src/api/routers/init.py @@ -5,23 +5,30 @@ import asyncio import datetime import json +from typing import Annotated import boto3 -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, Depends, HTTPException from loguru import logger -from pydantic import BaseModel +from pydantic import BaseModel, EmailStr, Field from agent.db import db -from agent.init_store import load_init, save_init -from api.auth import hash_password +from agent.init_store import ( + get_runtime_aws_region, + get_runtime_sqs_queue_url, + load_init_async, + save_init_async, +) +from api.auth import hash_password, require_admin from config import settings router = APIRouter(prefix="/api/init", tags=["init"]) class CreateUserBody(BaseModel): - username: str - password: str + username: EmailStr + password: str = Field(min_length=8) + org_name: str = Field(default="My Organization", min_length=1, max_length=100) class SetupBody(BaseModel): @@ -36,119 +43,204 @@ class SkipBody(BaseModel): @router.get("/status") async def status(): - data = load_init() - data["has_user"] = (await db.count_users()) > 0 + data = await load_init_async() + has_user = (await db.count_users()) > 0 + data["has_user"] = has_user + data["auth_enabled"] = bool(settings.jwt_secret) + data["needs_setup"] = not bool(data.get("setup_complete")) + data["initialized"] = bool(data.get("setup_complete")) + data["event_infra_enabled"] = bool(get_runtime_sqs_queue_url()) and bool( + settings.sqs_queue_url or data.get("event_infra_enabled") + ) return data @router.post("/create-user") async def create_user(body: CreateUserBody): - existing = await db.get_user_by_email(body.username) + username = str(body.username) + user_count = await db.count_users() + if user_count > 0: + raise HTTPException(status_code=409, detail="Initial admin account already exists") + existing = await db.get_user_by_email(username) if existing: - return {"error": "User already exists"} + raise HTTPException(status_code=409, detail="User already exists") + + # Create the organization first; slug = lowercase hyphenated name + import re + slug = re.sub(r"[^a-z0-9]+", "-", body.org_name.strip().lower()).strip("-") or "org" + org = await db.create_org(body.org_name.strip(), slug) + org_id = org["id"] if org else None + if org: + logger.info("Init: organization '{}' created (id={})", body.org_name, org_id) + user = await db.create_user( - email=body.username, - name=body.username, + email=username, + name=username, password_hash=hash_password(body.password), role="admin", + org_id=org_id, ) - logger.info("Init: admin user '{}' created", body.username) + logger.info("Init: admin user '{}' created", username) if user: - return {"id": user.get("id"), "username": body.username} - return {"error": "Failed to create user"} + return {"id": user.get("id"), "username": username, "org_id": org_id} + raise HTTPException(status_code=500, detail="Failed to create user") @router.post("/setup") -async def setup(body: SetupBody): - data = load_init() - data["sns_topic_arn"] = body.sns_topic_arn - data["aws_region"] = body.aws_region - if body.sqs_queue_url: - data["sqs_queue_url"] = body.sqs_queue_url - save_init(data) +async def setup( + body: SetupBody, + _: Annotated[dict | None, Depends(require_admin)], +): + data = await load_init_async() + data["sns_topic_arn"] = body.sns_topic_arn.strip() + data["aws_region"] = body.aws_region.strip() or settings.aws_region + data["sqs_queue_url"] = body.sqs_queue_url.strip() + data["event_infra_enabled"] = bool(data["sqs_queue_url"]) + data["event_infra_managed"] = False + if data["sqs_queue_url"]: + data["sqs_queue_arn"] = "" + data["eventbridge_rule_arns"] = {} + data = await save_init_async(data) logger.info("Init setup saved") return data @router.post("/check-permissions") -async def check_permissions_endpoint(): +async def check_permissions_endpoint( + _: Annotated[dict | None, Depends(require_admin)], +): from agent.permission_checker import check_permissions as _check - data = load_init() + data = await load_init_async() results = await asyncio.get_event_loop().run_in_executor( - None, _check, data.get("sns_topic_arn", "") + None, _check, data.get("sns_topic_arn", ""), data.get("aws_region") ) data["permissions"] = {svc: r["passed"] for svc, r in results.items()} - save_init(data) + await save_init_async(data) return {"permissions": results} @router.post("/skip-services") -async def skip_services(body: SkipBody): - data = load_init() +async def skip_services( + body: SkipBody, + _: Annotated[dict | None, Depends(require_admin)], +): + data = await load_init_async() data["skipped_services"] = body.services - save_init(data) + data["setup_complete"] = True + data["initialized"] = True + await save_init_async(data) return {"skipped_services": body.services} @router.post("/complete") -async def complete(): - data = load_init() +async def complete( + _: Annotated[dict | None, Depends(require_admin)], +): + data = await load_init_async() + region = data.get("aws_region") or settings.aws_region try: from agent.event_infra import setup_event_infra - result = await asyncio.get_event_loop().run_in_executor(None, setup_event_infra) + + result = await asyncio.get_event_loop().run_in_executor(None, setup_event_infra, region) data["sqs_queue_url"] = result["queue_url"] data["sqs_queue_arn"] = result["queue_arn"] data["eventbridge_rule_arns"] = result["rule_arns"] + data["event_infra_enabled"] = True + data["event_infra_managed"] = True logger.info("Init: event infra created") except Exception as e: logger.error("Init: event infra setup failed: {}", e) - return {"initialized": False, "error": str(e)} + raise HTTPException(status_code=500, detail=str(e)) + data["setup_complete"] = True data["initialized"] = True - save_init(data) + await save_init_async(data) logger.info("Init complete") # Hot-start the event consumer without requiring a server restart try: - from api.app import start_event_consumer + from api.app import start_event_consumer, stop_event_consumer + + await stop_event_consumer() start_event_consumer() except Exception as e: logger.warning("Could not hot-start event consumer (will start on next restart): {}", e) - return {"initialized": True, "error": None} + return {"initialized": True, "event_infra_enabled": True, "error": None} @router.delete("/infra") -async def teardown_infra(): +async def teardown_infra( + _: Annotated[dict | None, Depends(require_admin)], +): """Remove SQS queue and EventBridge rules created by /complete.""" - data = load_init() - queue_url = data.get("sqs_queue_url") or settings.sqs_queue_url + data = await load_init_async() + queue_url = data.get("sqs_queue_url") or "" rule_arns = data.get("eventbridge_rule_arns") or {} + region = data.get("aws_region") or settings.aws_region + managed = bool(data.get("event_infra_managed")) - if not queue_url: + if settings.sqs_queue_url and not managed: raise HTTPException( status_code=400, - detail="No SQS queue URL found — infrastructure may not be set up", + detail=( + "Event monitoring is configured by SQS_QUEUE_URL. Remove that environment " + "variable and restart to disable it from the app." + ), ) + if queue_url and not managed: + data["event_infra_enabled"] = False + data["sqs_queue_url"] = "" + await save_init_async(data) + try: + from api.app import stop_event_consumer + + await stop_event_consumer() + except Exception as e: + logger.warning("Could not stop event consumer after disabling external queue: {}", e) + return { + "torn_down": True, + "warnings": ["External SQS queue was disconnected but not deleted"], + } + + if not queue_url and not rule_arns: + data["event_infra_enabled"] = False + data["event_infra_managed"] = False + await save_init_async(data) + return {"torn_down": True, "warnings": ["No event infrastructure was configured"]} + try: from agent.event_infra import teardown_event_infra - await asyncio.get_event_loop().run_in_executor( - None, teardown_event_infra, queue_url, rule_arns + + result = await asyncio.get_event_loop().run_in_executor( + None, teardown_event_infra, queue_url, rule_arns, region ) except Exception as e: logger.error("Teardown failed: {}", e) raise HTTPException(status_code=500, detail=str(e)) - data["initialized"] = False + if result.get("errors"): + raise HTTPException(status_code=500, detail=result) + + data["event_infra_enabled"] = False data["sqs_queue_url"] = "" data["sqs_queue_arn"] = "" data["eventbridge_rule_arns"] = {} - save_init(data) + data["event_infra_managed"] = False + await save_init_async(data) + + try: + from api.app import stop_event_consumer + + await stop_event_consumer() + except Exception as e: + logger.warning("Could not stop event consumer after teardown: {}", e) + logger.info("Init: event infra torn down") - return {"torn_down": True} + return {"torn_down": True, "warnings": result.get("warnings", [])} class TestEventBody(BaseModel): @@ -156,10 +248,16 @@ class TestEventBody(BaseModel): @router.post("/test-event") -async def send_test_event(body: TestEventBody): +async def send_test_event( + body: TestEventBody, + _: Annotated[dict | None, Depends(require_admin)], +): """Put a synthetic event into the SQS queue to verify the pipeline end-to-end.""" - data = load_init() - queue_url = data.get("sqs_queue_url") or settings.sqs_queue_url + if body.service != "lambda": + raise HTTPException(status_code=400, detail="Only lambda test events are supported") + + await load_init_async() + queue_url = get_runtime_sqs_queue_url() if not queue_url: raise HTTPException( status_code=400, @@ -184,17 +282,19 @@ async def send_test_event(body: TestEventBody): ), }, "configuration": { - "metrics": [{ - "metricStat": { - "metric": { - "namespace": "AWS/Lambda", - "name": "Errors", - "dimensions": {}, - }, - "period": 300, - "stat": "Sum", + "metrics": [ + { + "metricStat": { + "metric": { + "namespace": "AWS/Lambda", + "name": "Errors", + "dimensions": {}, + }, + "period": 300, + "stat": "Sum", + } } - }] + ] }, }, } @@ -205,7 +305,7 @@ async def send_test_event(body: TestEventBody): if settings.aws_profile else boto3.Session() ) - sqs = s.client("sqs", region_name=settings.aws_region) + sqs = s.client("sqs", region_name=get_runtime_aws_region()) resp = await asyncio.get_event_loop().run_in_executor( None, lambda: sqs.send_message(QueueUrl=queue_url, MessageBody=json.dumps(event)), diff --git a/src/api/routers/settings.py b/src/api/routers/settings.py index 9f0138a..3d8ddeb 100644 --- a/src/api/routers/settings.py +++ b/src/api/routers/settings.py @@ -8,6 +8,7 @@ from fastapi import APIRouter, Depends +from agent.init_store import get_runtime_aws_region from api.auth import get_current_user from config import settings @@ -43,6 +44,7 @@ async def get_settings( ] agent = [ + {"key": "EFFECTIVE_AWS_REGION", "label": "Effective AWS region", "value": get_runtime_aws_region(), "hint": "Wizard setting, falling back to AWS_REGION"}, {"key": "MAX_TOOL_CALLS", "label": "Max tool calls", "value": str(settings.max_tool_calls), "hint": "Hard cap per investigation run"}, {"key": "INVESTIGATION_TIMEOUT", "label": "Timeout (s)", "value": str(settings.investigation_timeout), "hint": "Max seconds before run is cancelled"}, {"key": "TOOL_RESPONSE_MAX_CHARS", "label": "Tool response cap", "value": str(settings.tool_response_max_chars), "hint": "Chars before tool output is truncated"}, diff --git a/src/tools/_cache.py b/src/tools/_cache.py index d42f6ce..59edcd0 100644 --- a/src/tools/_cache.py +++ b/src/tools/_cache.py @@ -18,6 +18,7 @@ from cachetools import TTLCache, cached from cachetools.keys import hashkey +from agent.init_store import get_runtime_aws_region from config import settings F = TypeVar("F", bound=Callable[..., Any]) @@ -41,7 +42,7 @@ def _h(v: Any) -> Any: return hashkey( fn_name, settings.aws_profile, - settings.aws_region, + get_runtime_aws_region(), *[_h(a) for a in args], **{k: _h(v) for k, v in kwargs.items()}, ) diff --git a/src/tools/cloudtrail.py b/src/tools/cloudtrail.py index 6786b01..f57a470 100644 --- a/src/tools/cloudtrail.py +++ b/src/tools/cloudtrail.py @@ -5,17 +5,20 @@ import boto3 from botocore.exceptions import BotoCoreError, ClientError - from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings from tools._cache import tool_cached - def _ct_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("cloudtrail", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("cloudtrail", region_name=get_runtime_aws_region()) @tool_cached @@ -25,7 +28,7 @@ def lookup_cloudtrail_events( event_name: str | None = None, limit: int = 50, ) -> dict: - """Look up recent CloudTrail API events to find deployments, config changes, and permission changes. + """Look up recent CloudTrail API events. Args: hours: How far back to look. Default 2. @@ -44,9 +47,13 @@ def lookup_cloudtrail_events( "MaxResults": min(limit, 50), } if resource_name: - kwargs["LookupAttributes"] = [{"AttributeKey": "ResourceName", "AttributeValue": resource_name}] + kwargs["LookupAttributes"] = [ + {"AttributeKey": "ResourceName", "AttributeValue": resource_name} + ] elif event_name: - kwargs["LookupAttributes"] = [{"AttributeKey": "EventName", "AttributeValue": event_name}] + kwargs["LookupAttributes"] = [ + {"AttributeKey": "EventName", "AttributeValue": event_name} + ] resp = client.lookup_events(**kwargs) events = [] diff --git a/src/tools/cloudwatch.py b/src/tools/cloudwatch.py index 9b1beea..12038c9 100644 --- a/src/tools/cloudwatch.py +++ b/src/tools/cloudwatch.py @@ -6,22 +6,29 @@ import boto3 from botocore.exceptions import BotoCoreError, ClientError - from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings from tools._cache import tool_cached - def _cw_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("cloudwatch", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("cloudwatch", region_name=get_runtime_aws_region()) def _logs_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("logs", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("logs", region_name=get_runtime_aws_region()) @tool_cached @@ -117,10 +124,19 @@ def get_metric_data( Statistics=[stat], ) datapoints = sorted( - [{"timestamp": dp["Timestamp"].isoformat(), "value": dp[stat]} for dp in resp.get("Datapoints", [])], + [ + {"timestamp": dp["Timestamp"].isoformat(), "value": dp[stat]} + for dp in resp.get("Datapoints", []) + ], key=lambda x: x["timestamp"], ) - return {"namespace": namespace, "metric": metric, "stat": stat, "datapoints": datapoints, "count": len(datapoints)} + return { + "namespace": namespace, + "metric": metric, + "stat": stat, + "datapoints": datapoints, + "count": len(datapoints), + } except (BotoCoreError, ClientError) as e: logger.error("get_metric_data failed: {}", e) return {"error": str(e), "datapoints": []} @@ -214,7 +230,7 @@ def query_logs_insights( Supports the full Logs Insights query language: fields, filter, stats, sort, limit. Example queries: - - 'fields @timestamp, @message | filter @message like /ERROR/ | sort @timestamp desc | limit 50' + - 'fields @timestamp, @message | filter @message like /ERROR/ | limit 50' - 'stats count(*) as errors by bin(5m) | sort errors desc' - 'fields @timestamp, @message | filter @message like /WARN/ | limit 100' diff --git a/src/tools/ec2.py b/src/tools/ec2.py index 580a962..8d8e9e7 100644 --- a/src/tools/ec2.py +++ b/src/tools/ec2.py @@ -4,17 +4,20 @@ import boto3 from botocore.exceptions import BotoCoreError, ClientError - from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings from tools._cache import tool_cached - def _ec2_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("ec2", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("ec2", region_name=get_runtime_aws_region()) @tool_cached @@ -22,7 +25,7 @@ def describe_ec2_instances(filters: list[dict[str, Any]] | None = None) -> dict: """List EC2 instances with their state, type, and tags. Optionally filter by state or tag. Args: - filters: EC2 describe-instances filters, e.g. [{"Name": "instance-state-name", "Values": ["running"]}]. + filters: EC2 describe-instances filters. """ try: client = _ec2_client() diff --git a/src/tools/ecs.py b/src/tools/ecs.py index 46af9cc..d03f186 100644 --- a/src/tools/ecs.py +++ b/src/tools/ecs.py @@ -4,22 +4,29 @@ import boto3 from botocore.exceptions import BotoCoreError, ClientError - from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings from tools._cache import tool_cached - def _ecs_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("ecs", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("ecs", region_name=get_runtime_aws_region()) def _logs_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("logs", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("logs", region_name=get_runtime_aws_region()) @tool_cached diff --git a/src/tools/iam.py b/src/tools/iam.py index a35c965..16d0a37 100644 --- a/src/tools/iam.py +++ b/src/tools/iam.py @@ -4,22 +4,29 @@ import boto3 from botocore.exceptions import BotoCoreError, ClientError - from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings from tools._cache import tool_cached - def _iam_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("iam", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("iam", region_name=get_runtime_aws_region()) def _sts_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("sts", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("sts", region_name=get_runtime_aws_region()) @tool_cached diff --git a/src/tools/lambda_.py b/src/tools/lambda_.py index 1b6bc64..6463762 100644 --- a/src/tools/lambda_.py +++ b/src/tools/lambda_.py @@ -5,22 +5,29 @@ import boto3 from botocore.exceptions import BotoCoreError, ClientError - from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings from tools._cache import tool_cached - def _lambda_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("lambda", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("lambda", region_name=get_runtime_aws_region()) def _cw_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("cloudwatch", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("cloudwatch", region_name=get_runtime_aws_region()) @tool_cached @@ -100,7 +107,10 @@ def _get_sum(metric: str) -> list[dict[str, Any]]: Statistics=["Sum"], ) return sorted( - [{"timestamp": dp["Timestamp"].isoformat(), "value": dp["Sum"]} for dp in resp.get("Datapoints", [])], + [ + {"timestamp": dp["Timestamp"].isoformat(), "value": dp["Sum"]} + for dp in resp.get("Datapoints", []) + ], key=lambda x: x["timestamp"], ) @@ -118,7 +128,9 @@ def _get_sum(metric: str) -> list[dict[str, Any]]: "total_invocations": total_invocations, "total_errors": total_errors, "total_throttles": total_throttles, - "error_rate_pct": round(total_errors / total_invocations * 100, 2) if total_invocations else 0, + "error_rate_pct": round(total_errors / total_invocations * 100, 2) + if total_invocations + else 0, "errors_by_period": errors, "throttles_by_period": throttles, } diff --git a/src/tools/rds.py b/src/tools/rds.py index f57dd1d..626740f 100644 --- a/src/tools/rds.py +++ b/src/tools/rds.py @@ -5,17 +5,20 @@ import boto3 from botocore.exceptions import BotoCoreError, ClientError - from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings from tools._cache import tool_cached - def _rds_client() -> Any: - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("rds", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("rds", region_name=get_runtime_aws_region()) @tool_cached diff --git a/src/tools/sns.py b/src/tools/sns.py index d7a5041..440ee6e 100644 --- a/src/tools/sns.py +++ b/src/tools/sns.py @@ -3,12 +3,17 @@ import boto3 from loguru import logger +from agent.init_store import get_runtime_aws_region from config import settings def _sns_client(): - session = boto3.Session(profile_name=settings.aws_profile) if settings.aws_profile else boto3.Session() - return session.client("sns", region_name=settings.aws_region) + session = ( + boto3.Session(profile_name=settings.aws_profile) + if settings.aws_profile + else boto3.Session() + ) + return session.client("sns", region_name=get_runtime_aws_region()) def publish_sns_alert(topic_arn: str, subject: str, message: str) -> dict: From c16deb13999a1f6c03e15cd238aa90c9224413b7 Mon Sep 17 00:00:00 2001 From: AhmadHammad21 Date: Sat, 16 May 2026 23:34:59 +0300 Subject: [PATCH 05/16] fix: add /api/init/reset endpoint to clear setup state for re-running wizard --- src/api/routers/init.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/api/routers/init.py b/src/api/routers/init.py index 8624d2a..88c8b8b 100644 --- a/src/api/routers/init.py +++ b/src/api/routers/init.py @@ -41,6 +41,24 @@ class SkipBody(BaseModel): services: list[str] +@router.post("/reset") +async def reset_init_state( + _: Annotated[dict | None, Depends(require_admin)], +): + """Clear persisted init/setup state. Useful for re-running the wizard without touching AWS.""" + from pathlib import Path as _Path + + from agent.init_store import _INIT_FILE, _default, save_init_async + + await save_init_async(_default()) + try: + if _Path(_INIT_FILE).exists(): + _Path(_INIT_FILE).unlink() + except Exception: + pass + return {"reset": True} + + @router.get("/status") async def status(): data = await load_init_async() From 28d6a046c429aeaed70a0fb8e90cd92554a901f0 Mon Sep 17 00:00:00 2001 From: AhmadHammad21 Date: Sat, 16 May 2026 23:57:56 +0300 Subject: [PATCH 06/16] feat: move org name to Step 2, eliminate init.json, cache app_config in memory - InitPage: org name input moved from Step 1 to Step 2 (AWS config step) so it's shown regardless of whether the user is already logged in - /api/init/create-user no longer accepts org_name; /api/init/setup now creates the org when org_name is provided and none exists yet - init_store: replace file-based persistence with DB (app_config table) + in-memory _CACHE; sync load_init() reads from cache with zero I/O, safe for boto3 thread executors; cache is seeded from DB on startup via refresh_init_cache_from_db() in the FastAPI lifespan - Added reset_init_async() helper; /api/init/reset uses it directly --- frontend/src/pages/InitPage.tsx | 18 +++++---- src/agent/init_store.py | 72 ++++++++++++++++++--------------- src/api/routers/init.py | 36 ++++++++--------- 3 files changed, 67 insertions(+), 59 deletions(-) diff --git a/frontend/src/pages/InitPage.tsx b/frontend/src/pages/InitPage.tsx index 0288c8d..22a0838 100644 --- a/frontend/src/pages/InitPage.tsx +++ b/frontend/src/pages/InitPage.tsx @@ -89,14 +89,13 @@ export default function InitPage() { // ─── Step 1: create admin account ───────────────────────────────────────── const createUser = async () => { - if (!orgName.trim()) { setError('Organization name is required'); return; } if (!username.trim() || !password.trim()) { setError('Email and password are required'); return; } setLoading(true); setError(''); try { const r = await apiFetch('/api/init/create-user', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ username: username.trim(), password, org_name: orgName.trim() }), + body: JSON.stringify({ username: username.trim(), password }), }); const data = await r.json(); if (!r.ok || data.error) { setError(data.detail || data.error || 'Failed to create account'); return; } @@ -114,7 +113,7 @@ export default function InitPage() { const r = await apiFetch('/api/init/setup', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ aws_region: region, sns_topic_arn: snsArn, sqs_queue_url: '' }), + body: JSON.stringify({ aws_region: region, sns_topic_arn: snsArn, sqs_queue_url: '', org_name: orgName.trim() }), }); if (!r.ok) { setError('Failed to save configuration'); return; } setStep(3); @@ -220,10 +219,6 @@ export default function InitPage() { Set up the admin credentials you'll use to access the dashboard.

-
- - setOrgName(e.target.value)} placeholder="Acme Corp" className={inp} /> -
setUsername(e.target.value)} placeholder="you@example.com" className={inp} /> @@ -262,6 +257,15 @@ export default function InitPage() { Configure the AWS region and optional SNS topic for alert delivery.

+
+ + setOrgName(e.target.value)} placeholder="Acme Corp" className={inp} /> +

+ Used to group users. You can add team members later from Settings. +

+