diff --git a/README.md b/README.md index 7780ee1..f221fcd 100644 --- a/README.md +++ b/README.md @@ -1,154 +1,246 @@ # void-server -> **Turborepo Monorepo with Dockerized PostgreSQL + Redis, Node.js (Express), and Python (FastAPI)** +> Monorepo for an AI Incident Monitoring and Analysis system. Detects, evaluates, fingerprints, and forms engineering-ready incidents from AI agent execution telemetry, then generates GitHub Issues with root cause analysis. + +**Runtime:** TypeScript (Node.js, Express, BullMQ) + Python (FastAPI, Gemini, PydanticAI) +**Infrastructure:** PostgreSQL 16, Redis 7, Docker Compose --- -## 📁 Monorepo Structure +## Monorepo Structure -``` +```text void-server/ -├── docker-compose.yml # PostgreSQL 16, Redis 7, Adminer -├── turbo.json # Turborepo task runner configuration -├── package.json # npm workspaces root configuration -├── .env.example # Environment variables template +├── docker-compose.yml # PostgreSQL 16, Redis 7, Adminer +├── turbo.json # Turborepo task runner +├── package.json # npm workspaces root │ ├── apps/ -│ ├── node-api/ # Node.js TypeScript + Express API (Port 3001) -│ │ └── src/index.ts # Express server & Prisma database queries +│ ├── node-api/ # Express API (port 3001) + BullMQ worker +│ │ └── src/ +│ │ ├── index.ts # Express server: /health, /api/incidents +│ │ └── worker.ts # BullMQ worker: spawns Python evaluator subprocess │ │ -│ └── fastapi-api/ # Python FastAPI Service (Port 8000) -│ ├── main.py # FastAPI app & endpoints -│ ├── database.py # SQLAlchemy session setup -│ ├── models.py # Incidents & Reports ORM models -│ └── requirements.txt +│ └── fastapi-api/ # FastAPI service (port 8000), SQLAlchemy +│ ├── main.py # /health, /api/incidents, /api/reports +│ ├── database.py # SQLAlchemy session +│ └── models.py # Incident & Report ORM models │ -└── packages/ - ├── db/ # Shared Database Package (Prisma ORM) - │ └── prisma/schema.prisma # Incidents 1:N Reports schema definition - │ - ├── risk-engine/ # Deterministic risk evaluation engine - │ └── src/ - │ ├── evaluator.ts # Orchestrator: 8 policy checks → risk labels - │ ├── types.ts # RiskLabel enum (10 values), Execution, config types - │ └── policies/ # 8 pure policy functions (latency, tokens, crashes...) - │ - ├── incident-fingerprint/ # SHA-256 incident fingerprinting - │ └── src/ - │ ├── risk-labels.ts # normalizeRiskLabels() — validate, dedup, sort - │ ├── incident-fingerprint.ts # generateFingerprint() — SHA-256 hex hash - │ └── types.ts # RiskLabel enum (10 values), Severity type - │ - └── incident-formation/ # Incident persistence + BullMQ queue - └── src/ - ├── service.ts # IncidentFormationService — severity-routed orchestration - ├── repository.ts # PrismaIncidentRepository — CRUD via fingerprint - ├── queue.ts # BullMqIncidentQueue — Redis-backed, stable jobId - └── types.ts # IncidentInput, ProcessResult, interfaces +├── packages/ +│ ├── db/ # Shared Prisma client + schema +│ │ └── prisma/schema.prisma # Incident 1:N Report, JSONB fields +│ │ +│ ├── risk-engine/ # Deterministic risk evaluation (TS) +│ │ └── src/ +│ │ ├── evaluator.ts # Orchestrator: 8 policy checks → labels +│ │ ├── types.ts # RiskLabel enum (10 values) +│ │ └── policies/ # 8 policies (latency, tokens, crashes...) +│ │ +│ ├── incident-fingerprint/ # SHA-256 dedup fingerprinting (TS) +│ │ └── src/ +│ │ ├── risk-labels.ts # normalizeRiskLabels() — dedup, sort, validate +│ │ └── incident-fingerprint.ts # generateFingerprint() +│ │ +│ ├── incident-formation/ # Persistence + BullMQ queue (TS) +│ │ └── src/ +│ │ ├── service.ts # Severity-routed orchestration +│ │ ├── repository.ts # PrismaIncidentRepository +│ │ └── queue.ts # BullMqIncidentQueue +│ │ +│ ├── adaptive-sampling/ # 1-in-N sampling of healthy executions (TS) +│ │ └── src/ +│ │ ├── service.ts # Window-based random sampling +│ │ └── queue.ts # BullMqSamplingQueue +│ │ +│ ├── evaluator/ # AI incident evaluator (Python/Gemini) +│ │ └── src/evaluator/ +│ │ ├── __main__.py # CLI: stdin → JSON +│ │ ├── agent.py # Orchestrator: context → prompt → Gemini → validate → score +│ │ ├── context_builder.py # Parse incident → EvaluationContext +│ │ ├── prompt_builder.py # Build Gemini prompt (v3) +│ │ ├── gemini_client.py # Gemini API call (gemini-3.1-flash-lite) +│ │ ├── validator.py # JSON parse + Pydantic validation +│ │ └── scorer.py # Deterministic confidence adjustments +│ │ +│ └── issue-agent/ # AI GitHub issue generator (Python/PydanticAI) +│ └── src/issue_agent/ +│ ├── __main__.py # CLI: stdin → EngineeringReport / GitHub Issue +│ ├── agent.py # PydanticAI Agent with 4 tools +│ ├── evidence.py # Extract failed tool calls from trace +│ ├── repository.py # GitHubRepo (prod) / LocalRepo (dev) +│ └── schemas.py # IncidentSnapshot, EngineeringReport (18 fields) +│ +└── evaluation_dataset/ # 16 incident scenarios for testing + └── incidents/ + ├── example/ + ├── context-overflow/ + ├── tool-anomaly/ + ├── looping/ + └── ... (16 total) +``` + +--- + +## Pipeline + +```text +Agent Execution Telemetry + │ + ▼ + POST /api/traces + [apps/node-api] + │ + ▼ + Risk Engine (8 policies) + [packages/risk-engine] + │ + ▼ + normalizeRiskLabels() + [packages/incident-fingerprint] + │ + ▼ + IncidentFormationService + [packages/incident-formation] + │ + ┌────┴─────────┐ + │ │ + HEALTHY SUSPICIOUS + │ / CRITICAL + │ │ + ▼ ▼ + Adaptive persist + Sampling + queue + (1-in-N) [incident-analysis] + │ │ + ▼ ▼ + sampling- worker.ts + consumer.ts evaluator + (evaluator) → promotion gate + │ → issue agent + │ │ + ▼ ▼ + promotion? engineering report + │ + GitHub issue + ┌──┴──┐ + no yes + │ │ +skip IncidentFormationService + (SUSPICIOUS) + → incident-analysis queue + → worker.ts (same as above) ``` --- -## 🗄️ Database Schema Design (`Incidents` 1 ➔ N `Reports`) +## What We've Implemented -- **`incidents`**: `id`, `fingerprint` (unique), `trace_id`, `execution_id`, `title`, `severity`, `status`, `confidence`, `first_scene`, `last_scene`, `latest_report_id`, `occurrence`, `last_seen`, `analysis_status`, `latest_labels` (JSONB), `created_at`, `updated_at`. +| Layer | Package | Status | Details | +|---|---|---|---| +| **Risk Detection** | `risk-engine` | ✅ Complete | 8 deterministic policies, severity assignment, 90+ tests | +| **Fingerprinting** | `incident-fingerprint` | ✅ Complete | SHA-256 dedup, label normalization | +| **Incident Formation** | `incident-formation` | ✅ Complete | Severity routing, persistence, BullMQ queue, idempotency | +| **Adaptive Sampling** | `adaptive-sampling` | ✅ Complete | 1-in-N window sampling for healthy executions | +| **Database** | `@void-server/db` | ✅ Complete | Prisma schema, migrations, Incident 1:N Report | +| **Node API** | `node-api` | ✅ Complete | Express server, CRUD endpoints, BullMQ worker → Python subprocess | +| **FastAPI Service** | `fastapi-api` | ✅ Complete | Mirror API for Python-side access | +| **AI Evaluator** | `evaluator` (Python) | ✅ Complete | Gemini-powered failure mode detection, confidence scoring, 16-schema evaluation | +| **Issue Agent** | `issue-agent` (Python) | ✅ Complete | Repository investigation, code graph, timeline reconstruction, GitHub issue generation | +| **Test Scenarios** | `evaluation_dataset` | ✅ Complete | 16 incident scenarios covering all failure modes | +| **Infrastructure** | Docker Compose | ✅ Complete | PostgreSQL 16, Redis 7, Adminer, service Dockerfiles | +| **CI/CD** | turbo.json | ✅ Complete | Build pipeline, workspace orchestration | + +--- + +## Project Stats + +- **Languages:** TypeScript (~5,000 LOC) + Python (~3,500 LOC) +- **Packages:** 9 total (5 TS + 2 Python packages + 2 apps) +- **Tests:** ~180 unit/integration tests across all packages +- **Incident Evaluation Dataset:** 16 scenarios (example, context-overflow, handoff-failure, looping, tool-anomaly, crash-loop, critical-escalation, false-positive, insufficient-evidence, mixed-labels, rate-limit, recurring, transient-error, silent-hallucination, near-perfect, ambiguous-edge-case) +- **AI Models:** Google Gemini 3.1 Flash Lite (evaluator + issue agent) +- **Infrastructure:** PostgreSQL 16 (port 5435), Redis 7 (port 6379) + +--- + +## Database Schema (`Incidents` 1:N `Reports`) + +- **`incidents`**: `id`, `fingerprint` (unique), `trace_id`, `execution_id`, `title`, `severity`, `status`, `confidence`, `first_scene`, `last_scene`, `latest_report_id`, `occurrence`, `last_seen`, `analysis_status` (PENDING/PROCESSING/COMPLETED/FAILED), `latest_labels` (JSONB), `agent_steps` (JSON), `telemetry` (JSON), `engineering_report` (JSONB), `issue_url`, `created_at`, `updated_at`. - **`reports`**: `id`, `incident_id` (FK), `model`, `report` (JSONB), `generated_at`. --- -## 🔄 Pipeline +## Incident Formation Rules -``` -Risk Evaluation Result - ↓ -normalizeRiskLabels() [packages/incident-fingerprint] - ↓ -generateFingerprint() [packages/incident-fingerprint] - ↓ -IncidentFormationService [packages/incident-formation] - ↓ - ┌────┼────┐ - │ │ │ -HEALTHY SUSPICIOUS CRITICAL - │ │ │ - skip persist persist - queue queue -"evaluate-incident" "critical-incident" -``` +| Severity | Persisted? | Queued? | Job Name | +|---|---|---|---| +| HEALTHY | No (sampled 1-in-N) | Sampled to queue | `adaptive-sampling` | +| SUSPICIOUS | Yes | Yes (on creation) | `evaluate-incident` | +| CRITICAL | Yes | Yes (on creation / escalation) | `critical-incident` | --- -## ⚡ Quickstart +## Quickstart ### 1. Configure Environment & Start Infrastructure + ```bash -# Create local environment config cp .env.example .env - -# Start database & Redis infrastructure for local dev npm run db:up ``` -- **PostgreSQL**: `localhost:5435` (User: `void`, Pass: `voidpass`, DB: `void_db`) -- **Redis**: `localhost:6379` -- **Adminer**: [http://localhost:8088](http://localhost:8088) -### 2. Run Database Migrations / Push Schema +- PostgreSQL: `localhost:5435` (User: `void`, Pass: `voidpass`, DB: `void_db`) +- Redis: `localhost:6379` +- Adminer: http://localhost:8088 + +### 2. Database Migrations + ```bash npm run db:push ``` ### 3. Run Development Servers -```bash -# Run all workspaces via Turborepo -npm run dev -# Or run specific workspace -npm run dev --workspace=@void-server/node-api - -# Or run FastAPI Service (Port 8000) -cd apps/fastapi-api && pip install -r requirements.txt && python main.py +```bash +npm run dev # All workspaces via Turborepo +npm run dev --workspace=@void-server/node-api # Node API only (port 3001) +cd apps/fastapi-api && pip install -r requirements.txt && python main.py # FastAPI (port 8000) ``` -### 4. Run Tests +### 4. Run All Tests + ```bash -# All packages +# TypeScript packages (vitest) npm test --workspace=@void-server/risk-engine npm test --workspace=@void-server/incident-fingerprint npm test --workspace=@void-server/incident-formation +npm test --workspace=@void-server/adaptive-sampling + +# Python packages (unittest) — issue agent +PYTHONPATH=packages/issue-agent/src:packages/evaluator/src:packages/issue-agent/tests \ + python3 -m unittest discover -s packages/issue-agent/tests -p "test_*.py" -v ``` ### 5. Issue Agent E2E Monitoring + ```bash # Requires GOOGLE_API_KEY in .env - -# Install dependencies first pip install -e packages/evaluator -e packages/issue-agent +python3 packages/issue-agent/tests/e2e_monitor.py -# Run all 14 scenarios (produces full report per scenario) -packages/evaluator/.venv/bin/python3 packages/issue-agent/tests/e2e_monitor.py - -# Run specific scenario(s) -packages/evaluator/.venv/bin/python3 packages/issue-agent/tests/e2e_monitor.py example tool-anomaly - -# Run issue agent unit tests -PYTHONPATH=packages/issue-agent/src:packages/evaluator/src:packages/issue-agent/tests python3 -m unittest packages/issue-agent/tests/test_schemas.py packages/issue-agent/tests/test_agent.py packages/issue-agent/tests/test_repository.py packages/issue-agent/tests/test_mapper.py -v +# Run specific scenarios +python3 packages/issue-agent/tests/e2e_monitor.py example tool-anomaly ``` ---- - -## 🧩 Packages +### 6. Run Evaluator CLI -| Package | Responsibility | -|---|---| -| `@void-server/risk-engine` | Evaluates executions against 8 deterministic policies → risk labels | -| `@void-server/incident-fingerprint` | Normalizes labels, generates SHA-256 fingerprint | -| `@void-server/incident-formation` | Persists incidents, queues analysis via BullMQ | -| `@void-server/db` | Prisma client, shared database types | +```bash +echo '{"id":"test","execution_id":"test","trace_id":"test","severity":"SUSPICIOUS","status":"OPEN","confidence":0,"occurrence":1,"analysis_status":"PENDING","agent_steps":[],"telemetry":null}' | \ + PYTHONPATH=packages/evaluator/src python3 -m evaluator +``` -### Incident Formation Rules +### 7. Run Issue Agent CLI (Dev Mode) -| Severity | Persisted? | Queued? | Job Name | -|---|---|---|---| -| HEALTHY | No | No | — | -| SUSPICIOUS | Yes | Yes (on creation) | `evaluate-incident` | -| CRITICAL | Yes | Yes (on creation / escalation) | `critical-incident` | +```bash +cat evaluation_dataset/incidents/example/incident.json | \ + PYTHONPATH=packages/issue-agent/src VOID_DEV_MODE=1 python3 -m issue_agent +``` diff --git a/apps/node-api/package.json b/apps/node-api/package.json index 19718be..2379750 100644 --- a/apps/node-api/package.json +++ b/apps/node-api/package.json @@ -4,12 +4,17 @@ "main": "src/index.ts", "scripts": { "dev": "tsx watch src/index.ts", - "dev:worker": "PYTHONPATH=../../packages/evaluator/src tsx src/worker.ts", + "dev:worker": "PYTHONPATH=../../packages/evaluator/src:../../packages/issue-agent/src tsx src/worker.ts", + "dev:sampling": "PYTHONPATH=../../packages/evaluator/src tsx src/sampling-consumer.ts", "build": "tsc", "start": "node dist/index.js" }, "dependencies": { + "@void-server/adaptive-sampling": "*", "@void-server/db": "*", + "@void-server/incident-fingerprint": "*", + "@void-server/incident-formation": "*", + "@void-server/risk-engine": "*", "bullmq": "^5.0.0", "cors": "^2.8.5", "express": "^4.19.2" diff --git a/apps/node-api/src/WIRING.md b/apps/node-api/src/WIRING.md new file mode 100644 index 0000000..6d6422d --- /dev/null +++ b/apps/node-api/src/WIRING.md @@ -0,0 +1,327 @@ +# SDK Ingestion & Pipeline Wiring + +## Architecture + +Two entry points converge into the same incident pipeline: + +``` +SDK Application + │ POST /api/traces (agent steps + telemetry) + ▼ +/api/traces + │ + ├── risk-engine.evaluate() + │ └── severity + labels + │ + ├── HEALTHY ──▶ adaptive-sampling (1/N queue) + │ + └── SUSPICIOUS/CRITICAL ──▶ IncidentFormationService + │ + ├── DB upsert (create/update incident) + └── queue "incident-analysis" + │ + ▼ + worker.ts + ├── evaluator (Python LLM) + ├── promotion gate + └── issue agent (Python LLM) +``` + +## Entry point — `POST /api/traces` (`apps/node-api/src/index.ts`) + +### Imports + service singletons (lines 1-20) + +```typescript +import express from 'express'; +import cors from 'cors'; +import { db } from './db'; +import { evaluate } from "@void-server/risk-engine"; +import { config as defaultRiskConfig } from "@void-server/risk-engine"; +import { normalizeRiskLabels } from "@void-server/incident-fingerprint"; +import { IncidentFormationService, PrismaIncidentRepository, BullMqIncidentQueue } from "@void-server/incident-formation"; +import { AdaptiveSamplingService, BullMqSamplingQueue } from "@void-server/adaptive-sampling"; + +const app = express(); +const PORT = process.env.NODE_API_PORT || 3001; + +app.use(cors()); +app.use(express.json()); + +const repo = new PrismaIncidentRepository(db); +const queue = new BullMqIncidentQueue(); +const formationService = new IncidentFormationService(repo, queue); +const samplingQueue = new BullMqSamplingQueue(); +const sampler = new AdaptiveSamplingService(samplingQueue, {}); +``` + +### Endpoint (lines 92-167) + +```typescript +app.post("/api/traces", async (req, res) => { + try { + const body = req.body; + if (!body.execution_id) { + return res.status(400).json({ error: "execution_id is required" }); + } + + // Build risk-engine Execution from incoming telemetry + const execution = { + latencyMs: body.total_latency_ms ?? 0, + promptTokens: body.total_prompt_tokens ?? 0, + completionTokens: body.total_completion_tokens ?? 0, + toolExecutions: (body.steps ?? []).map((s: any) => ({ + toolName: s.tool_name ?? "unknown", + success: s.success ?? true, + })), + retryCount: body.retry_count ?? 0, + hasFinalResponse: body.crashed !== true, + crashed: body.crashed ?? false, + contextWindowExceeded: body.context_window_exceeded ?? false, + }; + + const risk = evaluate(execution, defaultRiskConfig); + const labels = normalizeRiskLabels({ + severity: risk.severity as any, + labels: risk.labels, + }); + const timestamp = new Date(); + const agentSteps = body.steps ?? []; + const telemetry = { + total_latency_ms: body.total_latency_ms, + total_prompt_tokens: body.total_prompt_tokens, + total_completion_tokens: body.total_completion_tokens, + tool_call_count: agentSteps.length, + failed_tool_calls: agentSteps.filter((s: any) => !s.success).length, + retry_count: body.retry_count, + }; + + // HEALTHY → adaptive sampling + if (risk.severity === "HEALTHY") { + const sampled = await sampler.process({ + executionId: body.execution_id, + traceId: body.trace_id, + timestamp, + agentSteps, + telemetry, + }); + return res.status(200).json({ + status: "healthy", + execution_id: body.execution_id, + severity: "HEALTHY", + labels, + sampled, + }); + } + + // SUSPICIOUS/CRITICAL → incident formation (creates incident + enqueues) + const result = await formationService.process({ + severity: risk.severity, + labels, + executionId: body.execution_id, + traceId: body.trace_id, + timestamp, + agent_steps: agentSteps, + telemetry, + }); + + const incidentId = result.action !== "SKIPPED" ? result.incident.id : undefined; + + return res.status(result.action === "CREATED" ? 201 : 200).json({ + status: result.action === "CREATED" ? "incident_created" : "incident_updated", + execution_id: body.execution_id, + severity: risk.severity, + labels, + incident_id: incidentId, + }); + + } catch (err) { + console.error("[traces] error:", err); + return res.status(500).json({ error: "Internal server error" }); + } +}); +``` + +**Key points:** +- `fingerprint` is NOT generated in the endpoint — `IncidentFormationService` owns it +- `normalizeRiskLabels` receives severity cast to avoid type mismatch between risk-engine's `"HEALTHY"` value and incident-fingerprint's `Severity` type +- Response is immediate — client does NOT wait for evaluator or issue agent + +--- + +## IncidentFormationService — fingerprint generation (`packages/incident-formation/src/service.ts`) + +`IncidentInput.fingerprint` is now optional. If absent, the service computes it from labels: + +```typescript +import { generateFingerprint } from "@void-server/incident-fingerprint"; + +async process(input: IncidentInput): Promise { + if (input.severity === "HEALTHY") { + return { action: "SKIPPED" }; + } + + const fingerprint = input.fingerprint ?? generateFingerprint(input.labels); + const existing = await this.repo.findByFingerprint(fingerprint); + // ... + created = await this.repo.create({ + fingerprint, // ← uses local var, not input.fingerprint + // ... + }); +} +``` + +--- + +## Adaptive sampling — queue payload (`packages/adaptive-sampling/src/queue.ts`) + +The enqueue method now includes `agentSteps` and `telemetry` so the sampling consumer has the trace data: + +```typescript +async enqueue(sample: SamplingInput): Promise { + const payload = { + executionId: sample.executionId, + traceId: sample.traceId, + timestamp: sample.timestamp.toISOString(), + agentSteps: sample.agentSteps, + telemetry: sample.telemetry, + }; + await this.queue.add("sample", payload, { + removeOnComplete: { count: 1000 }, + removeOnFail: { count: 1000 }, + }); +} +``` + +--- + +## Sampling consumer — promotion via IncidentFormationService (`apps/node-api/src/sampling-consumer.ts`) + +After the evaluator finds a REAL_INCIDENT, fingerprint is computed from evaluator failure_modes (not `sampled:{executionId}`), then piped through `IncidentFormationService`: + +```typescript +import { IncidentFormationService, PrismaIncidentRepository, BullMqIncidentQueue } from "@void-server/incident-formation"; +import { createHash } from "node:crypto"; + +const formationService = new IncidentFormationService( + new PrismaIncidentRepository(db), + new BullMqIncidentQueue() +); + +// ... evaluator runs, gets classification + failureModes ... + +if (!promoted) { + console.log(`[sampling] skipped ${sample.executionId}: ${classification} (confidence: ${confidence})`); + return; +} + +// Fingerprint from evaluator failure modes → same failure pattern = same fingerprint +const fingerprint = createHash("sha256") + .update(failureModes.sort().join("|")) + .digest("hex"); + +await formationService.process({ + fingerprint, + severity: "SUSPICIOUS", + labels: [], + executionId: sample.executionId, + traceId: sample.traceId, + timestamp: new Date(sample.timestamp), + agent_steps: sample.agentSteps, + telemetry: sample.telemetry, +}); +``` + +**This unifies both paths:** +- Deterministic SUSPICIOUS/CRITICAL → `IncidentFormationService` generates fingerprint from risk labels +- Promoted samples → explicit fingerprint from evaluator failure_modes, same service + +Both converge through `incident-analysis` queue into the same worker. + +--- + +## Request body (SDK-facing contract) + +```json +{ + "execution_id": "string (required)", + "trace_id": "string (optional)", + "agent": { "name": "string", "role": "string" }, + "model": "string", + "steps": [ + { + "tool_name": "string", + "input": {}, + "output": "string", + "success": true, + "latency_ms": 123, + "error": null + } + ], + "total_latency_ms": 5000, + "total_prompt_tokens": 1000, + "total_completion_tokens": 2000, + "retry_count": 0, + "crashed": false, + "context_window_exceeded": false +} +``` + +Only `execution_id` is required. Everything else defaults safely to 0 / false / []. + +--- + +## Response shapes + +```json +// HEALTHY (not sampled) +{ "status": "healthy", "execution_id": "...", "severity": "HEALTHY", "labels": [], "sampled": false } + +// HEALTHY (sampled — 1/N hit) +{ "status": "healthy", "execution_id": "...", "severity": "HEALTHY", "labels": [], "sampled": true } + +// SUSPICIOUS (new incident) +{ "status": "incident_created", "execution_id": "...", "severity": "SUSPICIOUS", "labels": ["HIGH_LATENCY"], "incident_id": "uuid" } + +// CRITICAL (duplicate — updated occurrence) +{ "status": "incident_updated", "execution_id": "...", "severity": "CRITICAL", "labels": ["TOOL_FAILURE"], "incident_id": "uuid" } + +// Missing execution_id +400 { "error": "execution_id is required" } +``` + +--- + +## Environment variables + +| Variable | Default | Used by | +|---|---|---| +| `REDIS_URL` | `redis://localhost:6379` | Both workers + endpoint | +| `PROMOTION_CONFIDENCE_THRESHOLD` | `0.7` | Sampling consumer | + +--- + +## Running + +```sh +# API server (includes /api/traces endpoint) +npm run dev --workspace=@void-server/node-api + +# Incident analysis worker (evaluator + issue agent) +npm run dev:worker --workspace=@void-server/node-api + +# Adaptive sampling consumer +npm run dev:sampling --workspace=@void-server/node-api +``` + +--- + +## Files changed + +| File | Change | +|---|---| +| `apps/node-api/package.json` | Added 4 workspace dependencies | +| `apps/node-api/src/index.ts` | Added `POST /api/traces` endpoint | +| `packages/incident-formation/src/types.ts` | `fingerprint` optional in `IncidentInput` | +| `packages/incident-formation/src/service.ts` | Generate fingerprint from labels if absent | +| `packages/adaptive-sampling/src/queue.ts` | Include `agentSteps` + `telemetry` in payload | +| `apps/node-api/src/sampling-consumer.ts` | Replace `db.incident.upsert` with `formationService.process()` | \ No newline at end of file diff --git a/apps/node-api/src/index.ts b/apps/node-api/src/index.ts index b02c0a2..131c1b0 100644 --- a/apps/node-api/src/index.ts +++ b/apps/node-api/src/index.ts @@ -1,6 +1,11 @@ import express from 'express'; import cors from 'cors'; import { db } from './db'; +import { evaluate } from "@void-server/risk-engine"; +import { config as defaultRiskConfig } from "@void-server/risk-engine"; +import { normalizeRiskLabels } from "@void-server/incident-fingerprint"; +import { IncidentFormationService, PrismaIncidentRepository, BullMqIncidentQueue } from "@void-server/incident-formation"; +import { AdaptiveSamplingService, BullMqSamplingQueue } from "@void-server/adaptive-sampling"; const app = express(); const PORT = process.env.NODE_API_PORT || 3001; @@ -8,6 +13,12 @@ const PORT = process.env.NODE_API_PORT || 3001; app.use(cors()); app.use(express.json()); +const repo = new PrismaIncidentRepository(db); +const queue = new BullMqIncidentQueue(); +const formationService = new IncidentFormationService(repo, queue); +const samplingQueue = new BullMqSamplingQueue(); +const sampler = new AdaptiveSamplingService(samplingQueue, {}); + // Healthcheck endpoint verifying PostgreSQL database connection app.get('/health', async (_req, res) => { try { @@ -78,6 +89,83 @@ app.post('/api/incidents', async (req, res) => { } }); +app.post("/api/traces", async (req, res) => { + try { + const body = req.body; + if (!body.execution_id) { + return res.status(400).json({ error: "execution_id is required" }); + } + + const execution = { + latencyMs: body.total_latency_ms ?? 0, + promptTokens: body.total_prompt_tokens ?? 0, + completionTokens: body.total_completion_tokens ?? 0, + toolExecutions: (body.steps ?? []).map((s: any) => ({ + toolName: s.tool_name ?? "unknown", + success: s.success ?? true, + })), + retryCount: body.retry_count ?? 0, + hasFinalResponse: body.crashed !== true, + crashed: body.crashed ?? false, + contextWindowExceeded: body.context_window_exceeded ?? false, + }; + + const risk = evaluate(execution, defaultRiskConfig); + const labels = normalizeRiskLabels({ severity: risk.severity as any, labels: risk.labels }); + const timestamp = new Date(); + const agentSteps = body.steps ?? []; + const telemetry = { + total_latency_ms: body.total_latency_ms, + total_prompt_tokens: body.total_prompt_tokens, + total_completion_tokens: body.total_completion_tokens, + tool_call_count: agentSteps.length, + failed_tool_calls: agentSteps.filter((s: any) => !s.success).length, + retry_count: body.retry_count, + }; + + if (risk.severity === "HEALTHY") { + const sampled = await sampler.process({ + executionId: body.execution_id, + traceId: body.trace_id, + timestamp, + agentSteps, + telemetry, + }); + return res.status(200).json({ + status: "healthy", + execution_id: body.execution_id, + severity: "HEALTHY", + labels, + sampled, + }); + } + + const result = await formationService.process({ + severity: risk.severity, + labels, + executionId: body.execution_id, + traceId: body.trace_id, + timestamp, + agent_steps: agentSteps, + telemetry, + }); + + const incidentId = result.action !== "SKIPPED" ? result.incident.id : undefined; + + return res.status(result.action === "CREATED" ? 201 : 200).json({ + status: result.action === "CREATED" ? "incident_created" : "incident_updated", + execution_id: body.execution_id, + severity: risk.severity, + labels, + incident_id: incidentId, + }); + + } catch (err) { + console.error("[traces] error:", err); + return res.status(500).json({ error: "Internal server error" }); + } +}); + app.listen(PORT, () => { console.log(`🚀 Node.js API running at http://localhost:${PORT}`); }); diff --git a/apps/node-api/src/python.ts b/apps/node-api/src/python.ts new file mode 100644 index 0000000..e25dbe2 --- /dev/null +++ b/apps/node-api/src/python.ts @@ -0,0 +1,37 @@ +import { execFile } from "child_process"; + +const PYTHON = process.env.EVALUATOR_PYTHON ?? "python3"; +const PYTHONPATH = process.env.EVALUATOR_PYTHONPATH ?? ""; +const DEFAULT_MAX_BUFFER = 10485760; +const parsedMaxBuffer = parseInt(process.env.EVALUATOR_MAX_BUFFER ?? "", 10); +const EVALUATOR_MAX_BUFFER = + Number.isFinite(parsedMaxBuffer) && parsedMaxBuffer > 0 + ? parsedMaxBuffer + : DEFAULT_MAX_BUFFER; + + +export function runPythonModule(module: string, inputJson: string, timeoutMs: number): Promise { + return new Promise((resolve, reject) => { + const child = execFile( + PYTHON, + ["-m", module], + { + env: { + ...process.env, + ...(PYTHONPATH ? { PYTHONPATH } : {}), + }, + maxBuffer: EVALUATOR_MAX_BUFFER, + timeout: timeoutMs, + }, + (err, stdout, stderr) => { + if (err) { + reject(new Error(`${module} exited: ${err.message}\nStderr: ${stderr}`)); + return; + } + resolve(stdout.trim()); + }, + ); + child.stdin?.write(inputJson); + child.stdin?.end(); + }); +} \ No newline at end of file diff --git a/apps/node-api/src/sampling-consumer.ts b/apps/node-api/src/sampling-consumer.ts new file mode 100644 index 0000000..ae7ac03 --- /dev/null +++ b/apps/node-api/src/sampling-consumer.ts @@ -0,0 +1,135 @@ +import { Worker, Job } from "bullmq"; +import { runPythonModule } from "./python"; +import { db } from "./db"; +import { IncidentFormationService, PrismaIncidentRepository, BullMqIncidentQueue } from "@void-server/incident-formation"; +import { hashJoin } from "@void-server/incident-fingerprint"; +import { shouldPromoteToIssueAgent } from "./worker"; + +const REDIS_URL = process.env.REDIS_URL ?? "redis://localhost:6379"; +const EVALUATOR_TIMEOUT_MS = parseInt(process.env.EVALUATOR_TIMEOUT_MS ?? "120000", 10); +const EVALUATOR_MODULE = process.env.EVALUATOR_MODULE ?? "evaluator"; +const PROMOTION_CONFIDENCE_THRESHOLD = parseFloat(process.env.PROMOTION_CONFIDENCE_THRESHOLD ?? "0.7"); + +const formationService = new IncidentFormationService( + new PrismaIncidentRepository(db), + new BullMqIncidentQueue() +); + +function parseRedisUrl(urlStr: string) { + const parsed = new URL(urlStr); + if (!parsed.hostname) throw new Error(`Invalid REDIS_URL: no hostname in "${urlStr}"`); + const config: Record = { + host: parsed.hostname, + port: parsed.port ? parseInt(parsed.port, 10) : 6379, + }; + if (parsed.username) config.username = decodeURIComponent(parsed.username); + if (parsed.password) config.password = decodeURIComponent(parsed.password); + const dbIndex = parsed.pathname ? parseInt(parsed.pathname.replace("/", ""), 10) : NaN; + if (!isNaN(dbIndex)) config.db = dbIndex; + if (parsed.protocol === "rediss:") config.tls = {}; + return config; +} + +async function processSample(job: Job) { + const sample = job.data as { + executionId: string; + traceId?: string; + timestamp: string; + agentSteps?: unknown[]; + telemetry?: Record; + }; + console.log(`[sampling] evaluating sampled execution ${sample.executionId}`); + + const input = { + id: sample.executionId, + execution_id: sample.executionId, + trace_id: sample.traceId ?? "", + severity: "HEALTHY", + status: "OPEN", + confidence: 0, + occurrence: 1, + analysis_status: "PENDING", + agent_steps: sample.agentSteps ?? [], + telemetry: sample.telemetry ?? null, + }; + + let raw: string; + try { + raw = await runPythonModule(EVALUATOR_MODULE, JSON.stringify(input), EVALUATOR_TIMEOUT_MS); + } catch (err) { + console.error(`[sampling] evaluator failed for ${sample.executionId}:`, (err as Error).message); + return; + } + + let parsed: Record; + try { + parsed = JSON.parse(raw); + } catch { + console.error(`[sampling] invalid evaluator output for ${sample.executionId}`); + return; + } + + if (parsed.error) { + console.error(`[sampling] evaluator error for ${sample.executionId}: ${parsed.error}`); + return; + } + + const evaluation = parsed.evaluation as Record | undefined; + if (!evaluation) { + console.log(`[sampling] no evaluation for ${sample.executionId}, skipping`); + return; + } + + const classification = String(evaluation.classification ?? ""); + const confidence = Number(evaluation.confidence ?? 0); + const failureModes = (evaluation.failure_modes as string[]) ?? []; + + const promoted = shouldPromoteToIssueAgent("evaluate-incident", classification, confidence, failureModes); + + if (!promoted) { + console.log(`[sampling] skipped ${sample.executionId}: ${classification} (confidence: ${confidence})`); + return; + } + + console.log(`[sampling] promoting ${sample.executionId} to incident`); + + const fingerprint = hashJoin(failureModes); + + await formationService.process({ + fingerprint, + severity: "SUSPICIOUS", + labels: [], + executionId: sample.executionId, + traceId: sample.traceId, + timestamp: new Date(sample.timestamp), + agent_steps: sample.agentSteps, + telemetry: sample.telemetry, + }); + + console.log(`[sampling] promoted ${sample.executionId} to incident pipeline`); +} + +const connection = parseRedisUrl(REDIS_URL); +const worker = new Worker("adaptive-sampling", processSample, { + connection, + concurrency: 1, + autorun: true, +}); + +worker.on("completed", (job) => { + console.log(`[sampling] job ${job?.id} completed`); +}); + +worker.on("failed", (job, err) => { + console.error(`[sampling] job ${job?.id} failed:`, err.message); +}); + +async function shutdown() { + console.log("[sampling] shutting down..."); + await worker.close(); + process.exit(0); +} +process.on("SIGTERM", shutdown); +process.on("SIGINT", shutdown); + +console.log("[sampling] listening on adaptive-sampling queue"); \ No newline at end of file diff --git a/apps/node-api/src/worker.ts b/apps/node-api/src/worker.ts index f2d248e..9b3e2c6 100644 --- a/apps/node-api/src/worker.ts +++ b/apps/node-api/src/worker.ts @@ -1,12 +1,13 @@ import { Worker, Job } from "bullmq"; -import { execFile } from "child_process"; +import { runPythonModule } from "./python"; import { db } from "./db"; const REDIS_URL = process.env.REDIS_URL ?? "redis://localhost:6379"; const EVALUATOR_TIMEOUT_MS = parseInt(process.env.EVALUATOR_TIMEOUT_MS ?? "120000", 10); -const EVALUATOR_PYTHON = process.env.EVALUATOR_PYTHON ?? "python3"; const EVALUATOR_MODULE = process.env.EVALUATOR_MODULE ?? "evaluator"; -const EVALUATOR_PYTHONPATH = process.env.EVALUATOR_PYTHONPATH ?? ""; +const ISSUE_AGENT_MODULE = process.env.ISSUE_AGENT_MODULE ?? "issue_agent"; +const PROMOTION_CONFIDENCE_THRESHOLD = parseFloat(process.env.PROMOTION_CONFIDENCE_THRESHOLD ?? "0.7"); +const ISSUE_AGENT_TIMEOUT_MS = parseInt(process.env.ISSUE_AGENT_TIMEOUT_MS ?? "180000", 10); function parseRedisUrl(urlStr: string) { const parsed = new URL(urlStr); @@ -24,34 +25,26 @@ function parseRedisUrl(urlStr: string) { } function runEvaluator(incidentJson: string): Promise { - return new Promise((resolve, reject) => { - const child = execFile( - EVALUATOR_PYTHON, - ["-m", EVALUATOR_MODULE], - { - env: { - ...process.env, - ...(EVALUATOR_PYTHONPATH ? { PYTHONPATH: EVALUATOR_PYTHONPATH } : {}), - }, - maxBuffer: 1024 * 1024, - timeout: EVALUATOR_TIMEOUT_MS, - }, - (err, stdout, stderr) => { - if (err) { - reject(new Error(`Evaluator exited: ${err.message}\nStderr: ${stderr}`)); - return; - } - resolve(stdout.trim()); - }, - ); - child.stdin?.write(incidentJson); - child.stdin?.end(); - }); + return runPythonModule(EVALUATOR_MODULE, incidentJson, EVALUATOR_TIMEOUT_MS); +} + +function runIssueAgent(snapshotJson: string): Promise { + return runPythonModule(ISSUE_AGENT_MODULE, snapshotJson, ISSUE_AGENT_TIMEOUT_MS); +} + +export function shouldPromoteToIssueAgent( + jobName: string, + classification: string, + confidence: number, + failureModes: string[], +): boolean { + if (jobName === "critical-incident") return true; + return classification === "REAL_INCIDENT" && confidence >= PROMOTION_CONFIDENCE_THRESHOLD && failureModes.length > 0 && !failureModes.includes("NONE_DETECTED"); } async function processJob(job: Job<{ incidentId: string }, void, string>) { const { incidentId } = job.data; - console.log(`[worker] processing incident ${incidentId} (job ${job.id})`); + console.log(`[worker] processing incident ${incidentId} (job ${job.id}, type ${job.name})`); const incident = await db.incident.findUnique({ where: { id: incidentId }, @@ -107,8 +100,16 @@ async function processJob(job: Job<{ incidentId: string }, void, string>) { return; } - const evaluation = parsed.evaluation as Record; - const metadata = parsed.metadata as Record; + const evaluation = parsed.evaluation as Record | undefined; + if (!evaluation) { + await db.incident.update({ + where: { id: incidentId }, + data: { analysis_status: "FAILED" }, + }); + console.error(`[worker] evaluator returned no evaluation for ${incidentId}`); + return; + } + const metadata = (parsed.metadata as Record) ?? {}; try { await db.$transaction(async (tx) => { @@ -139,8 +140,78 @@ async function processJob(job: Job<{ incidentId: string }, void, string>) { } console.log( - `[worker] completed ${incidentId}: ${String(evaluation.classification)} (confidence: ${String(evaluation.confidence)})`, + `[worker] evaluated ${incidentId}: ${String(evaluation.classification)} (confidence: ${String(evaluation.confidence)})`, ); + + if (!shouldPromoteToIssueAgent( + job.name, + String(evaluation.classification ?? ""), + Number(evaluation.confidence ?? 0), + (evaluation.failure_modes as string[]) ?? [], + )) { + console.log(`[worker] skipped issue agent for ${incidentId} (below promotion threshold)`); + return; + } + + console.log(`[worker] promoting ${incidentId} to issue agent`); + let issueOutput: string; + try { + const snapshot = { + incident_id: incidentId, + execution_trace: { + agent_steps: (incident as any).agent_steps ?? [], + model: (metadata.model_version as string) ?? "", + total_latency_ms: (reportData.telemetry as any)?.total_latency_ms ?? null, + tokens_used: (reportData.telemetry as any)?.total_prompt_tokens != null + ? (reportData.telemetry as any).total_prompt_tokens + ((reportData.telemetry as any)?.total_completion_tokens ?? 0) + : null, + }, + evaluation: { + failure_modes: (evaluation.failure_modes as string[]) ?? [], + confidence: Number(evaluation.confidence ?? 0), + reasoning: ((evaluation.reasoning as string[]) ?? []).join("\n"), + urgency_tier: ((evaluation.urgency as Record)?.tier as string) ?? "P2", + severity: incident.severity === "CRITICAL" ? "CRITICAL" : "HIGH", + }, + telemetry: (incident as any).telemetry ?? {}, + metadata: { incident_fingerprint: incident.fingerprint }, + }; + issueOutput = await runIssueAgent(JSON.stringify(snapshot)); + } catch (err) { + console.error(`[worker] issue agent failed for ${incidentId}:`, (err as Error).message); + return; + } + + let engineeringReport: Record; + let issueUrl: string | null = null; + try { + const parsedIssue = JSON.parse(issueOutput); + if (parsedIssue.issue_title) { + engineeringReport = parsedIssue; + } else if (typeof issueOutput === "string" && issueOutput.startsWith("GitHub issue created:")) { + issueUrl = issueOutput.replace("GitHub issue created: ", "").trim(); + engineeringReport = {}; + } else { + engineeringReport = {}; + } + } catch { + if (issueOutput.startsWith("GitHub issue created:")) { + issueUrl = issueOutput.replace("GitHub issue created: ", "").trim(); + engineeringReport = {}; + } else { + engineeringReport = { raw: issueOutput }; + } + } + + await db.incident.update({ + where: { id: incidentId }, + data: { + engineering_report: engineeringReport as object, + ...(issueUrl ? { issue_url: issueUrl } : {}), + }, + }); + + console.log(`[worker] issue agent completed for ${incidentId}${issueUrl ? ` — ${issueUrl}` : ""}`); } const connection = parseRedisUrl(REDIS_URL); diff --git a/package-lock.json b/package-lock.json index eddc9d3..607d730 100644 --- a/package-lock.json +++ b/package-lock.json @@ -24,7 +24,12 @@ "name": "@void-server/node-api", "version": "0.1.0", "dependencies": { + "@void-server/adaptive-sampling": "*", "@void-server/db": "*", + "@void-server/incident-fingerprint": "*", + "@void-server/incident-formation": "*", + "@void-server/risk-engine": "*", + "bullmq": "^5.0.0", "cors": "^2.8.5", "express": "^4.19.2" }, diff --git a/packages/adaptive-sampling/src/queue.ts b/packages/adaptive-sampling/src/queue.ts index c91f95f..c829c06 100644 --- a/packages/adaptive-sampling/src/queue.ts +++ b/packages/adaptive-sampling/src/queue.ts @@ -99,6 +99,8 @@ export class BullMqSamplingQueue implements SamplingQueue { executionId: sample.executionId, traceId: sample.traceId, timestamp: sample.timestamp.toISOString(), + agentSteps: sample.agentSteps, + telemetry: sample.telemetry, }; await this.queue.add("sample", payload, { removeOnComplete: { count: 1000 }, diff --git a/packages/adaptive-sampling/src/types.ts b/packages/adaptive-sampling/src/types.ts index 33cf9a1..6142965 100644 --- a/packages/adaptive-sampling/src/types.ts +++ b/packages/adaptive-sampling/src/types.ts @@ -6,6 +6,8 @@ export interface SamplingInput { executionId: string; traceId?: string; timestamp: Date; + agentSteps?: unknown[]; + telemetry?: Record; } export interface SamplingQueue { diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 6315c5a..6476834 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -30,6 +30,8 @@ model Incident { last_seen DateTime @default(now()) @db.Timestamptz(3) analysis_status AnalysisStatus @default(PENDING) latest_labels Json? + engineering_report Json? + issue_url String? agent_steps Json? telemetry Json? created_at DateTime @default(now()) diff --git a/packages/incident-fingerprint/src/incident-fingerprint.ts b/packages/incident-fingerprint/src/incident-fingerprint.ts index 0e55e59..a67d4c6 100644 --- a/packages/incident-fingerprint/src/incident-fingerprint.ts +++ b/packages/incident-fingerprint/src/incident-fingerprint.ts @@ -1,10 +1,22 @@ import { createHash } from "node:crypto"; import { RiskLabel } from "./types"; -export function generateFingerprint( +export function hashJoin(strings: readonly string[]): string { + return createHash("sha256") + .update(JSON.stringify([...strings].sort())) + .digest("hex"); +} + +export function generateLegacyFingerprint( labels: readonly RiskLabel[], ): string { return createHash("sha256") .update(labels.join("|")) .digest("hex"); } + +export function generateFingerprint( + labels: readonly RiskLabel[], +): string { + return hashJoin(labels); +} diff --git a/packages/incident-fingerprint/src/index.ts b/packages/incident-fingerprint/src/index.ts index 322b189..61af3a4 100644 --- a/packages/incident-fingerprint/src/index.ts +++ b/packages/incident-fingerprint/src/index.ts @@ -1,3 +1,3 @@ export { RiskLabel, type RiskEvaluationResult, type Severity } from "./types"; export { normalizeRiskLabels } from "./risk-labels"; -export { generateFingerprint } from "./incident-fingerprint"; +export { generateFingerprint, generateLegacyFingerprint, hashJoin } from "./incident-fingerprint"; diff --git a/packages/incident-fingerprint/tests/incident-fingerprint.test.ts b/packages/incident-fingerprint/tests/incident-fingerprint.test.ts index 0502446..6095258 100644 --- a/packages/incident-fingerprint/tests/incident-fingerprint.test.ts +++ b/packages/incident-fingerprint/tests/incident-fingerprint.test.ts @@ -1,6 +1,6 @@ import { describe, it, expect } from "vitest"; import { RiskLabel } from "../src/types"; -import { generateFingerprint } from "../src/incident-fingerprint"; +import { generateFingerprint, generateLegacyFingerprint, hashJoin } from "../src/incident-fingerprint"; describe("generateFingerprint", () => { it("identical labels produce identical fingerprint", () => { @@ -31,11 +31,22 @@ describe("generateFingerprint", () => { expect(fp).toMatch(/^[0-9a-f]{64}$/); }); - it("assumes input is already normalized (no internal dedup or sort)", () => { + it("sorts labels deterministically", () => { const sorted = [RiskLabel.HIGH_LATENCY, RiskLabel.TOOL_FAILURE]; const unsorted = [RiskLabel.TOOL_FAILURE, RiskLabel.HIGH_LATENCY]; - expect(generateFingerprint(sorted)).not.toBe( - generateFingerprint(unsorted), - ); + expect(generateFingerprint(sorted)).toBe(generateFingerprint(unsorted)); + }); + + it("prevents delimiter collision issues in hashJoin", () => { + const hashA = hashJoin(["a|b", "c"]); + const hashB = hashJoin(["a", "b|c"]); + expect(hashA).not.toBe(hashB); + }); + + it("generates legacy order-sensitive fingerprint for backward compatibility", () => { + const labels = [RiskLabel.TOOL_FAILURE, RiskLabel.HIGH_LATENCY]; + const legacyFp = generateLegacyFingerprint(labels); + expect(legacyFp).toMatch(/^[0-9a-f]{64}$/); + expect(legacyFp).not.toBe(generateFingerprint(labels)); }); }); diff --git a/packages/incident-formation/src/service.ts b/packages/incident-formation/src/service.ts index 5606fa1..b87b7a3 100644 --- a/packages/incident-formation/src/service.ts +++ b/packages/incident-formation/src/service.ts @@ -1,4 +1,5 @@ import type { RiskLabel } from "@void-server/incident-fingerprint"; +import { generateFingerprint, generateLegacyFingerprint } from "@void-server/incident-fingerprint"; import type { IncidentInput, IncidentRecord, IncidentRepository, IncidentQueue, ProcessResult } from "./types"; import { JOB_TYPES } from "./types"; @@ -18,7 +19,15 @@ export class IncidentFormationService { return { action: "SKIPPED" }; } - const existing = await this.repo.findByFingerprint(input.fingerprint); + const fingerprint = input.fingerprint ?? generateFingerprint(input.labels); + let existing = await this.repo.findByFingerprint(fingerprint); + + if (!existing && !input.fingerprint && input.labels.length > 0) { + const legacyFingerprint = generateLegacyFingerprint(input.labels); + if (legacyFingerprint !== fingerprint) { + existing = await this.repo.findByFingerprint(legacyFingerprint); + } + } if (existing) { return this.handleExisting(existing, input); @@ -27,7 +36,7 @@ export class IncidentFormationService { let created; try { created = await this.repo.create({ - fingerprint: input.fingerprint, + fingerprint, trace_id: input.traceId ?? "", execution_id: input.executionId, title: generateTitle(input.severity, input.labels), @@ -45,7 +54,13 @@ export class IncidentFormationService { }); } catch (err: any) { if (err?.code === "P2002") { - const raceExisting = await this.repo.findByFingerprint(input.fingerprint); + let raceExisting = await this.repo.findByFingerprint(fingerprint); + if (!raceExisting && !input.fingerprint && input.labels.length > 0) { + const legacyFingerprint = generateLegacyFingerprint(input.labels); + if (legacyFingerprint !== fingerprint) { + raceExisting = await this.repo.findByFingerprint(legacyFingerprint); + } + } if (raceExisting) return this.handleExisting(raceExisting, input); } throw err; diff --git a/packages/incident-formation/src/types.ts b/packages/incident-formation/src/types.ts index 4e7f4ae..9608f54 100644 --- a/packages/incident-formation/src/types.ts +++ b/packages/incident-formation/src/types.ts @@ -8,7 +8,7 @@ export type AnalysisStatus = "PENDING" | "PROCESSING" | "COMPLETED" | "FAILED"; export type JobType = "evaluate-incident" | "critical-incident"; export interface IncidentInput { - fingerprint: string; + fingerprint?: string; severity: RiskSeverity; labels: RiskLabel[]; executionId: string; diff --git a/packages/incident-formation/tests/formation.test.ts b/packages/incident-formation/tests/formation.test.ts index e1ce28a..9143e89 100644 --- a/packages/incident-formation/tests/formation.test.ts +++ b/packages/incident-formation/tests/formation.test.ts @@ -1,4 +1,4 @@ -import { RiskLabel } from "@void-server/incident-fingerprint"; +import { RiskLabel, generateLegacyFingerprint } from "@void-server/incident-fingerprint"; import { describe, it, expect, vi } from "vitest"; import { IncidentFormationService, generateTitle } from "../src/service"; import type { IncidentInput, IncidentRecord, IncidentRepository, IncidentQueue } from "../src/types"; @@ -152,6 +152,29 @@ describe("IncidentFormationService", () => { expect(result.action).toBe("UPDATED"); }); + it("falls back to legacy fingerprint lookup when new fingerprint finds no incident", async () => { + const repo = createMockRepo(); + const queue = createMockQueue(); + + const existing = makeRecord({ severity: "SUSPICIOUS", occurrence: 1 }); + repo.findByFingerprint = vi.fn().mockImplementation((fp: string) => { + if (fp === generateLegacyFingerprint([RiskLabel.HIGH_LATENCY])) { + return Promise.resolve(existing); + } + return Promise.resolve(null); + }); + repo.update = vi.fn().mockResolvedValue(makeRecord({ severity: "SUSPICIOUS", occurrence: 2 })); + + const service = new IncidentFormationService(repo, queue); + const result = await service.process(suspiciousInput({ executionId: "exec-2", fingerprint: undefined })); + + expect(repo.findByFingerprint).toHaveBeenCalledTimes(2); + expect(repo.update).toHaveBeenCalledWith("inc-1", expect.objectContaining({ + execution_id: "exec-2", + })); + expect(result.action).toBe("UPDATED"); + }); + it("enqueues critical job when escalating a suspicious incident to critical", async () => { const repo = createMockRepo(); const queue = createMockQueue();