diff --git a/CLAUDE.md b/CLAUDE.md index 3c3d74b..6d4e10d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -101,10 +101,10 @@ Everything below is built and working in the codebase: ### Event-Driven Detection - `event_consumer_loop()` long-polls SQS (20s wait), processes EventBridge events (CloudWatch alarm, ECS task failure, Lambda async error, RDS event, EC2 state change, CodePipeline failure, AWS Health), runs a full agent investigation per event, delivers results to SNS + Slack, persists to `alerts` table. - `context_collectors.collect_context()` pre-fetches resource facts deterministically before the LLM runs to reduce tool call count. -- Starts automatically on app startup if `event_consumer_enabled=True`, `sqs_queue_url` is set, or `init.json` marks the wizard as completed. +- Starts automatically on app startup if `event_consumer_enabled=True`, `sqs_queue_url` is set, or database-backed app config marks event infrastructure as enabled. Autonomous monitoring requires SQLite or PostgreSQL; memory mode is disabled for poller/consumer runs. ### Proactive Polling -- `polling_loop()` runs every `POLL_INTERVAL_SECONDS` seconds (disabled by default at 0); checks CloudWatch alarms in ALARM state and Lambda error rates above `POLL_ERROR_THRESHOLD`; auto-investigates new anomalies and posts to Slack. In-memory dedup via `_last_investigated` map. +- `polling_loop()` runs every `POLL_INTERVAL_SECONDS` seconds (disabled by default at 0); checks CloudWatch alarms in ALARM state and Lambda error rates above `POLL_ERROR_THRESHOLD`; auto-investigates new anomalies and posts to Slack/Telegram. Dedup uses canonical incident keys and durable DB-backed claims. ### Frontend - React 18 + TypeScript + Vite + Tailwind CSS + `@tailwindcss/typography` @@ -225,10 +225,10 @@ POLL_REINVESTIGATE_HOURS=1 # dedup window # Event-driven detection SNS_TOPIC_ARN= # SNS publish target after investigations SQS_QUEUE_URL= # SQS queue for EventBridge events -EVENT_CONSUMER_ENABLED=false # also auto-starts if SQS_QUEUE_URL is set or init.json exists +EVENT_CONSUMER_ENABLED=false # also auto-starts if SQS_QUEUE_URL is set or app config enables infra # Misc -DATA_DIR=data # server-side state files (init.json) +DATA_DIR=data # reserved for future file-based state ``` --- @@ -262,7 +262,7 @@ EventBridge rules (9 event types) ``` db.init() → init_agent(checkpointer) optional: asyncio.create_task(polling_loop()) if POLL_INTERVAL_SECONDS > 0 - optional: asyncio.create_task(event_consumer_loop()) if SQS configured or init.json present + optional: asyncio.create_task(event_consumer_loop()) if SQS configured or app config enables infra ``` ### Session continuity diff --git a/README.md b/README.md index c927d5c..40e1039 100644 --- a/README.md +++ b/README.md @@ -10,11 +10,11 @@ and gives actionable mitigation plans — without the AWS DevOps Agent price tag - **Sandboxed bash execution tool** — agent can run whitelisted read-only AWS CLI, kubectl, and docker commands as a last resort when the structured tools fall short; every command validated against an allowlist before execution; never uses `shell=True`; hard 30-second timeout - Includes **CloudWatch Logs Insights** (`query_logs_insights`) — full query language support: `fields`, `filter`, `stats`, `sort`, `limit`; results include scanned MB - **Streaming responses** — FastAPI SSE endpoint streams agent tokens in real time as the LLM reasons; tool calls appear as they complete -- **Event-driven incident detection** — EventBridge → SQS → long-poll consumer; 9 EventBridge rules cover CloudWatch alarms, ECS task failures, Lambda async errors, RDS events, EC2 state changes, CodePipeline failures, and AWS Health events; runs alongside the metric poller — see [docs/event_detection.md](docs/event_detection.md) +- **Event-driven incident detection** — EventBridge → SQS → long-poll consumer; 9 EventBridge rules cover CloudWatch alarms, ECS task failures, Lambda async errors, RDS events, EC2 state changes, CodePipeline failures, and AWS Health events; uses a DLQ plus database-backed incident claims to avoid duplicate investigations; runs alongside the metric poller — see [docs/event_detection.md](docs/event_detection.md) - **Context enrichment** — before the LLM runs, deterministic boto3 calls fetch facts about the affected resource (alarm details, recent logs, function config, etc.) to reduce tool call count and speed up investigations - **SNS alert delivery** — after each event-driven investigation, findings are published to a configured SNS topic as structured JSON; fires alongside Slack and Telegram so all channels receive results simultaneously - **Monitoring dashboard** — live incident feed showing all event-driven investigations: confidence level (or FAILED badge), affected service, root cause summary, SNS status; each alert links back to its original investigation session via **View investigation** so you can follow up without losing context — see [docs/monitoring.md](docs/monitoring.md) -- **AWS Configuration settings tab** — admin-only editable tab in Settings for SNS Topic ARN, SQS Queue URL, and AWS Region; shared org-wide (server-side `init.json`); includes an inline IAM permission checker per service +- **AWS Configuration settings tab** — admin-only editable tab in Settings for SNS Topic ARN, SQS Queue URL, and AWS Region; shared org-wide via database-backed app config; includes an inline IAM permission checker per service - **Web UI** — React + Vite SPA served by FastAPI: - **Chat page** — streaming responses, collapsible tool call inspector, cost/latency card, stop button; supports `?prompt=` deeplink for pre-seeded investigations from the Monitoring dashboard - **Session history sidebar** — lists all past conversations; click any to resume with full tool call inspector and cost card restored; new chat and delete (soft) buttons @@ -25,7 +25,7 @@ and gives actionable mitigation plans — without the AWS DevOps Agent price tag - **Team page** — admin-only user management: add, remove, and change roles - **Auth & RBAC** — optional password-based auth with `admin` and `user` roles; JWT tokens; first registered user auto-becomes admin; disabled by default (set `JWT_SECRET` to enable) — see [docs/auth.md](docs/auth.md) - **Three storage backends** — pick one via `CHECKPOINT_BACKEND` in `.env`; see [`docs/databases.md`](docs/databases.md) - - `memory` — zero config, no persistence; great for CI and quick testing + - `memory` — zero config, no persistence; great for CI and quick testing; autonomous polling/event monitoring is disabled in this mode - `sqlite` — local file, no external services; recommended for single-server and personal use - `postgres` — full production persistence via psycopg3 + `AsyncPostgresSaver` - Schema: `users`, `sessions`, `messages`, `tool_calls`, `usage_events` — see [`docs/schema.md`](docs/schema.md) @@ -210,7 +210,7 @@ docs/ # Feature reference — auth, schema, skills, databases, | `SLACK_WEBHOOK_URL` | none | Slack incoming webhook URL; leave unset to disable notifications | | `TELEGRAM_BOT_TOKEN` | none | Telegram bot token from @BotFather; leave unset to disable | | `TELEGRAM_CHAT_ID` | none | Target chat/group/channel ID (negative number for groups) | -| `POLL_INTERVAL_MINUTES` | `0` | Proactive polling interval in minutes; `0` disables the poller | +| `POLL_INTERVAL_SECONDS` | `0` | Proactive polling interval in seconds; `0` disables the poller | | `POLL_ERROR_THRESHOLD` | `5.0` | Lambda error rate % that triggers an automatic investigation | | `POLL_REINVESTIGATE_HOURS` | `1` | Cooldown period — skip re-investigating the same alarm within N hours | | `SUMMARIZATION_ENABLED` | `true` | Auto-compact sessions when they exceed the threshold | @@ -221,7 +221,7 @@ docs/ # Feature reference — auth, schema, skills, databases, | `SNS_TOPIC_ARN` | none | SNS topic to publish investigation findings to after each event-driven run | | `SQS_QUEUE_URL` | none | SQS queue URL for the event consumer to poll; also set via Settings → AWS Configuration | | `EVENT_CONSUMER_ENABLED` | `false` | Explicitly enable the SQS event consumer (also auto-starts if `SQS_QUEUE_URL` is set) | -| `DATA_DIR` | `data` | Directory for server-side state files (`init.json`) | +| `DATA_DIR` | `data` | Reserved data directory setting; init state is stored in the selected database backend | ## TODO / Roadmap @@ -254,7 +254,7 @@ docs/ # Feature reference — auth, schema, skills, databases, - [x] **SNS alert delivery** — findings published to SNS after each event-driven investigation; fires alongside Slack; structured JSON payload for easy downstream processing - [x] **Context enrichment** — deterministic boto3 calls per event type before LLM runs; reduces tool call count by front-loading relevant resource facts - [x] **Monitoring dashboard** — live incident feed, per-service health summary, alert detail page; "View investigation" opens the original agent session for follow-up; failed investigations flagged separately; see [docs/monitoring.md](docs/monitoring.md) -- [x] **AWS Configuration settings tab** — admin-only editable tab for SNS/SQS/region config; shared org-wide via server-side `init.json`; inline IAM permission checker +- [x] **AWS Configuration settings tab** — admin-only editable tab for SNS/SQS/region config; shared org-wide via database-backed app config; inline IAM permission checker ### Later - [ ] **Observability** — OpenTelemetry traces for agent steps, tool call latency, LLM token usage diff --git a/docs/auth.md b/docs/auth.md index 783221b..6071e0c 100644 --- a/docs/auth.md +++ b/docs/auth.md @@ -17,6 +17,14 @@ Restart the server. The login page will appear on next browser load. **Without `JWT_SECRET`:** the app runs in dev/open mode — no login required, all users treated as admin. Suitable for local development or single-user installs on a trusted network. +Auth works with all storage backends, but persistence follows the backend: + +| Backend | Auth behavior | +|---|---| +| `memory` | Users exist only until the server restarts; useful for quick local testing | +| `sqlite` | Users persist in the local SQLite database | +| `postgres` | Users persist in PostgreSQL and are suitable for production/team use | + --- ## First-time setup diff --git a/docs/databases.md b/docs/databases.md index 0bc539e..5e3ed50 100644 --- a/docs/databases.md +++ b/docs/databases.md @@ -9,7 +9,7 @@ OpenDevOps Agent supports three storage backends. Pick one per deployment — se | Backend | Persistence | External service | Dashboard | Best for | |------------|-------------|-----------------|-----------|-----------------------------------| -| `memory` | None | None | Counts only | CI, quick demos, local testing | +| `memory` | None | None | Counts only; no autonomous monitoring | CI, quick demos, local testing | | `sqlite` | Local file | None | Full | Single-server, personal use | | `postgres` | Database | PostgreSQL 14+ | Full | Production, teams, multi-instance | @@ -24,6 +24,13 @@ CHECKPOINT_BACKEND=memory Everything lives in Python dicts for the life of the process. On restart, all sessions and history are gone. The LangGraph checkpointer uses `MemorySaver`. +Autonomous monitoring is disabled in memory mode. The poller and SQS event +consumer require durable incident claims, so use SQLite or PostgreSQL for those +features. + +Auth/register/login work in memory mode when `JWT_SECRET` is set, but users are +lost on restart with the rest of the in-memory state. + **When to use:** CI pipelines, smoke-testing, one-off demos. > **Dashboard limitation:** summary counts (sessions, queries, tool calls, cost) update @@ -45,8 +52,7 @@ 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. +monitoring infrastructure state in `app_config`. The file and its parent directory are created automatically on first start. diff --git a/docs/event_detection.md b/docs/event_detection.md index 0316cde..26e20f2 100644 --- a/docs/event_detection.md +++ b/docs/event_detection.md @@ -4,7 +4,7 @@ OpenDevOps Agent supports two complementary detection modes that run simultaneou | Mode | Source | Trigger | |---|---|---| -| **Proactive polling** | CloudWatch Alarms + Lambda metrics | Checked every N minutes by a background loop | +| **Proactive polling** | CloudWatch Alarms + Lambda metrics | Checked every N seconds by a background loop | | **Event-driven consumer** | EventBridge → SQS | Real AWS infrastructure events delivered in near-real time | This page covers the event-driven consumer. For the polling loop see [slack_and_polling.md](slack_and_polling.md). @@ -16,7 +16,7 @@ This page covers the event-driven consumer. For the polling loop see [slack_and_ ``` AWS Service (ECS, Lambda, RDS, EC2…) → CloudWatch → EventBridge rule - → SQS queue (OpenDevOps) + → SQS queue (OpenDevOps, with DLQ) → SQS long-poll consumer (event_consumer_loop) → context_collectors (deterministic boto3 enrichment) → agent.ainvoke (LLM investigation) @@ -84,15 +84,19 @@ Context is appended to the investigation prompt, capped at 3 000 chars to stay w 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. Deduplicates: checks an in-memory set (`_in_progress`) and the database to skip events already being investigated or investigated within the last 3 minutes +3. Deduplicates: builds a canonical incident key and atomically claims it in the database before the agent runs 4. Calls `collect_context(event)` to enrich the prompt 5. Generates a `session_id` and runs a full agent investigation via `agent.ainvoke`, persisting the session and messages to the database (`source = 'event'`) 6. Persists the result to the `alerts` table (linked to the session via `session_id`) via `monitor_store.add_alert()` 7. Delivers findings via SNS and Slack (if configured); failed investigations are flagged with status `'failed'` but are still persisted and notified -8. Deletes the SQS message +8. Deletes the SQS message only after successful processing, an intentional ignore, or a duplicate claim. Processing failures are left on the queue for retry and eventual DLQ redrive. The consumer is started as an `asyncio.Task` in the FastAPI lifespan and shut down cleanly on server stop. +The setup wizard creates both `opendevops-agent-events` and `opendevops-agent-events-dlq`. The main queue has a redrive policy with `maxReceiveCount=5`; poison messages remain available in the DLQ for inspection instead of being dropped by the consumer. + +Deduplication uses canonical incident keys rather than hashes of whole event payloads. Examples: `cloudwatch_alarm:us-east-1:high-error-rate`, `lambda_errors:us-east-1:payment-processor`, and `lambda_error:us-east-1:payment-processor:`. Lambda EventBridge failures with different error signatures are separate incidents. Metric polling only sees function-level error rates, so those incidents are deduped at the function level during the cooldown window. + --- ## SNS Alert Delivery @@ -137,9 +141,9 @@ Each event-driven investigation creates a full session (messages, tool calls, us Sessions created by the event consumer carry `source = 'event'` and are hidden from the sidebar by default. They become visible after the user sends a follow-up message in the chat (`user_interacted` is then set to `true`). -The `alerts` table records `status` (`'completed'` or `'failed'`), `dedup_key` (MD5 fingerprint of stable event fields for cross-run deduplication), and `session_id` (FK to the session that produced it). +The `alerts` table records `status` (`'completed'` or `'failed'`), `dedup_key` (the canonical incident key), `session_id` (FK to the session that produced it), and `trigger_source` (`'poller'` or `'event_consumer'`). The `incident_claims` table stores the atomic pre-investigation claim used to prevent duplicate agent runs across workers and restarts. -All three storage backends (memory, SQLite, PostgreSQL) support alerts. For PostgreSQL, run migrations before starting the server: +Autonomous polling and event-driven monitoring require SQLite or PostgreSQL because incident claims must be durable. The memory backend remains available for chat, CI, and quick demos, but the poller, event consumer, and setup wizard are disabled in memory mode. For PostgreSQL, run migrations before starting the server: ```bash uv run python scripts/setup_db.py @@ -156,9 +160,10 @@ For SQLite, schema migrations are applied automatically on first start. | `src/agent/event_consumer.py` | SQS long-poll loop, filtering, prompt building, delivery | | `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 DB-backed `app_config` with `data/init.json` fallback/cache | +| `src/agent/monitor_store.py` | In-process service status + DB-backed alerts, notifications, and incident claims | +| `src/agent/init_store.py` | Persist init config to DB-backed `app_config` | | `src/agent/permission_checker.py` | IAM permission validation | | `src/tools/sns.py` | SNS publish wrapper | | `src/api/routers/init.py` | Setup wizard API endpoints | | `migrations/005_alerts.sql` | PostgreSQL alerts table + indexes | +| `migrations/012_incident_claims.sql` | PostgreSQL incident claim table for atomic deduplication | diff --git a/docs/monitoring.md b/docs/monitoring.md index 7e7f96a..89daf8a 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -11,7 +11,7 @@ OpenDevOps Agent has two complementary detection modes. Both write to the same ` | | Proactive Polling | Event-Driven (EventBridge → SQS) | |---|---|---| | **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 | +| **Setup** | Set `POLL_INTERVAL_SECONDS` 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 | 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 | @@ -28,7 +28,7 @@ OpenDevOps Agent has two complementary detection modes. Both write to the same ` ### What it does -Every `POLL_INTERVAL_MINUTES` minutes, the background poller runs two checks: +Every `POLL_INTERVAL_SECONDS` seconds, the background poller runs two checks: 1. **CloudWatch alarms** — fetches every alarm currently in `ALARM` state; runs a full agent investigation for each new one 2. **Lambda error rates** — checks the last-hour error rate for up to 20 Lambda functions; runs an investigation for any function above `POLL_ERROR_THRESHOLD`% @@ -36,7 +36,7 @@ Every `POLL_INTERVAL_MINUTES` minutes, the background poller runs two checks: ### Flow ``` -Background timer fires every N minutes +Background timer fires every N seconds → get_alarms("ALARM") — any alarm in ALARM state? → list_lambda_functions() — check each function's error rate → error_rate > threshold? @@ -54,16 +54,18 @@ Background timer fires every N minutes ### Configuration ```bash -POLL_INTERVAL_MINUTES=5 # 0 = disabled (default) +POLL_INTERVAL_SECONDS=300 # 0 = disabled (default) POLL_ERROR_THRESHOLD=5.0 # Lambda error rate % to trigger investigation POLL_REINVESTIGATE_HOURS=1 # cooldown — skip re-investigating the same alarm within N hours ``` -Set `POLL_INTERVAL_MINUTES=0` to disable entirely. The poller only starts when this value is greater than zero. +Set `POLL_INTERVAL_SECONDS=0` to disable entirely. The poller only starts when this value is greater than zero. ### Deduplication -An in-memory map (`_last_investigated`) tracks when each alarm or Lambda was last investigated. If the cooldown (`POLL_REINVESTIGATE_HOURS`) hasn't elapsed, the trigger is skipped. This resets on process restart — intentionally, so startup re-checks everything. +The poller builds canonical incident keys and atomically claims them in the database before running the agent. `POLL_REINVESTIGATE_HOURS` controls how long a completed claim blocks another investigation for the same alarm or Lambda metric incident. + +Because claims are durable, autonomous monitoring requires `CHECKPOINT_BACKEND=sqlite` or `postgres`. Memory mode is still useful for chat and quick demos, but the poller and event consumer are disabled there. ### Source @@ -138,10 +140,11 @@ This means you get event-driven Lambda coverage without creating per-function al Go to **Settings → AWS Configuration → Create Infrastructure**. This creates: - SQS queue: `opendevops-agent-events` +- SQS DLQ: `opendevops-agent-events-dlq` - 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. +Wizard and infrastructure state is stored in the database for SQLite/PostgreSQL deployments (`app_config`). Memory mode does not support autonomous monitoring. 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. @@ -175,7 +178,7 @@ SQS is the inbox. SNS is the outbox. You need SQS for event-driven detection; SN | Scenario | Recommendation | |---|---| -| Getting started, no AWS setup | Polling only (`POLL_INTERVAL_MINUTES=5`) | +| Getting started, no AWS setup | Polling only (`POLL_INTERVAL_SECONDS=300`) | | Lambda error monitoring only | Polling — faster and simpler | | ECS, RDS, EC2, GuardDuty coverage | Event-driven required | | Already have CloudWatch alarms | Event-driven — reuses your existing alarms | @@ -186,7 +189,7 @@ SQS is the inbox. SNS is the outbox. You need SQS for event-driven detection; SN ## Testing the Pipeline -**Test polling** — set `POLL_INTERVAL_MINUTES=1`, trigger Lambda errors, wait ~1–2 minutes. +**Test polling** — set `POLL_INTERVAL_SECONDS=60`, trigger Lambda errors, wait ~1–2 minutes. **Test event-driven (fast — bypasses CloudWatch):** ```bash @@ -262,7 +265,7 @@ GET /api/monitoring/services → service health summary | `frontend/src/pages/MonitoringPage.tsx` | Live incident feed UI | | `frontend/src/pages/AlertDetailPage.tsx` | Alert detail + investigate deeplink | | `src/api/routers/monitoring.py` | REST endpoints | -| `src/agent/monitor_store.py` | In-memory service status + alert persistence | +| `src/agent/monitor_store.py` | In-process service status + DB-backed alert, notification, and incident claim persistence | | `src/agent/poller.py` | Proactive polling loop | | `src/agent/event_consumer.py` | SQS long-poll consumer | | `src/agent/event_infra.py` | Create/teardown SQS + EventBridge + alarm | diff --git a/docs/schema.md b/docs/schema.md index 482e8dc..0d4d91c 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -17,6 +17,8 @@ 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 +# Continue through the remaining numbered migrations, including: +psql $DATABASE_URL -f migrations/012_incident_claims.sql ``` All migrations are idempotent (`IF NOT EXISTS`, `IF column does not already exist`). @@ -147,10 +149,26 @@ Persisted event-driven and proactive polling investigation results shown on `/mo | `resolution` | TEXT | Mitigation steps joined as text | | `confidence` | TEXT | `HIGH`, `MEDIUM`, or `LOW` | | `sns_sent` | BOOLEAN | Whether SNS publish succeeded | +| `dedup_key` | TEXT | Canonical incident key used by the poller/event consumer | +| `status` | TEXT | `completed` or `failed` | +| `session_id` | UUID FK → sessions | Investigation session that produced the alert | +| `trigger_source` | TEXT | `poller` or `event_consumer` | | `created_at` | TIMESTAMPTZ | | +### `incident_claims` +Durable pre-investigation claims used to prevent duplicate autonomous agent runs. + +| Column | Type | Notes | +|---|---|---| +| `incident_key` | TEXT PK | Canonical key such as `cloudwatch_alarm:us-east-1:high-error-rate` | +| `trigger_source` | TEXT | `poller` or `event_consumer` | +| `status` | TEXT | `claimed`, `completed`, or `failed` | +| `session_id` | UUID FK → sessions | Set when an investigation completes | +| `claimed_at` | TIMESTAMPTZ | Last time the incident was claimed | +| `completed_at` | TIMESTAMPTZ | Set after a completed or failed investigation | + ### `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. +Application-level key/value configuration shared by all server instances. The init wizard stores its setup and event-infrastructure state under key `init`. | Column | Type | Notes | |---|---|---| diff --git a/docs/slack_and_polling.md b/docs/slack_and_polling.md index 04fffe9..26b2683 100644 --- a/docs/slack_and_polling.md +++ b/docs/slack_and_polling.md @@ -13,7 +13,7 @@ OpenDevOps Agent can send investigation results to a Slack channel in two modes: | | Proactive Polling | Event-Driven (EventBridge → SQS) | |---|---|---| -| **Setup required** | Just set `POLL_INTERVAL_MINUTES` in `.env` | Create AWS infra via Settings → AWS Configuration | +| **Setup required** | Just set `POLL_INTERVAL_SECONDS` in `.env` | Create AWS infra via Settings → AWS Configuration | | **Detection latency** | Up to N minutes (your poll interval) | Near real-time (~seconds after event fires) | | **CloudWatch alarm required** | No — checks Lambda error rates directly | Yes — EventBridge only fires when an alarm trips | | **Works out of the box** | Yes | Requires CloudWatch alarms on your resources | @@ -33,7 +33,7 @@ Add these to your `.env`: SLACK_WEBHOOK_URL=https://hooks.slack.com/services/T.../B.../xxx # Proactive polling (leave at 0 to disable) -POLL_INTERVAL_MINUTES=5 # how often to check; 0 = reactive only +POLL_INTERVAL_SECONDS=300 # how often to check; 0 = reactive only POLL_ERROR_THRESHOLD=5.0 # Lambda error rate % that triggers an investigation POLL_REINVESTIGATE_HOURS=1 # cooldown — don't re-investigate the same alarm within N hours ``` @@ -60,7 +60,7 @@ One investigation = one Slack message. If the agent does not reach a conclusion ## Proactive Mode -When `POLL_INTERVAL_MINUTES > 0`, the app starts a background `asyncio` task (`polling_loop` in `src/agent/poller.py`) on startup. Every N minutes it runs two checks: +When `POLL_INTERVAL_SECONDS > 0`, the app starts a background `asyncio` task (`polling_loop` in `src/agent/poller.py`) on startup. Every N seconds it runs two checks: ### 1. CloudWatch Alarms Calls `get_alarms("ALARM")` and iterates over every alarm currently in ALARM state. For each one: @@ -76,7 +76,7 @@ Calls `list_lambda_functions` (capped at 20 functions) and checks each one's err - Saves the result to the `alerts` table — visible in the `/monitoring` page ### Dedup -An in-memory dict (`_last_investigated`) maps each trigger key (`alarm:` or `lambda_errors:`) to the last time it was investigated. If the cooldown (`POLL_REINVESTIGATE_HOURS`) hasn't elapsed, the alarm or function is skipped. This resets on process restart (intentional — re-checking on startup is fine). +The poller uses canonical incident keys and an atomic database claim before the agent runs. If the cooldown (`POLL_REINVESTIGATE_HOURS`) has not elapsed, the alarm or function is skipped. Autonomous polling requires SQLite or PostgreSQL; memory mode does not start the poller. --- diff --git a/docs/ui.md b/docs/ui.md index 66df26e..5483de0 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 stored in database-backed app config for SQLite/PostgreSQL deployments and mirrored to `data/init.json` as a local cache/fallback. 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. 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. diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index 4f6b53e..67a5f0f 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -29,7 +29,8 @@ export function fmtTok(n?: number): string { } export function relativeTime(isoStr: string): string { - const diff = Date.now() - new Date(isoStr).getTime(); + const normalized = isoStr.includes('Z') || isoStr.includes('+') ? isoStr : isoStr + 'Z'; + const diff = Date.now() - new Date(normalized).getTime(); const mins = Math.floor(diff / 60000); if (mins < 1) return 'just now'; if (mins < 60) return `${mins}m ago`; diff --git a/frontend/src/pages/AlertDetailPage.tsx b/frontend/src/pages/AlertDetailPage.tsx index eb56939..80cc3f3 100644 --- a/frontend/src/pages/AlertDetailPage.tsx +++ b/frontend/src/pages/AlertDetailPage.tsx @@ -108,7 +108,7 @@ export default function AlertDetailPage() { )}
- {new Date(alert.timestamp).toLocaleString()} + {new Date(alert.timestamp.includes('Z') || alert.timestamp.includes('+') ? alert.timestamp : alert.timestamp + 'Z').toLocaleString()} {alert.trigger_source && ( {alert.trigger_source === 'poller' ? '⏱ Poller' : '⚡ Event consumer'} @@ -129,11 +129,32 @@ export default function AlertDetailPage() {
+ {alert.evidence && alert.evidence.length > 0 && ( +
+

Evidence

+
+ {alert.evidence.map((item, i) => ( +
+ • +

{item}

+
+ ))} +
+
+ )} + {alert.resolution && (
-

Resolution

-
-

{alert.resolution}

+

Mitigation Steps

+
+ {alert.resolution.split('\n').filter(Boolean).map((step, i) => ( +
+ + {i + 1} + +

{step.replace(/^[-*]\s*/, '')}

+
+ ))}
)} diff --git a/frontend/src/pages/MonitoringPage.tsx b/frontend/src/pages/MonitoringPage.tsx index 0bad377..6f66c7a 100644 --- a/frontend/src/pages/MonitoringPage.tsx +++ b/frontend/src/pages/MonitoringPage.tsx @@ -7,8 +7,13 @@ import { fetchAlerts, fetchServices, apiFetch } from '../lib/api'; import { useAuth } from '../context/AuthContext'; import type { Alert, ServiceStatus } from '../types'; +function parseUTC(ts: string): Date { + // SQLite omits timezone suffix; append 'Z' so JS treats bare strings as UTC + return new Date(ts.includes('Z') || ts.includes('+') ? ts : ts + 'Z'); +} + function timeAgo(ts: string): string { - const diff = Date.now() - new Date(ts).getTime(); + const diff = Date.now() - parseUTC(ts).getTime(); const mins = Math.floor(diff / 60000); if (mins < 1) return 'just now'; if (mins < 60) return `${mins}m ago`; @@ -200,6 +205,11 @@ export default function MonitoringPage() { )}

{a.error}

+ {a.dedup_key && ( +

+ {a.dedup_key.split(':').pop()} +

+ )}
{timeAgo(a.timestamp)} diff --git a/frontend/src/types.ts b/frontend/src/types.ts index c7793cd..c93d960 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -110,12 +110,14 @@ export interface Alert { service: string; error: string; resolution: string; + evidence: string[]; confidence: 'HIGH' | 'MEDIUM' | 'LOW'; status: 'completed' | 'failed'; sns_sent: boolean; timestamp: string; session_id?: string; trigger_source?: 'poller' | 'event_consumer'; + dedup_key?: string; notifications?: AlertNotification[]; } diff --git a/migrations/012_incident_claims.sql b/migrations/012_incident_claims.sql new file mode 100644 index 0000000..57e79df --- /dev/null +++ b/migrations/012_incident_claims.sql @@ -0,0 +1,12 @@ +-- Atomic incident claims used by poller and event consumer before agent runs +CREATE TABLE IF NOT EXISTS incident_claims ( + incident_key TEXT PRIMARY KEY, + trigger_source TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'claimed', -- claimed | completed | failed + session_id UUID REFERENCES sessions(id) ON DELETE SET NULL, + claimed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS idx_incident_claims_claimed_at ON incident_claims (claimed_at DESC); +CREATE INDEX IF NOT EXISTS idx_incident_claims_status ON incident_claims (status); \ No newline at end of file diff --git a/migrations/013_alerts_evidence.sql b/migrations/013_alerts_evidence.sql new file mode 100644 index 0000000..189f522 --- /dev/null +++ b/migrations/013_alerts_evidence.sql @@ -0,0 +1,2 @@ +-- Add evidence array and expose dedup_key from alerts +ALTER TABLE alerts ADD COLUMN IF NOT EXISTS evidence TEXT NOT NULL DEFAULT '[]'; diff --git a/scripts/test_pipeline.py b/scripts/test_pipeline.py index 8ef619e..2c6eba7 100644 --- a/scripts/test_pipeline.py +++ b/scripts/test_pipeline.py @@ -40,6 +40,7 @@ import datetime import json import os +import sqlite3 import sys import boto3 @@ -68,16 +69,21 @@ def die(msg: str) -> None: def get_queue_url() -> str: - init_path = os.path.join(os.path.dirname(__file__), "..", "data", "init.json") - if os.path.exists(init_path): + if queue_url := os.environ.get("SQS_QUEUE_URL", ""): + return queue_url + + sqlite_path = os.environ.get("SQLITE_PATH", "./data/agent.db") + if os.path.exists(sqlite_path): try: - with open(init_path) as f: - url = json.load(f).get("sqs_queue_url", "") - if url: - return url + with sqlite3.connect(sqlite_path) as conn: + row = conn.execute( + "SELECT value FROM app_config WHERE key = 'init'" + ).fetchone() + if row: + return json.loads(row[0]).get("sqs_queue_url", "") except Exception: pass - return os.environ.get("SQS_QUEUE_URL", "") + return "" def list_functions(lam) -> list[str]: @@ -95,6 +101,7 @@ def run( function_name: str | None, list_only: bool, lambda_only: bool = False, + error_message: str | None = None, ) -> None: try: session = boto3.Session() @@ -250,6 +257,21 @@ def run( }, } + if error_message: + event["source"] = "aws.lambda" + event["detail-type"] = "Lambda Function Invocation Result - Failure" + event["detail"] = { + "requestContext": { + "functionArn": f"arn:aws:lambda:{region}:000000000000:function:{function_name}", + "condition": "RetriesExhausted", + }, + "responsePayload": { + "errorType": "TestError", + "errorMessage": error_message, + }, + } + log(f"Using custom error message: {error_message}") + log("Sending alarm event to SQS manually (bypasses EventBridge)...") try: msg_id = sqs.send_message( @@ -289,8 +311,13 @@ def main() -> None: dest="lambda_only", help="Invoke Lambda and let the CloudWatch alarm fire automatically", ) + parser.add_argument( + "--error-message", + dest="error_message", + help="Custom error message to embed in the SQS event (bypasses alarm format, uses lambda event format)", + ) args = parser.parse_args() - run(args.region, args.invocations, args.function, args.list, args.lambda_only) + run(args.region, args.invocations, args.function, args.list, args.lambda_only, args.error_message) if __name__ == "__main__": diff --git a/src/agent/db/base.py b/src/agent/db/base.py index be79daf..7d01296 100644 --- a/src/agent/db/base.py +++ b/src/agent/db/base.py @@ -141,6 +141,7 @@ async def add_alert( status: str = "completed", session_id: str | None = None, trigger_source: str | None = None, + evidence: list | None = None, ) -> str: return "" @@ -156,6 +157,28 @@ async def add_notification( async def is_recent_alert(self, dedup_key: str, within_minutes: int = 3) -> bool: return False + async def claim_incident( + self, + incident_key: str, + trigger_source: str, + within_minutes: int = 3, + ) -> bool: + return False + + async def complete_incident( + self, + incident_key: str, + status: str = "completed", + session_id: str | None = None, + ) -> None: + pass + + async def release_incident(self, incident_key: str) -> None: + pass + + async def is_incident_claimed(self, incident_key: str, within_minutes: int = 3) -> bool: + return False + async def get_alerts(self, limit: int = 50) -> list[dict]: return [] diff --git a/src/agent/db/memory.py b/src/agent/db/memory.py index ffd8b1f..32b089e 100644 --- a/src/agent/db/memory.py +++ b/src/agent/db/memory.py @@ -4,7 +4,7 @@ import uuid from collections import defaultdict -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import Any from loguru import logger @@ -21,7 +21,10 @@ def __init__(self) -> None: self._messages: dict[str, list[dict]] = defaultdict(list) self._tool_calls: dict[str, list[dict]] = defaultdict(list) self._usage: dict[str, list[dict]] = defaultdict(list) + self._orgs: dict[str, dict] = {} + self._users: dict[str, dict] = {} self._alerts: list[dict] = [] + self._incident_claims: dict[str, dict] = {} self._app_config: dict[str, dict] = {} async def init(self) -> Any: @@ -210,7 +213,10 @@ async def get_dashboard_stats(self) -> dict: sum(u["latency_ms"] for u in all_usage) // len(all_usage) if all_usage else 0 ) - summ_events = [u for u in all_usage if isinstance(u.get("metadata"), dict) and u["metadata"].get("summarization")] + summ_events = [ + u for u in all_usage + if isinstance(u.get("metadata"), dict) and u["metadata"].get("summarization") + ] return { "summary": { @@ -223,7 +229,9 @@ async def get_dashboard_stats(self) -> dict: "total_cost_usd": total_cost, "avg_latency_ms": avg_latency, "total_summarizations": len(summ_events), - "total_chars_compacted": sum(e["metadata"].get("chars_removed", 0) for e in summ_events), + "total_chars_compacted": sum( + e["metadata"].get("chars_removed", 0) for e in summ_events + ), }, "activity": [], "top_tools": [], @@ -241,6 +249,84 @@ async def get_history_stats(self, days: int = 30) -> dict: "trend": [], } + # ── User / auth ───────────────────────────────────────────────────────── + + async def count_users(self) -> int: + return sum(1 for user in self._users.values() if user.get("password_hash")) + + async def get_user_by_email(self, email: str) -> dict | None: + for user in self._users.values(): + if user["email"] == email: + return dict(user) + return None + + async def get_user_by_id(self, user_id: str) -> dict | None: + user = self._users.get(user_id) + return dict(user) if user else None + + async def create_org(self, name: str, slug: str) -> dict | None: + for org in self._orgs.values(): + if org["slug"] == slug: + org["name"] = name + return dict(org) + org_id = str(uuid.uuid4()) + org = {"id": org_id, "name": name, "slug": slug, "created_at": self._now()} + self._orgs[org_id] = org + return dict(org) + + async def get_first_org(self) -> dict | None: + if not self._orgs: + return None + org = sorted(self._orgs.values(), key=lambda item: item["created_at"])[0] + return dict(org) + + async def create_user( + self, + email: str, + name: str, + password_hash: str, + role: str, + org_id: str | None = None, + ) -> dict | None: + if await self.get_user_by_email(email): + return None + user_id = str(uuid.uuid4()) + user = { + "id": user_id, + "email": email, + "name": name, + "password_hash": password_hash, + "role": role, + "org_id": org_id, + "created_at": self._now(), + } + self._users[user_id] = user + return {k: v for k, v in user.items() if k != "password_hash"} + + async def list_users(self) -> list[dict]: + users = sorted(self._users.values(), key=lambda item: item["created_at"]) + return [ + {k: v for k, v in user.items() if k != "password_hash"} + for user in users + ] + + async def update_user(self, user_id: str, **fields: Any) -> dict | None: + user = self._users.get(user_id) + if not user: + return None + for key in ("name", "role", "password_hash"): + if key in fields: + user[key] = fields[key] + return {k: v for k, v in user.items() if k != "password_hash"} + + async def assign_org_to_users_without_org(self, org_id: str) -> None: + for user in self._users.values(): + if not user.get("org_id"): + user["org_id"] = org_id + + async def delete_user(self, user_id: str) -> None: + self._users.pop(user_id, None) + # ── Alerts ──────────────────────────────────────────────────────────────── async def add_alert( @@ -253,6 +339,7 @@ async def add_alert( dedup_key: str | None = None, status: str = "completed", session_id: str | None = None, + trigger_source: str | None = None, ) -> str: alert_id = str(uuid.uuid4()) self._alerts.append({ @@ -266,9 +353,88 @@ async def add_alert( "dedup_key": dedup_key, "status": status, "session_id": session_id, + "trigger_source": trigger_source, + "notifications": [], }) return alert_id + async def add_notification( + self, + alert_id: str, + channel: str, + status: str = "attempted", + error: str | None = None, + ) -> None: + for alert in self._alerts: + if alert["id"] == alert_id: + alert.setdefault("notifications", []).append({ + "channel": channel, + "status": status, + "error": error, + "sent_at": self._now(), + }) + return + + async def is_recent_alert(self, dedup_key: str, within_minutes: int = 3) -> bool: + cutoff = datetime.now(UTC) - timedelta(minutes=within_minutes) + for alert in self._alerts: + if alert.get("dedup_key") != dedup_key: + continue + try: + if datetime.fromisoformat(alert["timestamp"]) > cutoff: + return True + except Exception: + continue + return False + + async def claim_incident( + self, + incident_key: str, + trigger_source: str, + within_minutes: int = 3, + ) -> bool: + now = datetime.now(UTC) + existing = self._incident_claims.get(incident_key) + if existing: + last = existing.get("completed_at") or existing.get("claimed_at") or now + if now - last <= timedelta(minutes=within_minutes): + return False + self._incident_claims[incident_key] = { + "trigger_source": trigger_source, + "status": "claimed", + "session_id": None, + "claimed_at": now, + "completed_at": None, + } + return True + + async def complete_incident( + self, + incident_key: str, + status: str = "completed", + session_id: str | None = None, + ) -> None: + claim = self._incident_claims.setdefault( + incident_key, + {"trigger_source": "unknown", "claimed_at": datetime.now(UTC)}, + ) + claim.update({ + "status": status, + "session_id": session_id, + "completed_at": datetime.now(UTC), + }) + + async def release_incident(self, incident_key: str) -> None: + if self._incident_claims.get(incident_key, {}).get("status") == "claimed": + self._incident_claims.pop(incident_key, None) + + async def is_incident_claimed(self, incident_key: str, within_minutes: int = 3) -> bool: + claim = self._incident_claims.get(incident_key) + if not claim: + return False + last = claim.get("completed_at") or claim.get("claimed_at") + return bool(last and datetime.now(UTC) - last <= timedelta(minutes=within_minutes)) + async def get_alerts(self, limit: int = 50) -> list[dict]: return list(reversed(self._alerts))[:min(limit, 200)] diff --git a/src/agent/db/postgres.py b/src/agent/db/postgres.py index 31b0ac9..d000301 100644 --- a/src/agent/db/postgres.py +++ b/src/agent/db/postgres.py @@ -578,15 +578,18 @@ async def add_alert( status: str = "completed", session_id: str | None = None, trigger_source: str | None = None, + evidence: list | None = None, ) -> str: + import json as _json row = await self._fetchrow( "INSERT INTO alerts" " (service, error, resolution, confidence, sns_sent, dedup_key, status," - " session_id, trigger_source)" - " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id", + " session_id, trigger_source, evidence)" + " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id", service, error, resolution, confidence, sns_sent, dedup_key, status, uuid.UUID(session_id) if session_id else None, trigger_source, + _json.dumps(evidence or []), ) return str(row["id"]) if row else "" @@ -611,10 +614,63 @@ async def is_recent_alert(self, dedup_key: str, within_minutes: int = 3) -> bool ) return row is not None + async def claim_incident( + self, + incident_key: str, + trigger_source: str, + within_minutes: int = 3, + ) -> bool: + row = await self._fetchrow( + "INSERT INTO incident_claims" + " (incident_key, trigger_source, status, claimed_at, completed_at, session_id)" + " VALUES (%s, %s, 'claimed', NOW(), NULL, NULL)" + " ON CONFLICT (incident_key) DO UPDATE SET" + " trigger_source = EXCLUDED.trigger_source, status = 'claimed'," + " claimed_at = NOW(), completed_at = NULL, session_id = NULL" + " WHERE COALESCE(incident_claims.completed_at, incident_claims.claimed_at)" + " < NOW() - (%s * INTERVAL '1 minute')" + " RETURNING incident_key", + incident_key, trigger_source, within_minutes, + ) + return row is not None + + async def complete_incident( + self, + incident_key: str, + status: str = "completed", + session_id: str | None = None, + ) -> None: + await self._exec( + "INSERT INTO incident_claims" + " (incident_key, trigger_source, status, session_id, claimed_at, completed_at)" + " VALUES (%s, 'unknown', %s, %s, NOW(), NOW())" + " ON CONFLICT (incident_key) DO UPDATE SET" + " status = EXCLUDED.status, session_id = EXCLUDED.session_id, completed_at = NOW()", + incident_key, + status, + uuid.UUID(session_id) if session_id else None, + ) + + async def release_incident(self, incident_key: str) -> None: + await self._exec( + "DELETE FROM incident_claims WHERE incident_key = %s AND status = 'claimed'", + incident_key, + ) + + async def is_incident_claimed(self, incident_key: str, within_minutes: int = 3) -> bool: + row = await self._fetchrow( + "SELECT 1 FROM incident_claims WHERE incident_key = %s" + " AND COALESCE(completed_at, claimed_at) > NOW() - (%s * INTERVAL '1 minute')" + " LIMIT 1", + incident_key, within_minutes, + ) + return row is not None + async def get_alerts(self, limit: int = 50) -> list[dict]: + import json as _json rows = await self._fetchall( "SELECT id, service, error, resolution, confidence, sns_sent, status," - " created_at, session_id, trigger_source" + " created_at, session_id, trigger_source, dedup_key, evidence" " FROM alerts ORDER BY created_at DESC LIMIT %s", min(limit, 200), ) @@ -630,14 +686,17 @@ async def get_alerts(self, limit: int = 50) -> list[dict]: "timestamp": r["created_at"].isoformat() if r["created_at"] else None, "session_id": str(r["session_id"]) if r["session_id"] else None, "trigger_source": r["trigger_source"], + "dedup_key": r["dedup_key"], + "evidence": _json.loads(r["evidence"]) if r.get("evidence") else [], } for r in rows ] async def get_alert(self, alert_id: str) -> dict | None: + import json as _json row = await self._fetchrow( "SELECT id, service, error, resolution, confidence, sns_sent, status," - " created_at, session_id, trigger_source" + " created_at, session_id, trigger_source, dedup_key, evidence" " FROM alerts WHERE id = %s", uuid.UUID(alert_id), ) @@ -659,6 +718,8 @@ async def get_alert(self, alert_id: str) -> dict | None: "timestamp": row["created_at"].isoformat() if row["created_at"] else None, "session_id": str(row["session_id"]) if row["session_id"] else None, "trigger_source": row["trigger_source"], + "dedup_key": row["dedup_key"], + "evidence": _json.loads(row["evidence"]) if row.get("evidence") else [], "notifications": [ { "channel": n["channel"], diff --git a/src/agent/db/sqlite.py b/src/agent/db/sqlite.py index e30ad76..aa6dfb9 100644 --- a/src/agent/db/sqlite.py +++ b/src/agent/db/sqlite.py @@ -61,6 +61,26 @@ created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) ); +CREATE TABLE IF NOT EXISTS organizations ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + slug TEXT NOT NULL UNIQUE, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) +); + +CREATE TABLE IF NOT EXISTS users ( + id TEXT PRIMARY KEY, + org_id TEXT REFERENCES organizations(id), + email TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + password_hash TEXT, + role TEXT NOT NULL DEFAULT 'user', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')), + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) +); +CREATE INDEX IF NOT EXISTS users_email_idx ON users(email); + CREATE TABLE IF NOT EXISTS alerts ( id TEXT PRIMARY KEY, service TEXT NOT NULL, @@ -68,10 +88,37 @@ resolution TEXT NOT NULL DEFAULT '', confidence TEXT NOT NULL DEFAULT 'LOW', sns_sent INTEGER NOT NULL DEFAULT 0, + dedup_key TEXT, + status TEXT NOT NULL DEFAULT 'completed', + session_id TEXT, + trigger_source TEXT, + evidence TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) ); 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 INDEX IF NOT EXISTS idx_alerts_dedup_key_created ON alerts(dedup_key, created_at DESC); + +CREATE TABLE IF NOT EXISTS alert_notifications ( + id TEXT PRIMARY KEY, + alert_id TEXT NOT NULL REFERENCES alerts(id) ON DELETE CASCADE, + channel TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'attempted', + error TEXT, + sent_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')) +); +CREATE INDEX IF NOT EXISTS idx_alert_notifications_alert_id ON alert_notifications(alert_id); + +CREATE TABLE IF NOT EXISTS incident_claims ( + incident_key TEXT PRIMARY KEY, + trigger_source TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'claimed', + session_id TEXT, + claimed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')), + completed_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_incident_claims_claimed_at ON incident_claims(claimed_at DESC); +CREATE INDEX IF NOT EXISTS idx_incident_claims_status ON incident_claims(status); CREATE TABLE IF NOT EXISTS app_config ( key TEXT PRIMARY KEY, @@ -164,6 +211,8 @@ async def init(self) -> Any: "ALTER TABLE alerts ADD COLUMN dedup_key TEXT", "ALTER TABLE alerts ADD COLUMN status TEXT NOT NULL DEFAULT 'completed'", "ALTER TABLE alerts ADD COLUMN session_id TEXT", + "ALTER TABLE alerts ADD COLUMN trigger_source TEXT", + "ALTER TABLE alerts ADD COLUMN evidence TEXT NOT NULL DEFAULT '[]'", ): try: await self._conn.execute(col_sql) @@ -171,6 +220,30 @@ async def init(self) -> Any: except Exception: pass # column already exists + for stmt in ( + "CREATE INDEX IF NOT EXISTS idx_alerts_dedup_key_created " + "ON alerts(dedup_key, created_at DESC)", + "CREATE TABLE IF NOT EXISTS alert_notifications (" + "id TEXT PRIMARY KEY, alert_id TEXT NOT NULL REFERENCES alerts(id) ON DELETE CASCADE, " + "channel TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'attempted', error TEXT, " + "sent_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')))", + "CREATE INDEX IF NOT EXISTS idx_alert_notifications_alert_id " + "ON alert_notifications(alert_id)", + "CREATE TABLE IF NOT EXISTS incident_claims (" + "incident_key TEXT PRIMARY KEY, trigger_source TEXT NOT NULL, " + "status TEXT NOT NULL DEFAULT 'claimed', session_id TEXT, " + "claimed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%S', 'now')), " + "completed_at TEXT)", + "CREATE INDEX IF NOT EXISTS idx_incident_claims_claimed_at " + "ON incident_claims(claimed_at DESC)", + "CREATE INDEX IF NOT EXISTS idx_incident_claims_status ON incident_claims(status)", + ): + try: + await self._conn.execute(stmt) + await self._conn.commit() + except Exception: + pass + try: await self._conn.execute( "CREATE TABLE IF NOT EXISTS app_config (" @@ -346,15 +419,18 @@ async def get_messages(self, session_id: str) -> list[dict]: return [] messages = await self._fetchall( - "SELECT id, role, content, created_at FROM messages WHERE session_id = ? ORDER BY created_at ASC", + "SELECT id, role, content, created_at FROM messages " + "WHERE session_id = ? ORDER BY created_at ASC", session_id, ) tool_calls = await self._fetchall( - "SELECT message_id, tool_name, args, result, error FROM tool_calls WHERE session_id = ? ORDER BY created_at ASC", + "SELECT message_id, tool_name, args, result, error FROM tool_calls " + "WHERE session_id = ? ORDER BY created_at ASC", session_id, ) usage_rows = await self._fetchall( - "SELECT message_id, model, input_tokens, output_tokens, cost_usd, latency_ms FROM usage_events WHERE session_id = ?", + "SELECT message_id, model, input_tokens, output_tokens, cost_usd, latency_ms " + "FROM usage_events WHERE session_id = ?", session_id, ) @@ -365,7 +441,11 @@ async def get_messages(self, session_id: str) -> list[dict]: tc_by_msg.setdefault(mid, []).append({ "tool_name": tc["tool_name"], "args": json.loads(tc["args"]) if isinstance(tc["args"], str) else tc["args"], - "result": json.loads(tc["result"]) if isinstance(tc["result"], str) else tc["result"], + "result": ( + json.loads(tc["result"]) + if isinstance(tc["result"], str) + else tc["result"] + ), "error": tc["error"], }) @@ -400,7 +480,8 @@ async def get_messages(self, session_id: str) -> list[dict]: async def delete_session(self, session_id: str) -> None: await self._exec( - "UPDATE sessions SET is_deleted = 1, deleted_at = strftime('%Y-%m-%dT%H:%M:%S', 'now') WHERE id = ?", + "UPDATE sessions SET is_deleted = 1, " + "deleted_at = strftime('%Y-%m-%dT%H:%M:%S', 'now') WHERE id = ?", session_id, ) @@ -441,7 +522,9 @@ async def get_dashboard_stats(self) -> dict: JOIN sessions s ON s.id = ue.session_id WHERE s.is_deleted = 0 AND json_extract(ue.metadata, '$.summarization') = 1) AS total_summarizations, - (SELECT COALESCE(SUM(CAST(json_extract(ue.metadata, '$.chars_removed') AS INTEGER)), 0) + (SELECT COALESCE(SUM(CAST( + json_extract(ue.metadata, '$.chars_removed') AS INTEGER + )), 0) FROM usage_events ue JOIN sessions s ON s.id = ue.session_id WHERE s.is_deleted = 0 AND json_extract(ue.metadata, '$.summarization') = 1) AS total_chars_compacted @@ -648,6 +731,95 @@ async def get_history_stats(self, days: int = 30) -> dict: ], } + # ── User / auth ───────────────────────────────────────────────────────── + + async def count_users(self) -> int: + row = await self._fetchone( + "SELECT COUNT(*) AS n FROM users WHERE password_hash IS NOT NULL" + ) + return int(row["n"]) if row else 0 + + async def get_user_by_email(self, email: str) -> dict | None: + return await self._fetchone("SELECT * FROM users WHERE email = ?", email) + + async def get_user_by_id(self, user_id: str) -> dict | None: + return await self._fetchone("SELECT * FROM users WHERE id = ?", user_id) + + async def create_org(self, name: str, slug: str) -> dict | None: + org_id = str(uuid.uuid4()) + await self._exec( + "INSERT INTO organizations (id, name, slug) VALUES (?, ?, ?) " + "ON CONFLICT(slug) DO UPDATE SET " + "name = excluded.name, updated_at = strftime('%Y-%m-%dT%H:%M:%S', 'now')", + org_id, + name, + slug, + ) + return await self._fetchone( + "SELECT id, name, slug FROM organizations WHERE slug = ?", + slug, + ) + + async def get_first_org(self) -> dict | None: + return await self._fetchone( + "SELECT id, name, slug FROM organizations ORDER BY created_at ASC LIMIT 1" + ) + + async def create_user( + self, + email: str, + name: str, + password_hash: str, + role: str, + org_id: str | None = None, + ) -> dict | None: + user_id = str(uuid.uuid4()) + await self._exec( + "INSERT INTO users (id, email, name, password_hash, role, org_id) " + "VALUES (?, ?, ?, ?, ?, ?)", + user_id, + email, + name, + password_hash, + role, + org_id, + ) + return await self._fetchone( + "SELECT id, email, name, role, created_at FROM users WHERE id = ?", + user_id, + ) + + async def list_users(self) -> list[dict]: + return await self._fetchall( + "SELECT id, email, name, role, created_at FROM users ORDER BY created_at ASC" + ) + + async def update_user(self, user_id: str, **fields: Any) -> dict | None: + allowed = {"name", "role", "password_hash"} + updates = {key: value for key, value in fields.items() if key in allowed} + if not updates: + return await self._fetchone( + "SELECT id, email, name, role, created_at FROM users WHERE id = ?", + user_id, + ) + sets = ", ".join(f"{key} = ?" for key in updates) + await self._exec( + f"UPDATE users SET {sets}, updated_at = strftime('%Y-%m-%dT%H:%M:%S', 'now') " + "WHERE id = ?", + *updates.values(), + user_id, + ) + return await self._fetchone( + "SELECT id, email, name, role, created_at FROM users WHERE id = ?", + user_id, + ) + + async def assign_org_to_users_without_org(self, org_id: str) -> None: + await self._exec("UPDATE users SET org_id = ? WHERE org_id IS NULL", org_id) + + async def delete_user(self, user_id: str) -> None: + await self._exec("DELETE FROM users WHERE id = ?", user_id) + # ── Alerts ──────────────────────────────────────────────────────────────── async def add_alert( @@ -660,20 +832,113 @@ async def add_alert( dedup_key: str | None = None, status: str = "completed", session_id: str | None = None, + trigger_source: str | None = None, + evidence: list | None = None, ) -> str: + import json as _json alert_id = str(uuid.uuid4()) await self._exec( "INSERT INTO alerts" - " (id, service, error, resolution, confidence, sns_sent, dedup_key, status, session_id)" - " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + " (id, service, error, resolution, confidence, sns_sent, dedup_key, status," + " session_id, trigger_source, evidence)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", alert_id, service, error, resolution, confidence, - 1 if sns_sent else 0, dedup_key, status, session_id, + 1 if sns_sent else 0, dedup_key, status, session_id, trigger_source, + _json.dumps(evidence or []), ) return alert_id + async def add_notification( + self, + alert_id: str, + channel: str, + status: str = "attempted", + error: str | None = None, + ) -> None: + await self._exec( + "INSERT INTO alert_notifications (id, alert_id, channel, status, error)" + " VALUES (?, ?, ?, ?, ?)", + str(uuid.uuid4()), alert_id, channel, status, error, + ) + + async def is_recent_alert(self, dedup_key: str, within_minutes: int = 3) -> bool: + row = await self._fetchone( + "SELECT 1 FROM alerts WHERE dedup_key = ? " + "AND julianday(created_at) > julianday('now', ?) LIMIT 1", + dedup_key, + f"-{within_minutes} minutes", + ) + return row is not None + + async def claim_incident( + self, + incident_key: str, + trigger_source: str, + within_minutes: int = 3, + ) -> bool: + if self._conn is None: + return False + try: + cur = await self._conn.execute( + "INSERT OR IGNORE INTO incident_claims " + "(incident_key, trigger_source, status, claimed_at, completed_at, session_id) " + "VALUES (?, ?, 'claimed', strftime('%Y-%m-%dT%H:%M:%S', 'now'), NULL, NULL)", + (incident_key, trigger_source), + ) + await self._conn.commit() + if cur.rowcount == 1: + return True + + cur = await self._conn.execute( + "UPDATE incident_claims SET trigger_source = ?, status = 'claimed', " + "claimed_at = strftime('%Y-%m-%dT%H:%M:%S', 'now'), completed_at = NULL, " + "session_id = NULL WHERE incident_key = ? " + "AND julianday(COALESCE(completed_at, claimed_at)) < julianday('now', ?)", + (trigger_source, incident_key, f"-{within_minutes} minutes"), + ) + await self._conn.commit() + return cur.rowcount > 0 + except Exception as e: + logger.error("SQLite incident claim failed: {}", e) + return False + + async def complete_incident( + self, + incident_key: str, + status: str = "completed", + session_id: str | None = None, + ) -> None: + await self._exec( + "INSERT INTO incident_claims " + "(incident_key, trigger_source, status, session_id, claimed_at, completed_at) " + "VALUES (?, 'unknown', ?, ?, strftime('%Y-%m-%dT%H:%M:%S', 'now'), " + "strftime('%Y-%m-%dT%H:%M:%S', 'now')) " + "ON CONFLICT(incident_key) DO UPDATE SET " + "status = excluded.status, session_id = excluded.session_id, " + "completed_at = excluded.completed_at", + incident_key, status, session_id, + ) + + async def release_incident(self, incident_key: str) -> None: + await self._exec( + "DELETE FROM incident_claims WHERE incident_key = ? AND status = 'claimed'", + incident_key, + ) + + async def is_incident_claimed(self, incident_key: str, within_minutes: int = 3) -> bool: + row = await self._fetchone( + "SELECT 1 FROM incident_claims WHERE incident_key = ? " + "AND julianday(COALESCE(completed_at, claimed_at)) > julianday('now', ?) LIMIT 1", + incident_key, + f"-{within_minutes} minutes", + ) + return row is not None + async def get_alerts(self, limit: int = 50) -> list[dict]: + import json as _json rows = await self._fetchall( - "SELECT id, service, error, resolution, confidence, sns_sent, created_at " + "SELECT id, service, error, resolution, confidence, sns_sent, status, " + " created_at, session_id, trigger_source, dedup_key, evidence " "FROM alerts ORDER BY created_at DESC LIMIT ?", min(limit, 200), ) @@ -685,19 +950,31 @@ async def get_alerts(self, limit: int = 50) -> list[dict]: "resolution": r["resolution"], "confidence": r["confidence"], "sns_sent": bool(r["sns_sent"]), + "status": r.get("status", "completed"), "timestamp": r["created_at"], + "session_id": r.get("session_id"), + "trigger_source": r.get("trigger_source"), + "dedup_key": r.get("dedup_key"), + "evidence": _json.loads(r["evidence"]) if r.get("evidence") else [], } for r in rows ] async def get_alert(self, alert_id: str) -> dict | None: + import json as _json row = await self._fetchone( - "SELECT id, service, error, resolution, confidence, sns_sent, created_at " + "SELECT id, service, error, resolution, confidence, sns_sent, status, " + " created_at, session_id, trigger_source, dedup_key, evidence " "FROM alerts WHERE id = ?", alert_id, ) if not row: return None + notifications = await self._fetchall( + "SELECT channel, status, error, sent_at FROM alert_notifications " + "WHERE alert_id = ? ORDER BY sent_at ASC", + alert_id, + ) return { "id": row["id"], "service": row["service"], @@ -705,7 +982,13 @@ async def get_alert(self, alert_id: str) -> dict | None: "resolution": row["resolution"], "confidence": row["confidence"], "sns_sent": bool(row["sns_sent"]), + "status": row.get("status", "completed"), "timestamp": row["created_at"], + "session_id": row.get("session_id"), + "trigger_source": row.get("trigger_source"), + "dedup_key": row.get("dedup_key"), + "evidence": _json.loads(row["evidence"]) if row.get("evidence") else [], + "notifications": notifications, } async def get_app_config(self, key: str) -> dict | None: diff --git a/src/agent/event_consumer.py b/src/agent/event_consumer.py index 566921f..5a1654c 100644 --- a/src/agent/event_consumer.py +++ b/src/agent/event_consumer.py @@ -3,26 +3,30 @@ from __future__ import annotations import asyncio -import hashlib import json import uuid -from typing import Any +from typing import Any, Literal import boto3 from loguru import logger +from agent.incident_keys import event_incident_key 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, add_notification, is_recent_alert, update_service +from agent.monitor_store import ( + add_alert, + add_notification, + claim_incident, + complete_incident, + release_incident, + update_service, +) from config import settings -# In-memory set of dedup keys currently being investigated. -# Prevents a second identical event from starting a parallel agent run -# before the first one finishes and writes its key to the DB. -_in_progress: set[str] = set() +ProcessResult = Literal["processed", "ignored", "duplicate"] def _build_investigation_prompt(event: dict) -> str: @@ -127,61 +131,71 @@ def _f(obj: Any, key: str, default: Any = None) -> Any: usage_meta: Any = None try: - async for chunk, _meta in get_agent().astream( - {"messages": [{"role": "user", "content": prompt}]}, - config=config, - stream_mode="messages", - ): - um = getattr(chunk, "usage_metadata", None) - if um: - usage_meta = um - - for tcc in getattr(chunk, "tool_call_chunks", []) or []: - idx = _f(tcc, "index", 0) - if idx not in tc_accum: - tc_accum[idx] = {"id": "", "name": "", "args_str": ""} - if tc_id := _f(tcc, "id"): - tc_accum[idx]["id"] = tc_id - if name := (_f(tcc, "name") or ""): - tc_accum[idx]["name"] += name - if args := (_f(tcc, "args") or ""): - tc_accum[idx]["args_str"] += args - - for tc in getattr(chunk, "tool_calls", []) or []: - tc_id = _f(tc, "id") or "" - name = _f(tc, "name") or "" - args = _f(tc, "args") or {} - if tc_id and name: - pending_calls[tc_id] = {"tool": name, "args": args if isinstance(args, dict) else {}} - - content = getattr(chunk, "content", "") - tc_id = getattr(chunk, "tool_call_id", None) - - if content and isinstance(content, str) and not tc_id: - response_text += content - - if tc_id: - for entry in tc_accum.values(): - if eid := entry["id"]: - try: - eargs: Any = json.loads(entry["args_str"]) if entry["args_str"] else {} - except json.JSONDecodeError: - eargs = {} - pending_calls[eid] = {"tool": entry["name"], "args": eargs} - tc_accum.clear() - - call_info = pending_calls.pop( - tc_id, - {"tool": getattr(chunk, "name", None) or "unknown", "args": {}}, - ) - try: - result = json.loads(content) - except (json.JSONDecodeError, TypeError): - result = {"raw": str(content)[:500]} - - tool_calls_log.append({"tool": call_info["tool"], "args": call_info["args"], "result": result}) - if call_info["tool"] == "submit_investigation": - investigation_result = call_info["args"] + async with asyncio.timeout(settings.investigation_timeout): + async for chunk, _meta in get_agent().astream( + {"messages": [{"role": "user", "content": prompt}]}, + config=config, + stream_mode="messages", + ): + um = getattr(chunk, "usage_metadata", None) + if um: + usage_meta = um + + for tcc in getattr(chunk, "tool_call_chunks", []) or []: + idx = _f(tcc, "index", 0) + if idx not in tc_accum: + tc_accum[idx] = {"id": "", "name": "", "args_str": ""} + if tc_id := _f(tcc, "id"): + tc_accum[idx]["id"] = tc_id + if name := (_f(tcc, "name") or ""): + tc_accum[idx]["name"] += name + if args := (_f(tcc, "args") or ""): + tc_accum[idx]["args_str"] += args + + for tc in getattr(chunk, "tool_calls", []) or []: + tc_id = _f(tc, "id") or "" + name = _f(tc, "name") or "" + args = _f(tc, "args") or {} + if tc_id and name: + pending_calls[tc_id] = { + "tool": name, + "args": args if isinstance(args, dict) else {}, + } + + content = getattr(chunk, "content", "") + tc_id = getattr(chunk, "tool_call_id", None) + + if content and isinstance(content, str) and not tc_id: + response_text += content + + if tc_id: + for entry in tc_accum.values(): + if eid := entry["id"]: + try: + eargs: Any = ( + json.loads(entry["args_str"]) if entry["args_str"] else {} + ) + except json.JSONDecodeError: + eargs = {} + pending_calls[eid] = {"tool": entry["name"], "args": eargs} + tc_accum.clear() + + call_info = pending_calls.pop( + tc_id, + {"tool": getattr(chunk, "name", None) or "unknown", "args": {}}, + ) + try: + result = json.loads(content) + except (json.JSONDecodeError, TypeError): + result = {"raw": str(content)[:500]} + + tool_calls_log.append({ + "tool": call_info["tool"], + "args": call_info["args"], + "result": result, + }) + if call_info["tool"] == "submit_investigation": + investigation_result = call_info["args"] except Exception as e: logger.error("Agent investigation failed: {}", e) @@ -195,7 +209,11 @@ def _f(obj: Any, key: str, default: Any = None) -> Any: } # Build readable assistant summary - if investigation_result and investigation_result.get("root_cause_summary") and "_status" not in investigation_result: + if ( + investigation_result + and investigation_result.get("root_cause_summary") + and "_status" not in investigation_result + ): steps = investigation_result.get("mitigation_steps", []) steps_text = "\n".join(f"- {s}" for s in steps) if steps else "_No steps provided._" confidence = investigation_result.get("confidence", "MEDIUM") @@ -226,35 +244,6 @@ def _f(obj: Any, key: str, default: Any = None) -> Any: return investigation_result -_VOLATILE_KEYS = { - "time", "timestamp", "eventTime", "requestId", "eventID", - "approximateInvokeCount", "startTime", "endTime", "updatedAt", - "stateTransitionedAt", "lastUpdatedTime", -} - - -def _strip_volatile(obj: Any) -> Any: - if isinstance(obj, dict): - return {k: _strip_volatile(v) for k, v in obj.items() if k not in _VOLATILE_KEYS} - if isinstance(obj, list): - return [_strip_volatile(i) for i in obj] - return obj - - -def _event_dedup_key(event: dict) -> str: - """MD5 fingerprint of the event's stable fields — works for any AWS service.""" - source = event.get("source", "unknown") - fingerprint = { - "source": source, - "detail-type": event.get("detail-type", ""), - "detail": _strip_volatile(event.get("detail", {})), - } - digest = hashlib.md5( - json.dumps(fingerprint, sort_keys=True, default=str).encode() - ).hexdigest()[:12] - return f"{source}:{digest}" - - def _extract_aws_error(event: dict) -> str: """Pull the original AWS error message out of the event payload.""" source = event.get("source", "") @@ -363,6 +352,7 @@ async def _deliver(result: dict[str, Any], event: dict, dedup_key: str, session_ status=status, session_id=session_id, trigger_source="event_consumer", + evidence=result.get("evidence", []), ) if alert_id: if sns_attempted: @@ -392,7 +382,11 @@ def _is_real_failure(source: str, detail: dict) -> bool: return True -async def _process_event(event: dict) -> None: +def _claim_window_minutes() -> int: + return max(3, (settings.investigation_timeout // 60) + 2) + + +async def _process_event(event: dict) -> ProcessResult: """Filter → dedup check → enrich prompt → investigate → deliver.""" source = event.get("source", "unknown") detail_type = event.get("detail-type", "") @@ -400,20 +394,20 @@ async def _process_event(event: dict) -> None: if not _is_real_failure(source, detail): logger.debug("Skipping non-failure event: {} / {}", source, detail_type) - return + return "ignored" - dedup_key = _event_dedup_key(event) - if dedup_key in _in_progress or await is_recent_alert(dedup_key): + incident_key = event_incident_key(event) + if not await claim_incident(incident_key, "event_consumer", _claim_window_minutes()): logger.info( - "Dedup: skipping {} / {} (key={}) — {}", - source, detail_type, dedup_key, - "investigation already in progress" if dedup_key in _in_progress - else "same event investigated within last 3 min", + "Dedup: skipping {} / {} (incident_key={}) — already claimed recently", + source, detail_type, incident_key, ) - return + return "duplicate" - _in_progress.add(dedup_key) - logger.info("Processing event: {} / {}", source, detail_type) + logger.info("Processing event: {} / {} (incident_key={})", source, detail_type, incident_key) + aws_error = _extract_aws_error(event) + if aws_error: + logger.debug("Event consumer: error message for investigation: {}", aws_error) prompt = _build_investigation_prompt(event) @@ -434,14 +428,22 @@ async def _process_event(event: dict) -> None: session_id = str(uuid.uuid4()) try: - logger.debug("Event consumer: starting investigation for {} / {} (session {})", source, detail_type, session_id[:8]) + logger.debug( + "Event consumer: starting investigation for {} / {} (session {})", + source, detail_type, session_id[:8], + ) result = await _run_investigation(prompt, session_id) if not result: logger.warning("Investigation produced no result for {} / {}", source, detail_type) - return - await _deliver(result, event, dedup_key, session_id) - finally: - _in_progress.discard(dedup_key) + await release_incident(incident_key) + raise RuntimeError("Investigation produced no structured result") + status = result.get("_status", "completed") + await _deliver(result, event, incident_key, session_id) + await complete_incident(incident_key, status, session_id) + return "processed" + except Exception: + await release_incident(incident_key) + raise async def event_consumer_loop() -> None: @@ -472,12 +474,16 @@ async def event_consumer_loop() -> None: ) for msg in resp.get("Messages", []): + should_delete = False try: body = json.loads(msg["Body"]) await _process_event(body) + should_delete = True + except json.JSONDecodeError as e: + logger.error("Failed to decode SQS event body: {}", e) except Exception as e: logger.error("Failed to process event: {}", e) - finally: + if should_delete: await asyncio.get_event_loop().run_in_executor( None, lambda m=msg: sqs.delete_message( diff --git a/src/agent/event_infra.py b/src/agent/event_infra.py index f956f37..1484c7b 100644 --- a/src/agent/event_infra.py +++ b/src/agent/event_infra.py @@ -11,6 +11,8 @@ from config import settings QUEUE_NAME = "opendevops-agent-events" +DLQ_QUEUE_NAME = "opendevops-agent-events-dlq" +MAX_RECEIVE_COUNT = "5" # EventBridge rule definitions: name → event pattern RULES: dict[str, dict] = { @@ -107,16 +109,31 @@ def setup_event_infra(region: str | None = None) -> dict: sts = s.client("sts", region_name=region) account_id = sts.get_caller_identity()["Account"] + dlq_url = "" + dlq_arn = "" queue_url = "" rule_arns: dict[str, str] = {} try: + dlq_resp = sqs.create_queue( + QueueName=DLQ_QUEUE_NAME, + Attributes={"MessageRetentionPeriod": "1209600"}, + ) + dlq_url = dlq_resp["QueueUrl"] + dlq_attrs = sqs.get_queue_attributes(QueueUrl=dlq_url, AttributeNames=["QueueArn"]) + dlq_arn = dlq_attrs["Attributes"]["QueueArn"] + + visibility_timeout = str(max(120, settings.investigation_timeout + 60)) resp = sqs.create_queue( QueueName=QUEUE_NAME, Attributes={ "MessageRetentionPeriod": "86400", - "VisibilityTimeout": "120", + "VisibilityTimeout": visibility_timeout, "ReceiveMessageWaitTimeSeconds": "20", + "RedrivePolicy": json.dumps({ + "deadLetterTargetArn": dlq_arn, + "maxReceiveCount": MAX_RECEIVE_COUNT, + }), }, ) queue_url = resp["QueueUrl"] @@ -172,20 +189,28 @@ def setup_event_infra(region: str | None = None) -> dict: TreatMissingData="notBreaching", ) except Exception: - if queue_url or rule_arns: - teardown_event_infra(queue_url, rule_arns, region) + if queue_url or rule_arns or dlq_url: + teardown_event_infra(queue_url, rule_arns, region, dlq_url=dlq_url) raise logger.info( - "Event infra created: queue={} rules={} alarm={}", queue_url, len(rule_arns), ALARM_NAME + "Event infra created: queue={} dlq={} rules={} alarm={}", + queue_url, dlq_url, len(rule_arns), ALARM_NAME, ) - return {"queue_url": queue_url, "queue_arn": queue_arn, "rule_arns": rule_arns} + return { + "queue_url": queue_url, + "queue_arn": queue_arn, + "dlq_url": dlq_url, + "dlq_arn": dlq_arn, + "rule_arns": rule_arns, + } def teardown_event_infra( queue_url: str | None, rule_arns: dict[str, str] | None, region: str | None = None, + dlq_url: str | None = None, ) -> dict: """Remove EventBridge rules, aggregate CloudWatch alarm, and SQS queue.""" s = _session() @@ -234,5 +259,16 @@ def teardown_event_infra( else: warnings.append("No queue URL was configured") + if dlq_url: + try: + sqs.delete_queue(QueueUrl=dlq_url) + except Exception as e: + message = f"Failed to delete DLQ: {e}" + logger.warning(message) + if _missing_resource(e): + warnings.append(message) + else: + errors.append(message) + logger.info("Event infra torn down") return {"warnings": warnings, "errors": errors} diff --git a/src/agent/incident_keys.py b/src/agent/incident_keys.py new file mode 100644 index 0000000..d6c3f82 --- /dev/null +++ b/src/agent/incident_keys.py @@ -0,0 +1,129 @@ +"""Canonical incident keys shared by poller and event consumer.""" + +from __future__ import annotations + +import hashlib +import json +import re +from typing import Any + +from config import settings + +_VOLATILE_KEYS = { + "time", "timestamp", "eventTime", "requestId", "eventID", + "approximateInvokeCount", "startTime", "endTime", "updatedAt", + "stateTransitionedAt", "lastUpdatedTime", +} + + +def _part(value: Any, fallback: str = "unknown") -> str: + text = str(value or fallback).strip().lower() + text = re.sub(r"[^a-z0-9_.:/-]+", "-", text).strip("-") + return text[:160] or fallback + + +def _digest(value: Any, length: int = 12) -> str: + encoded = json.dumps(value, sort_keys=True, default=str).encode() + return hashlib.sha1(encoded).hexdigest()[:length] + + +def _strip_volatile(obj: Any) -> Any: + if isinstance(obj, dict): + return {k: _strip_volatile(v) for k, v in obj.items() if k not in _VOLATILE_KEYS} + if isinstance(obj, list): + return [_strip_volatile(i) for i in obj] + return obj + + +def _region(value: str | None = None) -> str: + return _part(value or settings.aws_region) + + +def alarm_incident_key(alarm_name: str, region: str | None = None) -> str: + return f"cloudwatch_alarm:{_region(region)}:{_part(alarm_name)}" + + +def lambda_metric_incident_key(function_name: str, region: str | None = None) -> str: + return f"lambda_errors:{_region(region)}:{_part(function_name)}" + + +def lambda_error_incident_key( + function_name: str, + error_signature: Any | None = None, + region: str | None = None, +) -> str: + if error_signature: + return f"lambda_error:{_region(region)}:{_part(function_name)}:{_digest(error_signature)}" + return lambda_metric_incident_key(function_name, region) + + +def event_incident_key(event: dict[str, Any]) -> str: + """Build a stable incident key from the resource identity, not event delivery metadata.""" + source = event.get("source", "unknown") + detail_type = event.get("detail-type", "") + detail = event.get("detail", {}) if isinstance(event.get("detail", {}), dict) else {} + region = event.get("region") or settings.aws_region + + if source == "aws.cloudwatch" and "alarmName" in detail: + return alarm_incident_key(str(detail.get("alarmName", "unknown")), region) + + if source == "aws.lambda": + request_context = detail.get("requestContext", {}) + function_name = ( + detail.get("functionName") + or str(request_context.get("functionArn", "")).split(":")[-1] + or str(detail.get("functionArn", "")).split(":")[-1] + or "unknown" + ) + payload = detail.get("responsePayload", {}) + payload = payload if isinstance(payload, dict) else {} + error_signature = { + "errorType": ( + payload.get("errorType") + or detail.get("errorCode") + or detail.get("condition") + ), + "errorMessage": payload.get("errorMessage") or detail.get("errorMessage"), + "detailType": detail_type, + } + return lambda_error_incident_key(function_name, error_signature, region) + + if source == "aws.ecs": + task_arn = detail.get("taskArn") or detail.get("taskArnStr") or "unknown-task" + cluster = detail.get("clusterArn") or "unknown-cluster" + reason = detail.get("stoppedReason") or detail.get("stopCode") or "stopped" + return ( + f"ecs_task:{_region(region)}:{_part(cluster)}:" + f"{_part(task_arn)}:{_digest(reason, 8)}" + ) + + if source == "aws.rds": + resource = detail.get("SourceIdentifier") or detail.get("SourceArn") or "unknown-db" + categories = detail.get("EventCategories", []) + message = detail.get("Message") or detail.get("EventMessage") or detail_type + signature = {"categories": categories, "message": message} + return f"rds_event:{_region(region)}:{_part(resource)}:{_digest(signature)}" + + if source == "aws.ec2": + instance_id = detail.get("instance-id") or detail.get("instanceId") or "unknown-instance" + state = detail.get("state") or detail_type + return f"ec2_state:{_region(region)}:{_part(instance_id)}:{_part(state)}" + + if source == "aws.codedeploy": + deployment_id = detail.get("deploymentId") or detail.get("deployment-id") or detail + return f"codedeploy:{_region(region)}:{_digest(deployment_id)}" + + if source == "aws.guardduty": + finding_id = detail.get("id") or detail.get("findingId") or detail + return f"guardduty:{_region(region)}:{_digest(finding_id)}" + + if source == "aws.health": + event_arn = detail.get("eventArn") or detail.get("eventTypeCode") or detail + return f"health:{_region(region)}:{_digest(event_arn)}" + + fingerprint = { + "source": source, + "detail-type": detail_type, + "detail": _strip_volatile(detail), + } + return f"event:{_part(source)}:{_digest(fingerprint)}" \ No newline at end of file diff --git a/src/agent/init_store.py b/src/agent/init_store.py index d837163..3c24496 100644 --- a/src/agent/init_store.py +++ b/src/agent/init_store.py @@ -39,6 +39,8 @@ def _default() -> dict: "aws_region": settings.aws_region, "sqs_queue_url": "", "sqs_queue_arn": "", + "sqs_dlq_url": "", + "sqs_dlq_arn": "", "eventbridge_rule_arns": {}, "permissions": { "cloudwatch": None, diff --git a/src/agent/monitor_store.py b/src/agent/monitor_store.py index 9ddc2c6..4006217 100644 --- a/src/agent/monitor_store.py +++ b/src/agent/monitor_store.py @@ -20,6 +20,7 @@ async def add_alert( status: str = "completed", session_id: str | None = None, trigger_source: str | None = None, + evidence: list | None = None, ) -> str: """Persist an alert to the DB backend.""" from agent.db import db @@ -27,7 +28,7 @@ async def add_alert( try: return await db.add_alert( service, error, resolution, confidence, sns_sent, - dedup_key, status, session_id, trigger_source, + dedup_key, status, session_id, trigger_source, evidence, ) except Exception as e: logger.error("Failed to persist alert: {}", e) @@ -59,6 +60,52 @@ async def is_recent_alert(dedup_key: str, within_minutes: int = 3) -> bool: return False +async def claim_incident( + incident_key: str, + trigger_source: str, + within_minutes: int = 3, +) -> bool: + """Atomically claim an incident before running an investigation.""" + from agent.db import db + + try: + return await db.claim_incident(incident_key, trigger_source, within_minutes) + except Exception as e: + logger.error("Failed to claim incident {}: {}", incident_key, e) + return False + + +async def complete_incident( + incident_key: str, + status: str = "completed", + session_id: str | None = None, +) -> None: + from agent.db import db + + try: + await db.complete_incident(incident_key, status, session_id) + except Exception as e: + logger.error("Failed to complete incident {}: {}", incident_key, e) + + +async def release_incident(incident_key: str) -> None: + from agent.db import db + + try: + await db.release_incident(incident_key) + except Exception as e: + logger.error("Failed to release incident {}: {}", incident_key, e) + + +async def is_incident_claimed(incident_key: str, within_minutes: int = 3) -> bool: + from agent.db import db + + try: + return await db.is_incident_claimed(incident_key, within_minutes) + except Exception: + return False + + def update_service(name: str, status: str, error: str | None = None) -> None: _services[name] = { "name": name, diff --git a/src/agent/poller.py b/src/agent/poller.py index 571df18..4997279 100644 --- a/src/agent/poller.py +++ b/src/agent/poller.py @@ -10,27 +10,16 @@ from __future__ import annotations import asyncio -from datetime import UTC, datetime, timedelta from typing import Any from loguru import logger +from agent.incident_keys import alarm_incident_key, lambda_error_incident_key, lambda_metric_incident_key from config import settings -# In-memory dedup: maps a trigger key → datetime last investigated. -# Resets on process restart (intentional — re-check on startup is fine). -_last_investigated: dict[str, datetime] = {} - -def _should_investigate(key: str) -> bool: - last = _last_investigated.get(key) - if last is None: - return True - return datetime.now(UTC) - last > timedelta(hours=settings.poll_reinvestigate_hours) - - -def _mark_investigated(key: str) -> None: - _last_investigated[key] = datetime.now(UTC) +def _claim_window_minutes() -> int: + return max(1, settings.poll_reinvestigate_hours * 60) async def _run_investigation( @@ -58,65 +47,71 @@ def _f(obj: Any, key: str, default: Any = None) -> Any: usage_meta: Any = None try: - async for chunk, _meta in get_agent().astream( - {"messages": [{"role": "user", "content": prompt}]}, - config=config, - stream_mode="messages", - ): - um = getattr(chunk, "usage_metadata", None) - if um: - usage_meta = um - - for tcc in getattr(chunk, "tool_call_chunks", []) or []: - idx = _f(tcc, "index", 0) - if idx not in tc_accum: - tc_accum[idx] = {"id": "", "name": "", "args_str": ""} - if tc_id := _f(tcc, "id"): - tc_accum[idx]["id"] = tc_id - if name := (_f(tcc, "name") or ""): - tc_accum[idx]["name"] += name - if args := (_f(tcc, "args") or ""): - tc_accum[idx]["args_str"] += args - - for tc in getattr(chunk, "tool_calls", []) or []: - tc_id = _f(tc, "id") or "" - name = _f(tc, "name") or "" - args = _f(tc, "args") or {} - if tc_id and name: - pending_calls[tc_id] = { - "tool": name, "args": args if isinstance(args, dict) else {} - } - - content = getattr(chunk, "content", "") - tc_id = getattr(chunk, "tool_call_id", None) - - if content and isinstance(content, str) and not tc_id: - response_text += content - - if tc_id: - for entry in tc_accum.values(): - if eid := entry["id"]: - try: - eargs: Any = json.loads(entry["args_str"]) if entry["args_str"] else {} - except json.JSONDecodeError: - eargs = {} - pending_calls[eid] = {"tool": entry["name"], "args": eargs} - tc_accum.clear() - - call_info = pending_calls.pop( - tc_id, - {"tool": getattr(chunk, "name", None) or "unknown", "args": {}}, - ) - try: - result = json.loads(content) - except (json.JSONDecodeError, TypeError): - result = {"raw": str(content)[:500]} - - tool_calls_log.append({ - "tool": call_info["tool"], "args": call_info["args"], "result": result - }) - if call_info["tool"] == "submit_investigation": - investigation_result = call_info["args"] + async with asyncio.timeout(settings.investigation_timeout): + async for chunk, _meta in get_agent().astream( + {"messages": [{"role": "user", "content": prompt}]}, + config=config, + stream_mode="messages", + ): + um = getattr(chunk, "usage_metadata", None) + if um: + usage_meta = um + + for tcc in getattr(chunk, "tool_call_chunks", []) or []: + idx = _f(tcc, "index", 0) + if idx not in tc_accum: + tc_accum[idx] = {"id": "", "name": "", "args_str": ""} + if tc_id := _f(tcc, "id"): + tc_accum[idx]["id"] = tc_id + if name := (_f(tcc, "name") or ""): + tc_accum[idx]["name"] += name + if args := (_f(tcc, "args") or ""): + tc_accum[idx]["args_str"] += args + + for tc in getattr(chunk, "tool_calls", []) or []: + tc_id = _f(tc, "id") or "" + name = _f(tc, "name") or "" + args = _f(tc, "args") or {} + if tc_id and name: + pending_calls[tc_id] = { + "tool": name, + "args": args if isinstance(args, dict) else {}, + } + + content = getattr(chunk, "content", "") + tc_id = getattr(chunk, "tool_call_id", None) + + if content and isinstance(content, str) and not tc_id: + response_text += content + + if tc_id: + for entry in tc_accum.values(): + if eid := entry["id"]: + try: + eargs: Any = ( + json.loads(entry["args_str"]) if entry["args_str"] else {} + ) + except json.JSONDecodeError: + eargs = {} + pending_calls[eid] = {"tool": entry["name"], "args": eargs} + tc_accum.clear() + + call_info = pending_calls.pop( + tc_id, + {"tool": getattr(chunk, "name", None) or "unknown", "args": {}}, + ) + try: + result = json.loads(content) + except (json.JSONDecodeError, TypeError): + result = {"raw": str(content)[:500]} + + tool_calls_log.append({ + "tool": call_info["tool"], + "args": call_info["args"], + "result": result, + }) + if call_info["tool"] == "submit_investigation": + investigation_result = call_info["args"] except Exception as e: logger.error("Poller investigation failed: {}", e) @@ -215,6 +210,7 @@ async def _persist_and_notify( status=status, session_id=session_id, trigger_source="poller", + evidence=result.get("evidence", []), ) if alert_id: if settings.slack_webhook_url: @@ -227,7 +223,12 @@ async def _check_alarms() -> None: """Check CloudWatch alarms — each new ALARM state triggers an investigation.""" import uuid - from agent.monitor_store import is_recent_alert + from agent.monitor_store import ( + claim_incident, + complete_incident, + is_recent_alert, + release_incident, + ) from tools.cloudwatch import get_alarms data = await asyncio.get_event_loop().run_in_executor(None, get_alarms, "ALARM") @@ -237,37 +238,71 @@ async def _check_alarms() -> None: name = alarm.get("name", "unknown") reason = alarm.get("reason", "") metric = alarm.get("metric", "") - key = f"alarm:{name}" + key = alarm_incident_key(name) - if not _should_investigate(key): - continue if await is_recent_alert(key): logger.debug("Poller: skipping alarm {} — already investigated recently", name) - _mark_investigated(key) + continue + if not await claim_incident(key, "poller", _claim_window_minutes()): continue logger.info("Poller: new ALARM state detected — {}", name) - _mark_investigated(key) - + if alarm.get("state_reason"): + logger.debug("Poller: alarm {} state reason: {}", name, alarm["state_reason"]) + + namespace = alarm.get("namespace", "") + dim_hint = ( + "IMPORTANT: If 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: call list_lambda_functions, then use " + "get_lambda_error_rate on likely functions to find which one(s) have errors. " + "Then pull that function's CloudWatch logs to find the actual error message." + if namespace == "AWS/Lambda" + else f"Namespace: {namespace}" + ) prompt = ( f'CloudWatch alarm "{name}" is in ALARM state.\n' f"Metric: {metric}\n" - f"Reason: {reason}\n" - "Please investigate the root cause and provide mitigation steps." + f"{dim_hint}\n" + f"Reason: {reason}\n\n" + "Investigate step by step:\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" ) session_id = str(uuid.uuid4()) - logger.debug("Poller: starting investigation for alarm {} (session {})", name, session_id[:8]) - result, tool_calls_log = await _run_investigation(prompt, session_id) - if result: - logger.info("Poller: investigation complete for alarm {} — {}", name, result.get("confidence", "?")) - await _persist_and_notify(result, tool_calls_log, session_id, key) + logger.debug( + "Poller: starting investigation for alarm {} (session {})", + name, session_id[:8], + ) + try: + result, tool_calls_log = await _run_investigation(prompt, session_id) + if result: + logger.info( + "Poller: investigation complete for alarm {} — {}", + name, result.get("confidence", "?"), + ) + status = result.get("_status", "completed") + await _persist_and_notify(result, tool_calls_log, session_id, key) + await complete_incident(key, status, session_id) + else: + await release_incident(key) + except Exception: + await release_incident(key) + raise async def _check_lambda_errors() -> None: """Check Lambda functions for error rates above the configured threshold.""" import uuid - from agent.monitor_store import is_recent_alert + from agent.monitor_store import ( + claim_incident, + complete_incident, + is_incident_claimed, + release_incident, + ) from tools.lambda_ import get_lambda_error_rate, list_lambda_functions funcs = await asyncio.get_event_loop().run_in_executor(None, list_lambda_functions) @@ -289,36 +324,53 @@ async def _check_lambda_errors() -> None: if error_rate < settings.poll_error_threshold: continue - key = f"lambda_errors:{name}" - if not _should_investigate(key): - continue - if await is_recent_alert(key): - logger.debug("Poller: skipping Lambda {} — already investigated recently", name) - _mark_investigated(key) - continue # The aggregate alarm covers all Lambda errors — if it fired recently, skip. from agent.event_infra import ALARM_NAME - if await is_recent_alert(f"alarm:{ALARM_NAME}"): + aggregate_key = alarm_incident_key(ALARM_NAME) + aggregate_window = max(3, (settings.investigation_timeout // 60) + 2) + if await is_incident_claimed(aggregate_key, aggregate_window): logger.debug("Poller: skipping Lambda {} — aggregate alarm already handled", name) - _mark_investigated(key) + continue + + error_message = metrics.get("error_message") + error_signature = {"errorMessage": error_message} if error_message else None + key = lambda_error_incident_key(name, error_signature) + if not await claim_incident(key, "poller", _claim_window_minutes()): continue logger.info("Poller: Lambda {} error rate {:.1f}% exceeds threshold", name, error_rate) - _mark_investigated(key) + if error_message: + logger.debug("Poller: Lambda {} error message: {}", name, error_message) prompt = ( f'Lambda function "{name}" has an error rate of {error_rate:.1f}% over the last hour, ' - f"which exceeds the {settings.poll_error_threshold}% threshold.\n" - "Please investigate the root cause and suggest a fix." + f"which exceeds the {settings.poll_error_threshold}% threshold.\n\n" + "Investigate step by step:\n" + "1. Pull the recent CloudWatch logs for this function\n" + "2. Find the actual error message or stack trace\n" + "3. Identify the root cause (code bug, missing dependency, config issue, etc.)\n" + "4. Provide specific, actionable fix steps based on the real error" ) session_id = str(uuid.uuid4()) - logger.debug("Poller: starting investigation for Lambda {} (session {})", name, session_id[:8]) - result, tool_calls_log = await _run_investigation(prompt, session_id) - if result: - logger.info( - "Poller: investigation complete for Lambda {} — {}", name, result.get("confidence", "?") - ) - await _persist_and_notify(result, tool_calls_log, session_id, key) + logger.debug( + "Poller: starting investigation for Lambda {} (session {})", + name, session_id[:8], + ) + try: + result, tool_calls_log = await _run_investigation(prompt, session_id) + if result: + logger.info( + "Poller: investigation complete for Lambda {} — {}", + name, result.get("confidence", "?"), + ) + status = result.get("_status", "completed") + await _persist_and_notify(result, tool_calls_log, session_id, key) + await complete_incident(key, status, session_id) + else: + await release_incident(key) + except Exception: + await release_incident(key) + raise async def polling_loop() -> None: diff --git a/src/api/app.py b/src/api/app.py index 3cee542..1a3edab 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -53,6 +53,15 @@ def start_event_consumer(app_instance: "FastAPI | None" = None) -> None: """Start the SQS event consumer as a background task (idempotent).""" import asyncio + from config import settings as _cfg + + if _cfg.checkpoint_backend == "memory": + logger.warning( + "Event consumer not started: monitoring requires CHECKPOINT_BACKEND=sqlite " + "or postgres for durable incident claims" + ) + return + target = app_instance or app existing: asyncio.Task | None = getattr(target, "_consumer_task", None) if existing and not existing.done(): @@ -94,13 +103,31 @@ async def lifespan(_app: FastAPI): poller_task = None if _cfg.poll_interval_seconds > 0: - from agent.poller import polling_loop + if _cfg.checkpoint_backend == "memory": + logger.warning( + "Proactive poller not started: monitoring requires CHECKPOINT_BACKEND=sqlite " + "or postgres for durable incident claims" + ) + else: + from agent.poller import polling_loop - poller_task = asyncio.create_task(polling_loop()) - logger.info("Proactive poller started (interval={}s)", _cfg.poll_interval_seconds) + poller_task = asyncio.create_task(polling_loop()) + logger.info("Proactive poller started (interval={}s)", _cfg.poll_interval_seconds) # Event consumer — started if explicitly enabled, SQS URL is set, or init wizard completed - if _cfg.event_consumer_enabled or _cfg.sqs_queue_url: + if _cfg.checkpoint_backend == "memory": + try: + from agent.init_store import is_event_infra_enabled + + init_enabled = is_event_infra_enabled() + except Exception: + init_enabled = False + if _cfg.event_consumer_enabled or _cfg.sqs_queue_url or init_enabled: + logger.warning( + "Event consumer disabled: monitoring requires CHECKPOINT_BACKEND=sqlite " + "or postgres" + ) + elif _cfg.event_consumer_enabled or _cfg.sqs_queue_url: start_event_consumer(_app) else: try: diff --git a/src/api/routers/auth.py b/src/api/routers/auth.py index 7da5c0e..71f1803 100644 --- a/src/api/routers/auth.py +++ b/src/api/routers/auth.py @@ -47,7 +47,7 @@ async def register(req: RegisterRequest) -> TokenResponse: if not user: raise HTTPException( status_code=501, - detail="User management requires CHECKPOINT_BACKEND=postgres", + detail="User management is not supported by the selected backend", ) return TokenResponse(access_token=create_access_token(str(user["id"]), role)) diff --git a/src/api/routers/init.py b/src/api/routers/init.py index 58226b2..a9f9024 100644 --- a/src/api/routers/init.py +++ b/src/api/routers/init.py @@ -103,6 +103,8 @@ async def setup( data["event_infra_managed"] = False if data["sqs_queue_url"]: data["sqs_queue_arn"] = "" + data["sqs_dlq_url"] = "" + data["sqs_dlq_arn"] = "" data["eventbridge_rule_arns"] = {} # Create org if name provided and none exists yet @@ -152,6 +154,12 @@ async def skip_services( async def complete( _: Annotated[dict | None, Depends(require_admin)], ): + if settings.checkpoint_backend == "memory": + raise HTTPException( + status_code=400, + detail="Event monitoring requires CHECKPOINT_BACKEND=sqlite or postgres.", + ) + data = await load_init_async() region = data.get("aws_region") or settings.aws_region @@ -161,6 +169,8 @@ async def complete( 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["sqs_dlq_url"] = result.get("dlq_url", "") + data["sqs_dlq_arn"] = result.get("dlq_arn", "") data["eventbridge_rule_arns"] = result["rule_arns"] data["event_infra_enabled"] = True data["event_infra_managed"] = True @@ -193,6 +203,7 @@ async def teardown_infra( """Remove SQS queue and EventBridge rules created by /complete.""" data = await load_init_async() queue_url = data.get("sqs_queue_url") or "" + dlq_url = data.get("sqs_dlq_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")) @@ -231,7 +242,8 @@ async def teardown_infra( from agent.event_infra import teardown_event_infra result = await asyncio.get_event_loop().run_in_executor( - None, teardown_event_infra, queue_url, rule_arns, region + None, + lambda: teardown_event_infra(queue_url, rule_arns, region, dlq_url=dlq_url), ) except Exception as e: logger.error("Teardown failed: {}", e) @@ -243,6 +255,8 @@ async def teardown_infra( data["event_infra_enabled"] = False data["sqs_queue_url"] = "" data["sqs_queue_arn"] = "" + data["sqs_dlq_url"] = "" + data["sqs_dlq_arn"] = "" data["eventbridge_rule_arns"] = {} data["event_infra_managed"] = False await save_init_async(data) diff --git a/src/config/appsettings.py b/src/config/appsettings.py index deb328d..19cf064 100644 --- a/src/config/appsettings.py +++ b/src/config/appsettings.py @@ -72,7 +72,7 @@ class Settings(BaseSettings): event_consumer_enabled: bool = False sqs_queue_url: str | None = None - # Directory for file-based state (init.json) + # Reserved for future file-based state; init config is stored in the DB backend. data_dir: str = "data" diff --git a/tests/integration/test_poller.py b/tests/integration/test_poller.py index 7947f6d..67a3218 100644 --- a/tests/integration/test_poller.py +++ b/tests/integration/test_poller.py @@ -5,20 +5,17 @@ import asyncio from agent import poller +from agent.incident_keys import alarm_incident_key, lambda_metric_incident_key -def test_should_investigate_respects_cooldown(): - poller._last_investigated.clear() - key = "alarm:HighErrorRate" - - assert poller._should_investigate(key) is True - poller._mark_investigated(key) - assert poller._should_investigate(key) is False +def test_claim_window_uses_reinvestigate_hours(monkeypatch): + monkeypatch.setattr(poller.settings, "poll_reinvestigate_hours", 2, raising=False) + assert poller._claim_window_minutes() == 120 def test_check_alarms_dispatches_once(monkeypatch): - poller._last_investigated.clear() - calls = {"run": 0, "persist": 0} + calls = {"run": 0, "persist": 0, "complete": 0} + claimed: set[str] = set() def fake_get_alarms(_state): return { @@ -48,10 +45,30 @@ async def fake_run(prompt, session_id): [], ) - async def fake_persist(_result, _tool_calls_log, _session_id, _dedup_key): + async def fake_persist(_result, _tool_calls_log, _session_id, dedup_key): + assert dedup_key == alarm_incident_key("HighErrorRate-payment-processor") calls["persist"] += 1 + async def fake_recent(_dedup_key, _within_minutes=3): + return False + + async def fake_claim(key, _trigger_source, _within_minutes=3): + if key in claimed: + return False + claimed.add(key) + return True + + async def fake_complete(_key, _status="completed", _session_id=None): + calls["complete"] += 1 + + async def fake_release(_key): + raise AssertionError("successful poller investigation should not release the claim") + monkeypatch.setattr("tools.cloudwatch.get_alarms", fake_get_alarms) + monkeypatch.setattr("agent.monitor_store.is_recent_alert", fake_recent) + monkeypatch.setattr("agent.monitor_store.claim_incident", fake_claim) + monkeypatch.setattr("agent.monitor_store.complete_incident", fake_complete) + monkeypatch.setattr("agent.monitor_store.release_incident", fake_release) monkeypatch.setattr(poller, "_run_investigation", fake_run) monkeypatch.setattr(poller, "_persist_and_notify", fake_persist) @@ -60,11 +77,11 @@ async def fake_persist(_result, _tool_calls_log, _session_id, _dedup_key): assert calls["run"] == 1 assert calls["persist"] == 1 + assert calls["complete"] == 1 def test_check_lambda_errors_dispatches_for_threshold_breach(monkeypatch): - poller._last_investigated.clear() - calls = {"run": 0, "persist": 0} + calls = {"run": 0, "persist": 0, "complete": 0} def fake_list_lambda_functions(): return {"functions": [{"name": "payment-processor"}]} @@ -89,11 +106,29 @@ async def fake_run(prompt, session_id): [], ) - async def fake_persist(_result, _tool_calls_log, _session_id, _dedup_key): + async def fake_persist(_result, _tool_calls_log, _session_id, dedup_key): + assert dedup_key == lambda_metric_incident_key("payment-processor") calls["persist"] += 1 + async def fake_claim(key, _trigger_source, _within_minutes=3): + assert key == lambda_metric_incident_key("payment-processor") + return True + + async def fake_complete(_key, _status="completed", _session_id=None): + calls["complete"] += 1 + + async def fake_release(_key): + raise AssertionError("successful poller investigation should not release the claim") + + async def fake_is_claimed(_key, _within_minutes=3): + return False + monkeypatch.setattr("tools.lambda_.list_lambda_functions", fake_list_lambda_functions) monkeypatch.setattr("tools.lambda_.get_lambda_error_rate", fake_get_lambda_error_rate) + monkeypatch.setattr("agent.monitor_store.claim_incident", fake_claim) + monkeypatch.setattr("agent.monitor_store.complete_incident", fake_complete) + monkeypatch.setattr("agent.monitor_store.release_incident", fake_release) + monkeypatch.setattr("agent.monitor_store.is_incident_claimed", fake_is_claimed) monkeypatch.setattr(poller, "_run_investigation", fake_run) monkeypatch.setattr(poller, "_persist_and_notify", fake_persist) monkeypatch.setattr(poller.settings, "poll_error_threshold", 5.0, raising=False) @@ -102,3 +137,4 @@ async def fake_persist(_result, _tool_calls_log, _session_id, _dedup_key): assert calls["run"] == 1 assert calls["persist"] == 1 + assert calls["complete"] == 1 diff --git a/tests/test_agent/test_incident_keys.py b/tests/test_agent/test_incident_keys.py new file mode 100644 index 0000000..7d0605c --- /dev/null +++ b/tests/test_agent/test_incident_keys.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from agent.incident_keys import event_incident_key, lambda_metric_incident_key + + +def test_lambda_event_keys_split_different_error_signatures(): + base_event = { + "source": "aws.lambda", + "detail-type": "Lambda Function Invocation Result - Failure", + "region": "us-east-1", + "detail": { + "functionName": "payment-processor", + "responsePayload": { + "errorType": "TimeoutError", + "errorMessage": "request timed out after 30s", + }, + }, + } + second_event = { + **base_event, + "detail": { + "functionName": "payment-processor", + "responsePayload": { + "errorType": "ValidationError", + "errorMessage": "missing customer_id", + }, + }, + } + + assert event_incident_key(base_event) != event_incident_key(second_event) + + +def test_lambda_metric_key_is_function_scoped(): + assert ( + lambda_metric_incident_key("payment-processor", "us-east-1") + == "lambda_errors:us-east-1:payment-processor" + ) + + +def test_cloudwatch_alarm_key_ignores_delivery_time(): + event_a = { + "source": "aws.cloudwatch", + "detail-type": "CloudWatch Alarm State Change", + "region": "us-east-1", + "time": "2026-05-20T00:00:00Z", + "detail": {"alarmName": "HighErrorRate"}, + } + event_b = {**event_a, "time": "2026-05-20T00:05:00Z"} + + assert event_incident_key(event_a) == event_incident_key(event_b) \ No newline at end of file diff --git a/tests/test_backends/test_backends.py b/tests/test_backends/test_backends.py index 2d873b5..dcaeffa 100644 --- a/tests/test_backends/test_backends.py +++ b/tests/test_backends/test_backends.py @@ -7,6 +7,8 @@ from __future__ import annotations import os +from collections.abc import AsyncGenerator + import pytest import pytest_asyncio @@ -16,7 +18,7 @@ @pytest_asyncio.fixture(params=["memory", "sqlite"]) -async def backend(request, tmp_path) -> DatabaseBackend: +async def backend(request, tmp_path) -> AsyncGenerator[DatabaseBackend, None]: """Parametrised fixture: runs every test twice — once per local backend.""" if request.param == "memory": from agent.db.memory import MemoryBackend @@ -32,7 +34,7 @@ async def backend(request, tmp_path) -> DatabaseBackend: @pytest_asyncio.fixture -async def pg_backend() -> DatabaseBackend: +async def pg_backend() -> AsyncGenerator[DatabaseBackend, None]: """PostgreSQL backend — skipped unless DATABASE_URL is set.""" url = os.environ.get("DATABASE_URL") if not url: @@ -188,12 +190,62 @@ async def test_save_usage_event_does_not_raise(backend: DatabaseBackend): ) +# ── Monitoring alerts / claims ─────────────────────────────────────────────── + + +async def test_alert_roundtrip_includes_monitoring_fields(backend: DatabaseBackend): + await _seed_session(backend) + alert_id = await backend.add_alert( + service="Lambda / payment-processor", + error="timeout spike", + resolution="increase timeout", + confidence="HIGH", + sns_sent=True, + dedup_key="lambda_errors:us-east-1:payment-processor", + status="completed", + session_id=SESSION_ID, + trigger_source="poller", + ) + await backend.add_notification(alert_id, "slack", "delivered") + + assert await backend.is_recent_alert("lambda_errors:us-east-1:payment-processor") is True + + alerts = await backend.get_alerts() + assert alerts[0]["id"] == alert_id + assert alerts[0]["status"] == "completed" + assert alerts[0]["session_id"] == SESSION_ID + assert alerts[0]["trigger_source"] == "poller" + + detail = await backend.get_alert(alert_id) + assert detail is not None + assert detail["notifications"][0]["channel"] == "slack" + assert detail["notifications"][0]["status"] == "delivered" + + +async def test_incident_claim_blocks_recent_duplicate(backend: DatabaseBackend): + key = "cloudwatch_alarm:us-east-1:high-error-rate" + + assert await backend.claim_incident(key, "poller", within_minutes=60) is True + assert await backend.claim_incident(key, "event_consumer", within_minutes=60) is False + assert await backend.is_incident_claimed(key, within_minutes=60) is True + + await backend.release_incident(key) + assert await backend.claim_incident(key, "event_consumer", within_minutes=60) is True + + await backend.complete_incident(key, "completed", SESSION_ID) + assert await backend.claim_incident(key, "poller", within_minutes=60) is False + + # ── Dashboard stats ─────────────────────────────────────────────────────────── -_DASHBOARD_KEYS = {"summary", "activity", "top_tools", "service_breakdown", "recent_sessions", "root_causes"} -_SUMMARY_KEYS = {"total_sessions", "total_queries", "total_tool_calls", "total_tool_errors", - "total_input_tokens", "total_output_tokens", "total_cost_usd", "avg_latency_ms", - "total_summarizations", "total_chars_compacted"} +_DASHBOARD_KEYS = { + "summary", "activity", "top_tools", "service_breakdown", "recent_sessions", "root_causes", +} +_SUMMARY_KEYS = { + "total_sessions", "total_queries", "total_tool_calls", "total_tool_errors", + "total_input_tokens", "total_output_tokens", "total_cost_usd", "avg_latency_ms", + "total_summarizations", "total_chars_compacted", +} async def test_dashboard_stats_has_required_keys(backend: DatabaseBackend): @@ -226,6 +278,44 @@ async def test_history_stats_has_required_keys(backend: DatabaseBackend): assert stats["days"] == 7 +# ── User / auth ────────────────────────────────────────────────────────────── + + +async def test_user_auth_roundtrip(backend: DatabaseBackend): + org = await backend.create_org("OpenDevOps", "opendevops") + assert org is not None + assert org["slug"] == "opendevops" + + user = await backend.create_user( + "admin@example.com", + "Admin User", + "hashed-password", + "admin", + org["id"], + ) + assert user is not None + assert user["email"] == "admin@example.com" + assert user["role"] == "admin" + assert await backend.count_users() == 1 + + login_user = await backend.get_user_by_email("admin@example.com") + assert login_user is not None + assert login_user["password_hash"] == "hashed-password" + + listed = await backend.list_users() + assert listed[0]["email"] == "admin@example.com" + assert "password_hash" not in listed[0] + + updated = await backend.update_user(user["id"], name="Renamed", role="user") + assert updated is not None + assert updated["name"] == "Renamed" + assert updated["role"] == "user" + + await backend.delete_user(user["id"]) + assert await backend.count_users() == 0 + assert await backend.get_user_by_email("admin@example.com") is None + + async def test_history_stats_empty_by_default(backend: DatabaseBackend): stats = await backend.get_history_stats() assert stats["top_alarms"] == []