diff --git a/.env.example b/.env.example index 5b26663..d3f4719 100644 --- a/.env.example +++ b/.env.example @@ -36,9 +36,11 @@ OLLAMA_BASE_URL=http://localhost:11434 # ── Embeddings provider ─────────────────────────────────────────────────────── # Controls semantic cluster merging at analysis time (explain / clusters / -# timeline / compare) and semantic `ask` over stored pgvector embeddings. +# timeline / compare), semantic `ask` over stored pgvector embeddings, and +# similar-incident search over cluster_embeddings. # Options: disabled | openai | local -# Default: disabled — fingerprint-only clustering; ask uses keyword retrieval +# Default: disabled — fingerprint-only clustering; ask uses keyword retrieval; +# similar uses fingerprint equality EMBEDDINGS_PROVIDER=disabled # Embedding model (used when provider is openai or local). @@ -65,6 +67,10 @@ CLUSTER_MERGE_MIN_COUNT=1 ASK_SEMANTIC_TOP_K=100 ASK_SEMANTIC_MIN_SIMILARITY=0.75 +# Similar-incident search (pgvector over cluster_embeddings). Falls back to +# exact fingerprint matching when the provider is down or no vectors exist. +SIMILAR_SEMANTIC_MIN_SIMILARITY=0.80 + # ── Analysis tuning ─────────────────────────────────────────────────────────── # Baseline comparison window for the explain command. diff --git a/README.md b/README.md index 54c13b9..0c2e35f 100644 --- a/README.md +++ b/README.md @@ -665,13 +665,14 @@ All settings are read from `.env`, environment variables, or CLI flags. Priority | `OPENAI_API_KEY` | _(empty)_ | API key for OpenAI or compatible endpoint | | `OPENAI_BASE_URL` | `https://api.openai.com/v1` | Base URL for OpenAI-compatible API | | `OLLAMA_BASE_URL` | `http://localhost:11434` | Ollama server URL | -| `EMBEDDINGS_PROVIDER` | `disabled` | `disabled`, `openai`, `local`. Cluster merge and semantic `ask` are skipped when `disabled` | +| `EMBEDDINGS_PROVIDER` | `disabled` | `disabled`, `openai`, `local`. Cluster merge, semantic `ask`, and `/similar` ANN skip when `disabled` | | `EMBEDDINGS_MODEL` | `text-embedding-3-small` | Embeddings model name | | `EMBEDDINGS_DIMENSIONS` | `1536` | Vector size passed to the OpenAI embeddings API. Persist/ask skip unless this is 1536 (stored column width) | | `CLUSTER_MERGE_SIMILARITY_THRESHOLD` | `0.92` | Cosine similarity at or above which fingerprint clusters merge. High on purpose so distinct errors stay separate | | `CLUSTER_MERGE_MIN_COUNT` | `1` | Minimum `count` for a cluster to participate in a merge | | `ASK_SEMANTIC_TOP_K` | `100` | Max log lines returned by semantic `ask` | | `ASK_SEMANTIC_MIN_SIMILARITY` | `0.75` | Minimum cosine similarity for a semantic `ask` hit. Looser than cluster-merge because questions paraphrase | +| `SIMILAR_SEMANTIC_MIN_SIMILARITY` | `0.80` | Minimum cosine similarity for `POST /v1/query/similar`. Falls back to fingerprint equality when embeddings are down | | `DEFAULT_BASELINE_WINDOW` | `24h` | How far back to compare for baseline | | `MAX_CLUSTERS_FOR_EXPLAIN` | `10` | Max clusters sent to the explain pipeline | | `MAX_EVIDENCE_ITEMS` | `8` | Max evidence lines in output | @@ -855,8 +856,9 @@ Fingerprinting can still split one incident across multiple clusters when wordin - **Disabled (default).** `EMBEDDINGS_PROVIDER=disabled` skips the merge pass entirely. Clustering is fingerprint-only and deterministic — the same logs always produce the same clusters. - **Enabled.** With `openai` or `local`, representatives are embedded at analysis time (in memory; not written to pgvector). Pairs with cosine similarity ≥ `CLUSTER_MERGE_SIMILARITY_THRESHOLD` (default **0.92**) are merged via connected components. Merged `count` is the sum of member counts; services and levels are summed; `first_seen` is the earliest timestamp and `last_seen` the latest; importance is recomputed. The canonical fingerprint is the member with the highest importance score. `ClusterRun.algorithm` is `fingerprint+semantic` when embeddings were used, even if no pair crossed the threshold. -- **Fail open.** If the embeddings backend is missing, raises, or returns unusable vectors, clustering continues with the fingerprint-only set. -- **Ask vs merge.** Semantic `ask` uses the *stored* `log_embeddings` table (populated by `raglogs ingest --with-embeddings`) and `ASK_SEMANTIC_MIN_SIMILARITY` (default **0.75**). Cluster merge still uses its own in-memory pass and threshold. Similar-incident search (`POST /v1/query/similar` / historical cluster ANN) is not in this release. Compare still applies its own heuristic collapse for webhook retries / queue growth after clustering. +- **Fail open.** If the embeddings backend is missing, raises, or returns unusable vectors, clustering continues with the fingerprint-only set. The same fail-open applies when upserting `cluster_embeddings` for similar-incident search: a provider outage never fails the cluster run. +- **Cluster template persist.** After clustering, raglogs upserts each cluster's representative template into `cluster_embeddings` keyed by `(scope, fingerprint)` when the embeddings provider is available. Similar-incident search queries those rows (not raw `log_embeddings`). Skip happens automatically when `EMBEDDINGS_PROVIDER=disabled`. +- **Ask vs merge vs similar.** Semantic `ask` uses the *stored* `log_embeddings` table (populated by `raglogs ingest --with-embeddings`) and `ASK_SEMANTIC_MIN_SIMILARITY` (default **0.75**). Cluster merge still uses its own in-memory pass and threshold. Similar-incident search (`POST /v1/query/similar`) uses the `cluster_embeddings` table (upserted at analysis time when an embeddings provider is available) and `SIMILAR_SEMANTIC_MIN_SIMILARITY` (default **0.80**). Compare still applies its own heuristic collapse for webhook retries / queue growth after clustering. Local embeddings require the optional extra: `pip install 'raglogs[local-embeddings]'` (`sentence-transformers`). If that import fails, merge is skipped. @@ -972,6 +974,15 @@ Keys are stored argon2-hashed with a short indexed prefix. Each key is **pinned* A pinned key that sends a different non-empty scope returns **403** `SCOPE_MISMATCH`. The CLI is scope-optional and defaults to `default` (`raglogs ingest --scope incident:INC-9`, `raglogs explain --scope …`). Each new key also gets a `whsec_…` webhook signing secret (shown once; `keys list` shows `whsec_****` only). +**`POST /v1/query/similar` cross-scope permissions.** Similar-incident search can look across isolation scopes ("we saw this in INC-1188") when the caller is allowed to see those scopes. Matches from a scope the caller cannot see are never returned. + +| Caller | Cross-scope similar | +|---|---| +| `admin` keys | Yes by default. Pass `"cross_scope": false` to pin to the resolved scope. | +| `query` keys that are pinned | Same-scope only. `"cross_scope": true` is ignored. | +| `query` keys minted with `--allow-scope-override` | Same-scope unless the body sets `"cross_scope": true`. | +| `AUTH_ENABLED=false` (local CLI / demo) | Cross-scope allowed by default. Pass `"cross_scope": false` to pin. | + Optional OIDC: set `AUTH_MODE=oidc` or `both` and `OIDC_ISSUER`. A JWT (three dotted segments) is validated via JWKS (`iss`, `exp`, and `aud` when `OIDC_AUDIENCE` is set). Role comes from claim `raglogs_role` or `roles`, defaulting to `query`. When `AUTH_MODE=api_key`, JWTs are rejected. If auth is disabled and the process binds a non-loopback address (`0.0.0.0`, `::`, a public IP), raglogs logs a warning. Set `AUTH_REFUSE_INSECURE_BIND=true` to refuse startup instead. @@ -992,6 +1003,7 @@ If auth is disabled and the process binds a non-loopback address (`0.0.0.0`, `:: | `POST` | `/v1/query/clusters` | List top clusters | | `POST` | `/v1/query/timeline` | Reconstruct incident timeline for a window | | `POST` | `/v1/query/compare` | Diff two time windows (same semantics as `raglogs compare`) | +| `POST` | `/v1/query/similar` | Prior incidents with nearby fingerprints ("we saw this in INC-1188") | | `GET` | `/v1/config` | Read effective configuration | **Push NDJSON.** `POST /v1/ingestions/lines` accepts newline-delimited lines (`Content-Type: application/x-ndjson`, `application/jsonl`, or `text/plain`). Each line is a raw log string or a JSON object with at least `message` / `raw` / `text` (optional `timestamp`, `service`, `level`, `host`, `env`). Cap is `INGEST_PUSH_MAX_LINES` (default 5000); over the cap returns 400. @@ -1133,7 +1145,15 @@ curl -X POST http://localhost:8000/v1/query/timeline \ **Compare** — `POST /v1/query/compare` matches `raglogs compare`: either `"since"` + `"baseline"` (durations, window A ends at request time) or explicit `window_a_from` / `window_a_to` / `window_b_from` / `window_b_to`. Optional `"format": "text"` adds `rendered_text` (and a `text` alias). Cluster diffs include a `marker` (`+` / `-` / `↑` / `↓`; triggers use `+⚡` / `-⚡`). Compare is rules-only. -Published JSON Schema files live in `clients/jsonschema/` (`explain.v1.json`, `timeline.v1.json`, `compare.v1.json`, `ask.v1.json`). Export with `make jsonschema`. +**Similar** — `POST /v1/query/similar` finds prior incidents whose cluster templates are near the current window's primary fingerprint(s). Pass `since` / `from_time` / `to_time` or `ingestion_job_id` to cluster the current incident, or skip clustering with `"fingerprint"` / `"fingerprints"`. Optional `"top"` (default 10) and `"cross_scope"` (see permissions above). Response `retrieval_mode` is `"semantic"` when pgvector ANN over `cluster_embeddings` hits, otherwise `"fingerprint"` (exact fingerprint equality, never HTTP 500 when embeddings are down). Rules-only (`llm.used` is always false). + +```bash +curl -X POST http://localhost:8000/v1/query/similar \ + -H "Content-Type: application/json" \ + -d '{"since": "1h", "cross_scope": true}' +``` + +Published JSON Schema files live in `clients/jsonschema/` (`explain.v1.json`, `timeline.v1.json`, `compare.v1.json`, `ask.v1.json`, `clusters.v1.json`, `similar.v1.json`). Export with `make jsonschema`. ```bash curl -X POST http://localhost:8000/v1/query/compare \ diff --git a/clients/README.md b/clients/README.md index 76a9a18..9a91225 100644 --- a/clients/README.md +++ b/clients/README.md @@ -16,7 +16,7 @@ Versioned response bodies (`schema_version` 1.0) are published as JSON Schema: ```bash make jsonschema -# writes clients/jsonschema/explain.v1.json (and timeline/compare/ask/clusters) +# writes clients/jsonschema/explain.v1.json (and timeline/compare/ask/clusters/similar) ``` ## Python diff --git a/clients/jsonschema/similar.v1.json b/clients/jsonschema/similar.v1.json new file mode 100644 index 0000000..b04906c --- /dev/null +++ b/clients/jsonschema/similar.v1.json @@ -0,0 +1,195 @@ +{ + "$defs": { + "LlmProvenance": { + "properties": { + "used": { + "title": "Used", + "type": "boolean" + }, + "provider": { + "title": "Provider", + "type": "string" + }, + "model": { + "title": "Model", + "type": "string" + }, + "fell_back": { + "title": "Fell Back", + "type": "boolean" + } + }, + "required": [ + "used", + "provider", + "model", + "fell_back" + ], + "title": "LlmProvenance", + "type": "object" + }, + "SimilarMatchModel": { + "properties": { + "scope": { + "title": "Scope", + "type": "string" + }, + "fingerprint": { + "title": "Fingerprint", + "type": "string" + }, + "template": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Template" + }, + "similarity": { + "title": "Similarity", + "type": "number" + }, + "first_seen": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "First Seen" + }, + "last_seen": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Last Seen" + }, + "count": { + "default": 0, + "title": "Count", + "type": "integer" + } + }, + "required": [ + "scope", + "fingerprint", + "similarity" + ], + "title": "SimilarMatchModel", + "type": "object" + }, + "SimilarQueryCluster": { + "properties": { + "fingerprint": { + "title": "Fingerprint", + "type": "string" + }, + "template": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Template" + } + }, + "required": [ + "fingerprint" + ], + "title": "SimilarQueryCluster", + "type": "object" + }, + "TimeWindow": { + "description": "Inclusive analysis window. Serialized as ``from`` / ``to`` (ISO-8601).", + "properties": { + "from": { + "title": "From", + "type": "string" + }, + "to": { + "title": "To", + "type": "string" + } + }, + "required": [ + "from", + "to" + ], + "title": "TimeWindow", + "type": "object" + } + }, + "properties": { + "schema_version": { + "default": "1.0", + "title": "Schema Version", + "type": "string" + }, + "scope": { + "default": "default", + "title": "Scope", + "type": "string" + }, + "window": { + "$ref": "#/$defs/TimeWindow" + }, + "query_clusters": { + "items": { + "$ref": "#/$defs/SimilarQueryCluster" + }, + "title": "Query Clusters", + "type": "array" + }, + "matches": { + "items": { + "$ref": "#/$defs/SimilarMatchModel" + }, + "title": "Matches", + "type": "array" + }, + "retrieval_mode": { + "default": "fingerprint", + "title": "Retrieval Mode", + "type": "string" + }, + "llm": { + "$ref": "#/$defs/LlmProvenance" + }, + "rendered_text": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Rendered Text" + } + }, + "required": [ + "window", + "llm" + ], + "title": "SimilarResponse", + "type": "object" +} diff --git a/clients/openapi.json b/clients/openapi.json index 4e326fd..e9c0926 100644 --- a/clients/openapi.json +++ b/clients/openapi.json @@ -95,6 +95,24 @@ "summary": "List Ingestions", "description": "List recent completed ingestion jobs, newest first. Used by the web UI's ingestion picker.", "operationId": "v1_ingestions_list_ingestions__v1_ingestions", + "parameters": [ + { + "name": "scope", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" + } + } + ], "responses": { "200": { "description": "Successful Response", @@ -105,6 +123,16 @@ } } } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } } } } @@ -160,6 +188,24 @@ "summary": "Get Latest Ingestion", "description": "Return the ID of the most recently completed ingestion job, or null if\nnone exists yet \u2014 the same default the CLI and GET /ingestions apply\n(latest ingestion, not all ingestions merged). Not currently called by\nthe web UI (its picker defaults to GET /ingestions[0] instead), but kept\nas a lighter-weight lookup for other integrations.\n\nRegistered before /{ingestion_job_id} so \"latest\" isn't swallowed by\nthat path param.", "operationId": "v1_ingestions_get_latest_ingestion__v1_ingestions_latest", + "parameters": [ + { + "name": "scope", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" + } + } + ], "responses": { "200": { "description": "Successful Response", @@ -170,6 +216,16 @@ } } } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } } } } @@ -455,6 +511,24 @@ "description": "List recent completed ingestion jobs, newest first. Used by the web UI's ingestion picker.", "operationId": "legacy_ingestions_list_ingestions__ingestions", "deprecated": true, + "parameters": [ + { + "name": "scope", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" + } + } + ], "responses": { "200": { "description": "Successful Response", @@ -465,6 +539,16 @@ } } } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } } } } @@ -521,6 +605,25 @@ "summary": "Get Latest Ingestion", "description": "Return the ID of the most recently completed ingestion job, or null if\nnone exists yet \u2014 the same default the CLI and GET /ingestions apply\n(latest ingestion, not all ingestions merged). Not currently called by\nthe web UI (its picker defaults to GET /ingestions[0] instead), but kept\nas a lighter-weight lookup for other integrations.\n\nRegistered before /{ingestion_job_id} so \"latest\" isn't swallowed by\nthat path param.", "operationId": "legacy_ingestions_get_latest_ingestion__ingestions_latest", + "deprecated": true, + "parameters": [ + { + "name": "scope", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" + } + } + ], "responses": { "200": { "description": "Successful Response", @@ -531,9 +634,18 @@ } } } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } } - }, - "deprecated": true + } } }, "/ingestions/lines": { @@ -1167,6 +1279,89 @@ "deprecated": true } }, + "/v1/query/similar": { + "post": { + "tags": [ + "query" + ], + "summary": "Similar Endpoint", + "operationId": "v1_query_similar_endpoint__v1_query_similar", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SimilarRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SimilarResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, + "/query/similar": { + "post": { + "tags": [ + "query" + ], + "summary": "Similar Endpoint", + "operationId": "legacy_query_similar_endpoint__query_similar", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SimilarRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SimilarResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + }, + "deprecated": true + } + }, "/v1/config": { "get": { "tags": [ @@ -1271,6 +1466,17 @@ } ], "title": "Ingestion Job Id" + }, + "scope": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" } }, "type": "object", @@ -1587,6 +1793,17 @@ } ], "title": "Ingestion Job Id" + }, + "scope": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" } }, "type": "object", @@ -1755,6 +1972,17 @@ ], "title": "Format", "default": "json" + }, + "scope": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" } }, "type": "object", @@ -2063,6 +2291,17 @@ ], "title": "Format", "default": "json" + }, + "scope": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" } }, "type": "object", @@ -2223,6 +2462,16 @@ } ], "title": "Tail Jobs" + }, + "llm_breaker": { + "anyOf": [ + { + "$ref": "#/components/schemas/LlmBreakerHealth" + }, + { + "type": "null" + } + ] } }, "type": "object", @@ -2368,6 +2617,17 @@ } ], "title": "Callback Url" + }, + "scope": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" } }, "type": "object", @@ -2561,6 +2821,29 @@ ], "title": "LatestIngestionResponse" }, + "LlmBreakerHealth": { + "properties": { + "state": { + "type": "string", + "title": "State" + }, + "consecutive_failures": { + "type": "integer", + "title": "Consecutive Failures" + }, + "cooldown_remaining_seconds": { + "type": "number", + "title": "Cooldown Remaining Seconds" + } + }, + "type": "object", + "required": [ + "state", + "consecutive_failures", + "cooldown_remaining_seconds" + ], + "title": "LlmBreakerHealth" + }, "LlmProvenance": { "properties": { "used": { @@ -2627,6 +2910,275 @@ ], "title": "PushLinesResponse" }, + "SimilarMatchModel": { + "properties": { + "scope": { + "type": "string", + "title": "Scope" + }, + "fingerprint": { + "type": "string", + "title": "Fingerprint" + }, + "template": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Template" + }, + "similarity": { + "type": "number", + "title": "Similarity" + }, + "first_seen": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "First Seen" + }, + "last_seen": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Last Seen" + }, + "count": { + "type": "integer", + "title": "Count", + "default": 0 + } + }, + "type": "object", + "required": [ + "scope", + "fingerprint", + "similarity" + ], + "title": "SimilarMatchModel" + }, + "SimilarQueryCluster": { + "properties": { + "fingerprint": { + "type": "string", + "title": "Fingerprint" + }, + "template": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Template" + } + }, + "type": "object", + "required": [ + "fingerprint" + ], + "title": "SimilarQueryCluster" + }, + "SimilarRequest": { + "properties": { + "since": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Since" + }, + "from_time": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "From Time" + }, + "to_time": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "To Time" + }, + "service": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Service" + }, + "env": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Env" + }, + "ingestion_job_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Ingestion Job Id" + }, + "fingerprint": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Fingerprint" + }, + "fingerprints": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Fingerprints" + }, + "top": { + "type": "integer", + "maximum": 50.0, + "minimum": 1.0, + "title": "Top", + "default": 10 + }, + "cross_scope": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "title": "Cross Scope" + }, + "scope": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" + } + }, + "type": "object", + "title": "SimilarRequest" + }, + "SimilarResponse": { + "properties": { + "schema_version": { + "type": "string", + "title": "Schema Version", + "default": "1.0" + }, + "scope": { + "type": "string", + "title": "Scope", + "default": "default" + }, + "window": { + "$ref": "#/components/schemas/TimeWindow" + }, + "query_clusters": { + "items": { + "$ref": "#/components/schemas/SimilarQueryCluster" + }, + "type": "array", + "title": "Query Clusters" + }, + "matches": { + "items": { + "$ref": "#/components/schemas/SimilarMatchModel" + }, + "type": "array", + "title": "Matches" + }, + "retrieval_mode": { + "type": "string", + "title": "Retrieval Mode", + "default": "fingerprint" + }, + "llm": { + "$ref": "#/components/schemas/LlmProvenance" + }, + "rendered_text": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Rendered Text" + } + }, + "type": "object", + "required": [ + "window", + "llm" + ], + "title": "SimilarResponse" + }, "TailLifecycleResponse": { "properties": { "ingestion_job_id": { @@ -2809,6 +3361,17 @@ ], "title": "Format", "default": "json" + }, + "scope": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Scope" } }, "type": "object", diff --git a/migrations/versions/0009_cluster_embeddings.py b/migrations/versions/0009_cluster_embeddings.py new file mode 100644 index 0000000..9460276 --- /dev/null +++ b/migrations/versions/0009_cluster_embeddings.py @@ -0,0 +1,96 @@ +"""cluster embeddings table and ANN indexes on vector columns + +Revision ID: 0009_cluster_embeddings +Revises: 0008_scope_isolation +Create Date: 2026-08-17 00:00:00.000000 + +Adds ``cluster_embeddings`` keyed by ``(scope, fingerprint)`` so similar-incident +search can look up historical cluster templates. Creates an ANN index on +``log_embeddings.embedding`` (missing since 0001) and on the new cluster +vector column. Prefers HNSW (safe on empty tables); IVFFlat ``lists=1`` is +the fallback when HNSW is unavailable. +""" + +from typing import Sequence, Union + +import pgvector.sqlalchemy +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import context, op + +revision: str = "0009_cluster_embeddings" +down_revision: Union[str, None] = "0008_scope_isolation" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +LOG_EMBEDDING_ANN_INDEX = "ix_log_embeddings_embedding_ann" +CLUSTER_EMBEDDING_ANN_INDEX = "ix_cluster_embeddings_embedding_ann" + + +def _create_ann_index(table: str, column: str, index_name: str) -> None: + """Create HNSW if the pgvector build supports it; else IVFFlat lists=1.""" + hnsw_sql = ( + f"CREATE INDEX IF NOT EXISTS {index_name} ON {table} " + f"USING hnsw ({column} vector_cosine_ops)" + ) + ivf_sql = ( + f"CREATE INDEX IF NOT EXISTS {index_name} ON {table} " + f"USING ivfflat ({column} vector_cosine_ops) WITH (lists = 1)" + ) + if context.is_offline_mode(): + # Offline SQL cannot probe the server; HNSW is the documented default + # (pgvector 0.5+, including the pgvector/pgvector:pg16 image). + op.execute(sa.text(hnsw_sql)) + return + + conn = op.get_bind() + try: + with conn.begin_nested(): + conn.execute(sa.text(hnsw_sql)) + return + except Exception: + pass + conn.execute(sa.text(ivf_sql)) + + +def upgrade() -> None: + op.create_table( + "cluster_embeddings", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column("scope", sa.String(255), nullable=False, server_default="default"), + sa.Column("fingerprint", sa.String(64), nullable=False), + sa.Column("template", sa.Text, nullable=True), + sa.Column("embedding", pgvector.sqlalchemy.Vector(1536), nullable=False), + sa.Column("model_name", sa.String(255), nullable=False), + sa.Column("first_seen", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_seen", sa.DateTime(timezone=True), nullable=True), + sa.Column("count", sa.Integer, nullable=False, server_default="0"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.UniqueConstraint( + "scope", + "fingerprint", + name="ux_cluster_embeddings_scope_fingerprint", + ), + ) + op.create_index( + "ix_cluster_embeddings_fingerprint", + "cluster_embeddings", + ["fingerprint"], + ) + op.create_index( + "ix_cluster_embeddings_scope", + "cluster_embeddings", + ["scope"], + ) + _create_ann_index("log_embeddings", "embedding", LOG_EMBEDDING_ANN_INDEX) + _create_ann_index("cluster_embeddings", "embedding", CLUSTER_EMBEDDING_ANN_INDEX) + + +def downgrade() -> None: + op.drop_index(CLUSTER_EMBEDDING_ANN_INDEX, table_name="cluster_embeddings") + op.drop_index(LOG_EMBEDDING_ANN_INDEX, table_name="log_embeddings") + op.drop_index("ix_cluster_embeddings_scope", table_name="cluster_embeddings") + op.drop_index("ix_cluster_embeddings_fingerprint", table_name="cluster_embeddings") + op.drop_table("cluster_embeddings") diff --git a/scripts/export_jsonschema.py b/scripts/export_jsonschema.py index d423ed4..7a6f79e 100644 --- a/scripts/export_jsonschema.py +++ b/scripts/export_jsonschema.py @@ -19,6 +19,7 @@ ClustersResponse, CompareResponse, ExplainResponse, + SimilarResponse, TimelineResponse, ) @@ -28,6 +29,7 @@ ("compare.v1.json", CompareResponse), ("ask.v1.json", AskResponse), ("clusters.v1.json", ClustersResponse), + ("similar.v1.json", SimilarResponse), ) diff --git a/src/api/app.py b/src/api/app.py index 78fa427..b43d0d5 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -12,7 +12,7 @@ from src.api.auth.scope import ScopeResolutionError, scope_error_response from src.api.deprecation import DeprecationHeaderMiddleware from src.api.ratelimit import RateLimitMiddleware -from src.api.routes import ask, clusters, compare_windows, config, explain, health, ingestions, timeline, ui +from src.api.routes import ask, clusters, compare_windows, config, explain, health, ingestions, similar, timeline, ui _OPENAPI_DESCRIPTION = """Incident explanation API — ask your logs what happened. @@ -101,6 +101,7 @@ def _include_v1_and_alias(router: APIRouter, *, suffix: str, tags: list[str]) -> _include_v1_and_alias(clusters.router, suffix="/query", tags=["query"]) _include_v1_and_alias(timeline.router, suffix="/query", tags=["query"]) _include_v1_and_alias(compare_windows.router, suffix="/query", tags=["query"]) +_include_v1_and_alias(similar.router, suffix="/query", tags=["query"]) _include_v1_and_alias(config.router, suffix="/config", tags=["config"]) app.mount( diff --git a/src/api/routes/similar.py b/src/api/routes/similar.py new file mode 100644 index 0000000..cda2513 --- /dev/null +++ b/src/api/routes/similar.py @@ -0,0 +1,171 @@ +"""Similar-incident search API (G11). + +POST /query/similar — given a scope's primary cluster(s), return prior +incidents with nearby fingerprints. Dual-mounted at /v1/query/similar and +the deprecated /query/similar alias. +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from typing import Optional + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel, Field + +from src.api.schemas.v1 import ( + SCHEMA_VERSION, + SimilarMatchModel, + SimilarQueryCluster, + SimilarResponse, + llm_rules_only, + window_from_bounds, +) + +router = APIRouter() + + +class SimilarRequest(BaseModel): + since: Optional[str] = None + from_time: Optional[datetime] = None + to_time: Optional[datetime] = None + service: Optional[str] = None + env: Optional[str] = None + ingestion_job_id: Optional[str] = None + fingerprint: Optional[str] = None + fingerprints: Optional[list[str]] = None + top: int = Field(default=10, ge=1, le=50) + cross_scope: Optional[bool] = None + scope: Optional[str] = None + + +def _iso(value: Optional[datetime]) -> Optional[str]: + return value.isoformat() if value is not None else None + + +@router.post( + "/similar", + response_model=SimilarResponse, + response_model_exclude_unset=True, + response_model_by_alias=True, +) +def similar_endpoint(request: SimilarRequest, http_request: Request) -> SimilarResponse: + from src.api.auth.middleware import AuthPrincipal + from src.api.auth.scope import bind_request_scope + from src.config import get_settings + from src.core.clustering.clusterer import run_clustering + from src.core.retrieval.similar import ( + collect_query_fingerprints, + find_similar_incidents, + query_clusters_from_cluster_data, + query_clusters_from_fingerprints, + render_similar_summary, + resolve_similar_visibility, + ) + from src.db.session import get_db + from src.utils.time import resolve_window + + scope = bind_request_scope(http_request, request.scope) + settings = get_settings() + principal = getattr(http_request.state, "auth_principal", None) + if principal is not None and not isinstance(principal, AuthPrincipal): + principal = None + + visibility = resolve_similar_visibility( + auth_enabled=bool(settings.auth_enabled), + resolved_scope=scope, + cross_scope_requested=request.cross_scope, + role=principal.role if principal is not None else "", + allow_scope_override=bool(principal.allow_scope_override) + if principal is not None + else False, + ) + + fingerprints = collect_query_fingerprints(request.fingerprint, request.fingerprints) + + window_start: Optional[datetime] = None + window_end: Optional[datetime] = None + try: + window_start, window_end = resolve_window( + since=request.since, + from_time=request.from_time, + to_time=request.to_time, + ) + except ValueError as exc: + if not fingerprints: + raise HTTPException(status_code=400, detail=str(exc)) from exc + now = datetime.now(tz=timezone.utc) + window_start, window_end = now, now + + ingestion_job_id: Optional[uuid.UUID] = None + if request.ingestion_job_id: + try: + ingestion_job_id = uuid.UUID(request.ingestion_job_id) + except ValueError as exc: + raise HTTPException( + status_code=400, detail="Invalid ingestion_job_id" + ) from exc + + try: + with get_db() as db: + if fingerprints: + query_clusters = query_clusters_from_fingerprints(fingerprints) + else: + _, clusters = run_clustering( + db=db, + window_start=window_start, + window_end=window_end, + service=request.service, + environment=request.env, + max_clusters=max(request.top, 5), + save_to_db=False, + ingestion_job_id=ingestion_job_id, + scope=scope, + ) + query_clusters = query_clusters_from_cluster_data(clusters) + + result = find_similar_incidents( + db, + query_clusters, + query_scope=scope, + visibility=visibility, + window_start=window_start, + window_end=window_end, + top=request.top, + ) + except Exception: + from src.core.retrieval.similar import SimilarResult + + result = SimilarResult( + query_clusters=query_clusters_from_fingerprints(fingerprints), + matches=[], + retrieval_mode="fingerprint", + window_start=window_start, + window_end=window_end, + ) + + return SimilarResponse( + schema_version=SCHEMA_VERSION, + scope=scope, + window=window_from_bounds(result.window_start, result.window_end), + query_clusters=[ + SimilarQueryCluster(fingerprint=c.fingerprint, template=c.template or None) + for c in result.query_clusters + ], + matches=[ + SimilarMatchModel( + scope=m.scope, + fingerprint=m.fingerprint, + template=m.template, + similarity=round(m.similarity, 4), + first_seen=_iso(m.first_seen), + last_seen=_iso(m.last_seen), + count=m.count, + ) + for m in result.matches + ], + retrieval_mode=result.retrieval_mode, + llm=llm_rules_only(), + rendered_text=render_similar_summary(result.matches, result.retrieval_mode), + ) diff --git a/src/api/schemas/v1.py b/src/api/schemas/v1.py index b0f6646..92c6705 100644 --- a/src/api/schemas/v1.py +++ b/src/api/schemas/v1.py @@ -3,6 +3,7 @@ Pydantic models here are the HTTP contract. CLI still uses ``ExplainResult`` string confidence and ``summary_text``; this module only shapes JSON bodies. """ + from __future__ import annotations from datetime import datetime @@ -36,8 +37,13 @@ class TimeWindow(BaseModel): model_config = ConfigDict(populate_by_name=True) - from_: str = Field(validation_alias=AliasChoices("from", "from_", "start"), serialization_alias="from") - to: str = Field(validation_alias=AliasChoices("to", "end"), serialization_alias="to") + from_: str = Field( + validation_alias=AliasChoices("from", "from_", "start"), + serialization_alias="from", + ) + to: str = Field( + validation_alias=AliasChoices("to", "end"), serialization_alias="to" + ) @model_validator(mode="before") @classmethod @@ -195,6 +201,34 @@ class ClustersResponse(BaseModel): ingestion_job_id: Optional[str] = None +class SimilarQueryCluster(BaseModel): + fingerprint: str + template: Optional[str] = None + + +class SimilarMatchModel(BaseModel): + scope: str + fingerprint: str + template: Optional[str] = None + similarity: float + first_seen: Optional[str] = None + last_seen: Optional[str] = None + count: int = 0 + + +class SimilarResponse(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + schema_version: str = SCHEMA_VERSION + scope: str = "default" + window: TimeWindow + query_clusters: list[SimilarQueryCluster] = Field(default_factory=list) + matches: list[SimilarMatchModel] = Field(default_factory=list) + retrieval_mode: str = "fingerprint" + llm: LlmProvenance + rendered_text: Optional[str] = None + + def scope_from_request(http_request: Optional[Request] = None) -> str: """Resolved G8 scope when ``bind_request_scope`` ran; else principal or ``default``.""" if http_request is None: @@ -242,7 +276,9 @@ def window_from_bounds(start: datetime, end: datetime) -> TimeWindow: return TimeWindow(from_=start.isoformat(), to=end.isoformat()) -def window_from_mapping(raw: Any, fallback_start: str = "", fallback_end: str = "") -> TimeWindow: +def window_from_mapping( + raw: Any, fallback_start: str = "", fallback_end: str = "" +) -> TimeWindow: if isinstance(raw, TimeWindow): return raw if not isinstance(raw, dict): @@ -293,7 +329,9 @@ def _services_list(raw: Any) -> list[str]: def cluster_from_mapping(raw: Optional[dict[str, Any]]) -> Optional[ClusterEvidence]: if not raw: return None - template = raw.get("template") or raw.get("message") or raw.get("representative_message") + template = ( + raw.get("template") or raw.get("message") or raw.get("representative_message") + ) template_str = str(template) if template is not None else None return ClusterEvidence( fingerprint=raw.get("fingerprint"), @@ -342,7 +380,9 @@ def trigger_from_candidates( primary: Optional[ClusterEvidence] = None, ) -> TriggerInfo: if not candidates: - return TriggerInfo(detected=False, type=None, service=None, at=None, correlation=None) + return TriggerInfo( + detected=False, type=None, service=None, at=None, correlation=None + ) first = candidates[0] if isinstance(first, dict): message = str(first.get("message") or "") @@ -369,7 +409,9 @@ def trigger_from_candidates( ) -def trigger_from_mapping(raw: Any, candidates: list[Any], primary: Optional[ClusterEvidence]) -> TriggerInfo: +def trigger_from_mapping( + raw: Any, candidates: list[Any], primary: Optional[ClusterEvidence] +) -> TriggerInfo: if isinstance(raw, TriggerInfo): return raw if isinstance(raw, dict) and "detected" in raw: @@ -399,7 +441,9 @@ def explain_from_result( ) -> ExplainResponse: primary = cluster_from_mapping(result.primary_cluster) secondary = [ - c for c in (cluster_from_mapping(item) for item in result.secondary_clusters) if c is not None + c + for c in (cluster_from_mapping(item) for item in result.secondary_clusters) + if c is not None ] prose = result.summary_text or "" return ExplainResponse( @@ -432,10 +476,16 @@ def explain_from_cached( """Upgrade a cached dict (old or v1) so the response always has v1 fields.""" primary = cluster_from_mapping(payload.get("primary_cluster")) secondary_raw = payload.get("secondary_clusters") or [] - secondary = [c for c in (cluster_from_mapping(item) for item in secondary_raw) if c is not None] + secondary = [ + c + for c in (cluster_from_mapping(item) for item in secondary_raw) + if c is not None + ] candidates = payload.get("trigger_candidates") or [] prose = str(payload.get("rendered_text") or payload.get("summary") or "") - already_v1 = isinstance(payload.get("confidence"), dict) and bool(payload.get("schema_version")) + already_v1 = isinstance(payload.get("confidence"), dict) and bool( + payload.get("schema_version") + ) if already_v1: summary_text = str(payload.get("summary") or short_summary(prose)) else: diff --git a/src/clients/v1.py b/src/clients/v1.py index fbff3c9..c42c46b 100644 --- a/src/clients/v1.py +++ b/src/clients/v1.py @@ -306,3 +306,36 @@ def compare( "format": format, }, ) + + def similar( + self, + *, + since: Optional[str] = None, + from_time: Optional[str] = None, + to_time: Optional[str] = None, + service: Optional[str] = None, + env: Optional[str] = None, + ingestion_job_id: Optional[str] = None, + fingerprint: Optional[str] = None, + fingerprints: Optional[list[str]] = None, + top: int = 10, + cross_scope: Optional[bool] = None, + scope: Optional[str] = None, + ) -> dict[str, Any]: + """POST /v1/query/similar — prior incidents with nearby fingerprints.""" + return self._post( + "/v1/query/similar", + { + "since": since, + "from_time": from_time, + "to_time": to_time, + "service": service, + "env": env, + "ingestion_job_id": ingestion_job_id, + "fingerprint": fingerprint, + "fingerprints": fingerprints, + "top": top, + "cross_scope": cross_scope, + "scope": scope, + }, + ) diff --git a/src/config/settings.py b/src/config/settings.py index 0e14166..7b05945 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -24,6 +24,9 @@ class Settings(BaseSettings): ask_semantic_top_k: int = 100 ask_semantic_min_similarity: float = 0.75 + # Similar-incident search (pgvector over cluster_embeddings; fingerprint fallback) + similar_semantic_min_similarity: float = 0.80 + llm_provider: Literal["disabled", "openai", "ollama"] = "disabled" llm_model: str = "gpt-4.1-mini" diff --git a/src/core/clustering/clusterer.py b/src/core/clustering/clusterer.py index efae84e..e6bd28b 100644 --- a/src/core/clustering/clusterer.py +++ b/src/core/clustering/clusterer.py @@ -11,7 +11,13 @@ from src.core.clustering.scoring import compute_importance_score from src.core.embeddings.provider import EmbeddingsProvider from src.core.normalization.patterns import is_trigger_message -from src.db.models import DEFAULT_LOG_SCOPE, Cluster, ClusterMember, ClusterRun, LogEntry +from src.db.models import ( + DEFAULT_LOG_SCOPE, + Cluster, + ClusterMember, + ClusterRun, + LogEntry, +) from src.db.scope_filter import filter_log_entries_by_scope from src.utils.time import resolve_baseline_window @@ -74,17 +80,21 @@ def run_clustering( rows = db.execute(q).all() if not rows: - cluster_run = _create_cluster_run(db, window_start, window_end, service, environment, save=save_to_db) + cluster_run = _create_cluster_run( + db, window_start, window_end, service, environment, save=save_to_db + ) return cluster_run, [] # 2. Group by fingerprint - groups: dict[str, dict] = defaultdict(lambda: { - "messages": [], - "services": defaultdict(int), - "levels": defaultdict(int), - "timestamps": [], - "ids": [], - }) + groups: dict[str, dict] = defaultdict( + lambda: { + "messages": [], + "services": defaultdict(int), + "levels": defaultdict(int), + "timestamps": [], + "ids": [], + } + ) for row in rows: fp = row.fingerprint @@ -106,7 +116,9 @@ def run_clustering( if ingestion_job_id: baseline_counts: dict[str, int] = {} else: - baseline_start, baseline_end = resolve_baseline_window(window_start, window_end, baseline_window_str) + baseline_start, baseline_end = resolve_baseline_window( + window_start, window_end, baseline_window_str + ) baseline_counts = get_baseline_counts( db, baseline_start, @@ -132,6 +144,7 @@ def run_clustering( rep_msg = "" if g["messages"]: from collections import Counter + rep_msg = Counter(g["messages"]).most_common(1)[0][0] is_trigger = is_trigger_message(rep_msg) @@ -144,20 +157,22 @@ def run_clustering( is_trigger_correlated=False, # refined below after sorting ) - clusters.append(ClusterData( - fingerprint=fp, - representative_message=rep_msg, - count=count, - services=services, - levels=levels, - first_seen=timestamps[0] if timestamps else None, - last_seen=timestamps[-1] if timestamps else None, - baseline_count=baseline_count, - change_ratio=change_ratio, - importance_score=importance, - is_trigger=is_trigger, - log_entry_ids=g["ids"], - )) + clusters.append( + ClusterData( + fingerprint=fp, + representative_message=rep_msg, + count=count, + services=services, + levels=levels, + first_seen=timestamps[0] if timestamps else None, + last_seen=timestamps[-1] if timestamps else None, + baseline_count=baseline_count, + change_ratio=change_ratio, + importance_score=importance, + is_trigger=is_trigger, + log_entry_ids=g["ids"], + ) + ) # 5. Optional semantic merge, then rank and cap top_clusters, algorithm = rank_and_merge_clusters(clusters, max_clusters) @@ -176,6 +191,8 @@ def run_clustering( if save_to_db: _persist_clusters(db, cluster_run, top_clusters) + _maybe_persist_cluster_embeddings(db, top_clusters, scope) + return cluster_run, top_clusters @@ -222,13 +239,17 @@ def _create_cluster_run( return run -def _persist_clusters(db: Session, cluster_run: ClusterRun, clusters: list[ClusterData]) -> None: +def _persist_clusters( + db: Session, cluster_run: ClusterRun, clusters: list[ClusterData] +) -> None: for cd in clusters: cluster = Cluster( id=uuid.uuid4(), cluster_run_id=cluster_run.id, cluster_key=cd.fingerprint, - representative_message=cd.representative_message[:2048] if cd.representative_message else None, + representative_message=cd.representative_message[:2048] + if cd.representative_message + else None, fingerprint=cd.fingerprint, count=cd.count, services_json=list(cd.services.keys()), @@ -253,3 +274,25 @@ def _persist_clusters(db: Session, cluster_run: ClusterRun, clusters: list[Clust db.add(member) db.flush() + + +def _maybe_persist_cluster_embeddings( + db: Session, + clusters: list[ClusterData], + scope: str, +) -> None: + """Upsert cluster template vectors. Fail-open so clustering never raises.""" + if not clusters: + return + try: + from src.core.embeddings.store import persist_cluster_embeddings + + persist_cluster_embeddings(db, clusters, scope=scope) + except Exception: + import structlog + + structlog.get_logger().warning( + "cluster_embeddings_hook_failed", + count=len(clusters), + exc_info=True, + ) diff --git a/src/core/embeddings/__init__.py b/src/core/embeddings/__init__.py index 16e5147..990459e 100644 --- a/src/core/embeddings/__init__.py +++ b/src/core/embeddings/__init__.py @@ -8,7 +8,10 @@ from src.core.embeddings.store import ( STORED_EMBEDDING_DIMS, build_embedding_rows, + cluster_embedding_row_values, + cluster_embedding_upsert_statement, ingest_embeddings_provider, + persist_cluster_embeddings, persist_log_embeddings, ) @@ -19,7 +22,10 @@ "OpenAIEmbeddingsProvider", "STORED_EMBEDDING_DIMS", "build_embedding_rows", + "cluster_embedding_row_values", + "cluster_embedding_upsert_statement", "get_embeddings_provider", "ingest_embeddings_provider", + "persist_cluster_embeddings", "persist_log_embeddings", ] diff --git a/src/core/embeddings/store.py b/src/core/embeddings/store.py index 2901096..94fce52 100644 --- a/src/core/embeddings/store.py +++ b/src/core/embeddings/store.py @@ -1,21 +1,34 @@ -"""Persist log-line embeddings for semantic retrieval. +"""Persist log-line and cluster-template embeddings for semantic retrieval. Ingest writes ``LogEmbedding`` rows after ``LogEntry`` PKs exist. -Ask reads those vectors via pgvector cosine similarity. The stored -column is ``Vector(1536)`` (migration 0001); other dimensions are skipped. +Clustering upserts ``ClusterEmbedding`` rows keyed by ``(scope, fingerprint)``. +Ask reads log-line vectors; similar-incident search reads cluster vectors. +The stored column is ``Vector(1536)`` (migration 0001); other dimensions are skipped. """ from __future__ import annotations import uuid +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Any, Optional import structlog +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.dialects.postgresql.dml import Insert from sqlalchemy.orm import Session from src.config import get_settings from src.config.settings import Settings from src.core.embeddings.provider import EmbeddingsProvider, get_embeddings_provider -from src.db.models import LogEmbedding, LogEntry +from src.db.models import ( + CLUSTER_EMBEDDING_UNIQUE, + ClusterEmbedding, + LogEmbedding, + LogEntry, +) + +if TYPE_CHECKING: + from src.core.clustering.clusterer import ClusterData log = structlog.get_logger() @@ -155,3 +168,130 @@ def persist_log_embeddings( log.warning("embeddings_persist_failed", count=len(rows), exc_info=True) return 0 return len(rows) + + +def cluster_embedding_row_values( + *, + scope: str, + fingerprint: str, + template: str, + vector: list[float], + model_name: str, + first_seen: Optional[datetime] = None, + last_seen: Optional[datetime] = None, + count: int = 0, + row_id: Optional[uuid.UUID] = None, + expected_dimensions: int = STORED_EMBEDDING_DIMS, +) -> Optional[dict[str, Any]]: + """Build an upsert payload for one cluster template. ``None`` if unusable.""" + if not fingerprint or not template.strip(): + return None + if len(vector) != expected_dimensions: + return None + now = datetime.now(tz=timezone.utc) + return { + "id": row_id or uuid.uuid4(), + "scope": scope, + "fingerprint": fingerprint, + "template": template, + "embedding": vector, + "model_name": model_name, + "first_seen": first_seen, + "last_seen": last_seen, + "count": count, + "updated_at": now, + } + + +def cluster_embedding_upsert_statement(rows: list[dict[str, Any]]) -> Insert: + """INSERT … ON CONFLICT (scope, fingerprint) DO UPDATE.""" + stmt = pg_insert(ClusterEmbedding).values(rows) + excluded = stmt.excluded + return stmt.on_conflict_do_update( + constraint=CLUSTER_EMBEDDING_UNIQUE, + set_={ + "template": excluded.template, + "embedding": excluded.embedding, + "model_name": excluded.model_name, + "first_seen": excluded.first_seen, + "last_seen": excluded.last_seen, + "count": excluded.count, + "updated_at": excluded.updated_at, + }, + ) + + +def persist_cluster_embeddings( + db: Session, + clusters: list["ClusterData"], + *, + scope: str, + provider: Optional[EmbeddingsProvider] = None, + settings: Optional[Settings] = None, +) -> int: + """Embed cluster templates and upsert ``cluster_embeddings``. + + Fail-open: provider down, dimension mismatch, or write errors log and + return 0 so the cluster run still succeeds. Writes run in a savepoint. + """ + if not clusters: + return 0 + if settings is None: + settings = get_settings() + if settings.embeddings_dimensions != STORED_EMBEDDING_DIMS: + return 0 + if provider is None: + provider = get_embeddings_provider(settings) + if not provider.is_available(): + return 0 + + eligible: list[Any] = [] + texts: list[str] = [] + for cluster in clusters: + template = (cluster.representative_message or "").strip() + if not template or not cluster.fingerprint: + continue + eligible.append(cluster) + texts.append(template) + if not texts: + return 0 + + try: + vectors = provider.embed_texts(texts) + except Exception: + log.warning("cluster_embeddings_batch_failed", count=len(texts), exc_info=True) + return 0 + + if len(vectors) != len(eligible): + log.warning( + "cluster_embeddings_count_mismatch", + expected=len(eligible), + got=len(vectors), + ) + return 0 + + rows: list[dict[str, Any]] = [] + for cluster, vector in zip(eligible, vectors): + payload = cluster_embedding_row_values( + scope=scope, + fingerprint=cluster.fingerprint, + template=cluster.representative_message, + vector=vector, + model_name=settings.embeddings_model, + first_seen=cluster.first_seen, + last_seen=cluster.last_seen, + count=cluster.count, + ) + if payload is not None: + rows.append(payload) + if not rows: + return 0 + + try: + with db.begin_nested(): + db.execute(cluster_embedding_upsert_statement(rows)) + db.flush() + except Exception: + log.warning("cluster_embeddings_persist_failed", count=len(rows), exc_info=True) + return 0 + return len(rows) diff --git a/src/core/retrieval/similar.py b/src/core/retrieval/similar.py new file mode 100644 index 0000000..98f29a4 --- /dev/null +++ b/src/core/retrieval/similar.py @@ -0,0 +1,459 @@ +"""Cross-incident similar-fingerprint search (G11 / POST /v1/query/similar). + +Semantic path: pgvector cosine NN over ``cluster_embeddings``. +Fingerprint fallback: exact fingerprint match when the embeddings provider +is down, vectors are missing, or ANN returns nothing. Never raises to the +caller — empty matches with ``retrieval_mode="fingerprint"`` on failure. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Optional + +import structlog +from sqlalchemy import Select, func, select +from sqlalchemy.orm import Session + +from src.config import get_settings +from src.config.settings import Settings +from src.core.embeddings.provider import EmbeddingsProvider, get_embeddings_provider +from src.core.embeddings.store import STORED_EMBEDDING_DIMS +from src.db.models import DEFAULT_LOG_SCOPE, ClusterEmbedding, LogEntry + +log = structlog.get_logger() + +RETRIEVAL_SEMANTIC = "semantic" +RETRIEVAL_FINGERPRINT = "fingerprint" +FINGERPRINT_MATCH_SIMILARITY = 1.0 + + +@dataclass(frozen=True) +class SimilarVisibility: + """Which scopes a caller may see in similar-incident results.""" + + cross_scope: bool + # None = every scope is visible; otherwise SQL is pinned to this scope. + visible_scope: Optional[str] + + +@dataclass +class QueryCluster: + fingerprint: str + template: str = "" + + +@dataclass +class SimilarMatch: + scope: str + fingerprint: str + template: Optional[str] + similarity: float + first_seen: Optional[datetime] + last_seen: Optional[datetime] + count: int + + +@dataclass +class SimilarResult: + query_clusters: list[QueryCluster] + matches: list[SimilarMatch] + retrieval_mode: str + window_start: datetime + window_end: datetime + + +def resolve_similar_visibility( + *, + auth_enabled: bool, + resolved_scope: str, + cross_scope_requested: Optional[bool], + role: str = "", + allow_scope_override: bool = False, +) -> SimilarVisibility: + """Decide cross-scope vs same-scope for ``/similar``. + + * ``AUTH_ENABLED=false``: cross-scope allowed (default on). + * ``admin`` role: cross-scope by default; ``cross_scope=false`` pins. + * ``query`` with ``allow_scope_override``: cross-scope only when + ``cross_scope=true``. + * Pinned ``query`` (and OIDC): same-scope only, even if requested. + """ + if not auth_enabled: + cross = True if cross_scope_requested is None else bool(cross_scope_requested) + return SimilarVisibility( + cross_scope=cross, + visible_scope=None if cross else resolved_scope, + ) + + if role == "admin": + cross = True if cross_scope_requested is None else bool(cross_scope_requested) + return SimilarVisibility( + cross_scope=cross, + visible_scope=None if cross else resolved_scope, + ) + + if allow_scope_override: + cross = bool(cross_scope_requested) + return SimilarVisibility( + cross_scope=cross, + visible_scope=None if cross else resolved_scope, + ) + + return SimilarVisibility(cross_scope=False, visible_scope=resolved_scope) + + +def collect_query_fingerprints( + fingerprint: Optional[str], + fingerprints: Optional[list[str]], +) -> list[str]: + """Merge ``fingerprint`` / ``fingerprints`` and drop blanks, preserving order.""" + seen: set[str] = set() + out: list[str] = [] + for value in [fingerprint, *(fingerprints or [])]: + if not value: + continue + stripped = value.strip() + if not stripped or stripped in seen: + continue + seen.add(stripped) + out.append(stripped) + return out + + +def query_clusters_from_fingerprints( + fingerprints: list[str], + templates: Optional[dict[str, str]] = None, +) -> list[QueryCluster]: + """Build query clusters without re-clustering. Missing templates stay empty.""" + lookup = templates or {} + return [ + QueryCluster(fingerprint=fp, template=lookup.get(fp, "")) for fp in fingerprints + ] + + +def query_clusters_from_cluster_data( + clusters: list[object], *, limit: int = 5 +) -> list[QueryCluster]: + """Take the top clustered fingerprints (already importance-sorted) as the query.""" + out: list[QueryCluster] = [] + for cluster in clusters[:limit]: + fp = str(getattr(cluster, "fingerprint", "") or "") + if not fp: + continue + template = str(getattr(cluster, "representative_message", "") or "") + out.append(QueryCluster(fingerprint=fp, template=template)) + return out + + +def render_similar_summary(matches: list[SimilarMatch], retrieval_mode: str) -> str: + """Short rules-only prose for ``rendered_text``.""" + if not matches: + if retrieval_mode == RETRIEVAL_FINGERPRINT: + return "No similar incidents found (fingerprint match)." + return "No similar incidents found." + first = matches[0] + extra = f" and {len(matches) - 1} more" if len(matches) > 1 else "" + return ( + f"Saw fingerprint {first.fingerprint} in {first.scope} " + f"(similarity {first.similarity:.2f}){extra}." + ) + + +def apply_visibility_filters( + stmt: Select, + *, + query_scope: str, + query_fingerprints: list[str], + visibility: SimilarVisibility, + scope_column: object, + fingerprint_column: object, +) -> Select: + """Restrict a select to scopes the caller can see; drop the query pairs.""" + if visibility.cross_scope: + if visibility.visible_scope is not None: + stmt = stmt.where(scope_column == visibility.visible_scope) + stmt = stmt.where(scope_column != query_scope) + else: + pinned = visibility.visible_scope or query_scope + stmt = stmt.where(scope_column == pinned) + if query_fingerprints: + stmt = stmt.where(fingerprint_column.notin_(query_fingerprints)) + return stmt + + +def cluster_embedding_semantic_statement( + query_vector: list[float], + *, + query_scope: str, + query_fingerprints: list[str], + visibility: SimilarVisibility, + min_similarity: float, + limit: int, +) -> Select: + """Compile-able ANN select over ``cluster_embeddings`` (cosine distance).""" + distance = ClusterEmbedding.embedding.cosine_distance(query_vector) + similarity = 1 - distance + stmt = select( + ClusterEmbedding.scope, + ClusterEmbedding.fingerprint, + ClusterEmbedding.template, + ClusterEmbedding.first_seen, + ClusterEmbedding.last_seen, + ClusterEmbedding.count, + similarity.label("similarity"), + ).where(similarity >= min_similarity) + stmt = apply_visibility_filters( + stmt, + query_scope=query_scope, + query_fingerprints=query_fingerprints, + visibility=visibility, + scope_column=ClusterEmbedding.scope, + fingerprint_column=ClusterEmbedding.fingerprint, + ) + return stmt.order_by(distance.asc()).limit(limit) + + +def cluster_embedding_fingerprint_statement( + fingerprints: list[str], + *, + query_scope: str, + visibility: SimilarVisibility, + limit: int, +) -> Select: + """Exact fingerprint match on ``cluster_embeddings``.""" + stmt = select( + ClusterEmbedding.scope, + ClusterEmbedding.fingerprint, + ClusterEmbedding.template, + ClusterEmbedding.first_seen, + ClusterEmbedding.last_seen, + ClusterEmbedding.count, + ).where(ClusterEmbedding.fingerprint.in_(fingerprints)) + stmt = apply_visibility_filters( + stmt, + query_scope=query_scope, + query_fingerprints=fingerprints, + visibility=visibility, + scope_column=ClusterEmbedding.scope, + fingerprint_column=ClusterEmbedding.fingerprint, + ) + return stmt.limit(limit) + + +def log_entry_fingerprint_statement( + fingerprints: list[str], + *, + query_scope: str, + visibility: SimilarVisibility, + limit: int, +) -> Select: + """Fingerprint-equality fallback over ``log_entries`` when no cluster vectors exist.""" + stmt = ( + select( + LogEntry.scope, + LogEntry.fingerprint, + func.min(LogEntry.normalized_message).label("template"), + func.min(LogEntry.timestamp).label("first_seen"), + func.max(LogEntry.timestamp).label("last_seen"), + func.count().label("count"), + ) + .where(LogEntry.fingerprint.in_(fingerprints)) + .where(LogEntry.fingerprint.isnot(None)) + .group_by(LogEntry.scope, LogEntry.fingerprint) + ) + stmt = apply_visibility_filters( + stmt, + query_scope=query_scope, + query_fingerprints=fingerprints, + visibility=visibility, + scope_column=LogEntry.scope, + fingerprint_column=LogEntry.fingerprint, + ) + return stmt.limit(limit) + + +def _row_to_match(row: object, similarity: float) -> Optional[SimilarMatch]: + scope = getattr(row, "scope", None) + fingerprint = getattr(row, "fingerprint", None) + if not scope or not fingerprint: + return None + return SimilarMatch( + scope=str(scope), + fingerprint=str(fingerprint), + template=getattr(row, "template", None), + similarity=float(similarity), + first_seen=getattr(row, "first_seen", None), + last_seen=getattr(row, "last_seen", None), + count=int(getattr(row, "count", 0) or 0), + ) + + +def _merge_matches( + existing: dict[tuple[str, str], SimilarMatch], match: SimilarMatch +) -> None: + key = (match.scope, match.fingerprint) + previous = existing.get(key) + if previous is None or match.similarity > previous.similarity: + existing[key] = match + + +def _ranked( + matches: dict[tuple[str, str], SimilarMatch], limit: int +) -> list[SimilarMatch]: + ranked = sorted(matches.values(), key=lambda m: m.similarity, reverse=True) + return ranked[:limit] + + +def search_similar_semantic( + db: Session, + query_clusters: list[QueryCluster], + *, + query_scope: str, + visibility: SimilarVisibility, + provider: EmbeddingsProvider, + min_similarity: float, + limit: int, +) -> list[SimilarMatch]: + """Embed query templates and ANN-search ``cluster_embeddings``.""" + texts = [c.template.strip() or c.fingerprint for c in query_clusters] + vectors = provider.embed_texts(texts) + if len(vectors) != len(query_clusters): + return [] + fingerprints = [c.fingerprint for c in query_clusters] + merged: dict[tuple[str, str], SimilarMatch] = {} + for vector in vectors: + if len(vector) != STORED_EMBEDDING_DIMS: + continue + stmt = cluster_embedding_semantic_statement( + vector, + query_scope=query_scope, + query_fingerprints=fingerprints, + visibility=visibility, + min_similarity=min_similarity, + limit=limit, + ) + for row in db.execute(stmt).all(): + similarity = float(getattr(row, "similarity", 0.0) or 0.0) + match = _row_to_match(row, similarity) + if match is not None: + _merge_matches(merged, match) + return _ranked(merged, limit) + + +def search_similar_fingerprint( + db: Session, + query_clusters: list[QueryCluster], + *, + query_scope: str, + visibility: SimilarVisibility, + limit: int, +) -> list[SimilarMatch]: + """Exact fingerprint match on cluster embeddings, then log entries.""" + fingerprints = [c.fingerprint for c in query_clusters if c.fingerprint] + if not fingerprints: + return [] + merged: dict[tuple[str, str], SimilarMatch] = {} + stmt = cluster_embedding_fingerprint_statement( + fingerprints, + query_scope=query_scope, + visibility=visibility, + limit=limit, + ) + for row in db.execute(stmt).all(): + match = _row_to_match(row, FINGERPRINT_MATCH_SIMILARITY) + if match is not None: + _merge_matches(merged, match) + if merged: + return _ranked(merged, limit) + + fallback = log_entry_fingerprint_statement( + fingerprints, + query_scope=query_scope, + visibility=visibility, + limit=limit, + ) + for row in db.execute(fallback).all(): + match = _row_to_match(row, FINGERPRINT_MATCH_SIMILARITY) + if match is not None: + _merge_matches(merged, match) + return _ranked(merged, limit) + + +def find_similar_incidents( + db: Session, + query_clusters: list[QueryCluster], + *, + query_scope: str = DEFAULT_LOG_SCOPE, + visibility: Optional[SimilarVisibility] = None, + window_start: Optional[datetime] = None, + window_end: Optional[datetime] = None, + top: int = 10, + provider: Optional[EmbeddingsProvider] = None, + settings: Optional[Settings] = None, +) -> SimilarResult: + """Return nearby historical fingerprints. Never raises.""" + now = datetime.now(tz=timezone.utc) + start = window_start or now + end = window_end or now + vis = visibility or SimilarVisibility(cross_scope=False, visible_scope=query_scope) + empty = SimilarResult( + query_clusters=list(query_clusters), + matches=[], + retrieval_mode=RETRIEVAL_FINGERPRINT, + window_start=start, + window_end=end, + ) + if not query_clusters: + return empty + + if settings is None: + settings = get_settings() + if provider is None: + provider = get_embeddings_provider(settings) + + limit = max(1, top) + try: + if ( + provider.is_available() + and settings.embeddings_dimensions == STORED_EMBEDDING_DIMS + ): + matches = search_similar_semantic( + db, + query_clusters, + query_scope=query_scope, + visibility=vis, + provider=provider, + min_similarity=settings.similar_semantic_min_similarity, + limit=limit, + ) + if matches: + return SimilarResult( + query_clusters=list(query_clusters), + matches=matches, + retrieval_mode=RETRIEVAL_SEMANTIC, + window_start=start, + window_end=end, + ) + except Exception: + log.warning("similar_semantic_failed", exc_info=True) + + try: + matches = search_similar_fingerprint( + db, + query_clusters, + query_scope=query_scope, + visibility=vis, + limit=limit, + ) + return SimilarResult( + query_clusters=list(query_clusters), + matches=matches, + retrieval_mode=RETRIEVAL_FINGERPRINT, + window_start=start, + window_end=end, + ) + except Exception: + log.warning("similar_fingerprint_failed", exc_info=True) + return empty diff --git a/src/db/__init__.py b/src/db/__init__.py index dfee273..a8839d8 100644 --- a/src/db/__init__.py +++ b/src/db/__init__.py @@ -3,6 +3,7 @@ AppConfig, Base, Cluster, + ClusterEmbedding, ClusterMember, ClusterRun, Explanation, @@ -23,6 +24,7 @@ "LogEmbedding", "ClusterRun", "Cluster", + "ClusterEmbedding", "ClusterMember", "Explanation", "ApiKey", diff --git a/src/db/models.py b/src/db/models.py index daa1c14..0906971 100644 --- a/src/db/models.py +++ b/src/db/models.py @@ -12,6 +12,7 @@ Integer, String, Text, + UniqueConstraint, text, ) from sqlalchemy.dialects.postgresql import JSONB, UUID @@ -137,6 +138,35 @@ class LogEmbedding(Base): log_entry: Mapped["LogEntry"] = relationship("LogEntry", back_populates="embedding") +CLUSTER_EMBEDDING_UNIQUE = "ux_cluster_embeddings_scope_fingerprint" + + +class ClusterEmbedding(Base): + """Cluster-template vector keyed by ``(scope, fingerprint)`` for similar-incident search.""" + + __tablename__ = "cluster_embeddings" + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + scope: Mapped[str] = mapped_column( + String(255), nullable=False, default=DEFAULT_LOG_SCOPE, server_default="default" + ) + fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) + template: Mapped[str | None] = mapped_column(Text, nullable=True) + embedding: Mapped[Any] = mapped_column(Vector(1536), nullable=False) + model_name: Mapped[str] = mapped_column(String(255), nullable=False) + first_seen: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_seen: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + __table_args__ = ( + UniqueConstraint("scope", "fingerprint", name=CLUSTER_EMBEDDING_UNIQUE), + Index("ix_cluster_embeddings_fingerprint", "fingerprint"), + Index("ix_cluster_embeddings_scope", "scope"), + ) + + class ClusterRun(Base): __tablename__ = "cluster_runs" diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index 56ee953..7629e8c 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -981,6 +981,105 @@ def test_invalid_ingestion_job_id_returns_400(self): assert resp.status_code == 400 +# ── POST /query/similar ─────────────────────────────────────────────────────── + +class TestSimilarEndpoint: + def _result(self, *, mode: str = "semantic"): + from datetime import datetime as dt + from src.core.retrieval.similar import QueryCluster, SimilarMatch, SimilarResult + + start = dt(2026, 3, 12, 13, 0, 0, tzinfo=timezone.utc) + end = dt(2026, 3, 12, 14, 0, 0, tzinfo=timezone.utc) + return SimilarResult( + query_clusters=[QueryCluster(fingerprint="abc123", template="connection refused")], + matches=[ + SimilarMatch( + scope="incident:INC-1188", + fingerprint="abc123", + template="connection refused", + similarity=0.94, + first_seen=start, + last_seen=end, + count=12, + ) + ], + retrieval_mode=mode, + window_start=start, + window_end=end, + ) + + def test_v1_and_unversioned_alias(self): + mock_db = _ctx_db() + result = self._result() + with patch("src.db.session.get_db", side_effect=lambda: mock_db), \ + patch( + "src.core.retrieval.similar.find_similar_incidents", + return_value=result, + ): + versioned = client.post( + "/v1/query/similar", + json={"fingerprint": "abc123", "since": "1h"}, + ) + unversioned = client.post( + "/query/similar", + json={"fingerprint": "abc123", "since": "1h"}, + ) + + assert versioned.status_code == 200 + assert unversioned.status_code == 200 + assert versioned.json() == unversioned.json() + body = versioned.json() + assert body["schema_version"] == "1.0" + assert body["retrieval_mode"] == "semantic" + assert body["matches"][0]["scope"] == "incident:INC-1188" + assert set(body["llm"]) >= {"used", "provider", "model", "fell_back"} + assert body["llm"]["used"] is False + assert unversioned.headers.get("deprecation") == "true" + assert versioned.headers.get("deprecation") is None + + def test_g7_fields_present(self): + mock_db = _ctx_db() + with patch("src.db.session.get_db", side_effect=lambda: mock_db), \ + patch( + "src.core.retrieval.similar.find_similar_incidents", + return_value=self._result(mode="fingerprint"), + ): + resp = client.post("/v1/query/similar", json={"fingerprint": "abc123"}) + + assert resp.status_code == 200 + data = resp.json() + for key in ("schema_version", "scope", "window", "query_clusters", "matches", "retrieval_mode", "llm"): + assert key in data + assert data["schema_version"] == "1.0" + assert data["window"]["from"] + assert data["window"]["to"] + assert data["llm"]["fell_back"] is False + assert data["rendered_text"] + + def test_degradation_never_500(self): + mock_db = _ctx_db() + with patch("src.db.session.get_db", side_effect=lambda: mock_db), \ + patch( + "src.core.clustering.clusterer.run_clustering", + side_effect=RuntimeError("db exploded"), + ): + resp = client.post("/v1/query/similar", json={"since": "1h"}) + + assert resp.status_code == 200 + data = resp.json() + assert data["retrieval_mode"] == "fingerprint" + assert data["matches"] == [] + assert data["schema_version"] == "1.0" + assert data["llm"]["used"] is False + + def test_invalid_ingestion_job_id_returns_400(self): + resp = client.post("/v1/query/similar", json={ + "since": "1h", + "ingestion_job_id": "not-a-uuid", + }) + assert resp.status_code == 400 + + # ── POST /query/timeline ───────────────────────────────────────────────────── class TestTimelineEndpoint: diff --git a/tests/unit/test_embeddings_store.py b/tests/unit/test_embeddings_store.py index a771eba..fc232c6 100644 --- a/tests/unit/test_embeddings_store.py +++ b/tests/unit/test_embeddings_store.py @@ -8,7 +8,10 @@ from src.core.embeddings.store import ( STORED_EMBEDDING_DIMS, build_embedding_rows, + cluster_embedding_row_values, + cluster_embedding_upsert_statement, ingest_embeddings_provider, + persist_cluster_embeddings, persist_log_embeddings, ) from src.db.models import LogEntry @@ -53,7 +56,9 @@ class TestBuildEmbeddingRows: def test_sets_model_name_and_vector(self) -> None: entry = _entry("connection refused") vector = [0.1] * STORED_EMBEDDING_DIMS - rows = build_embedding_rows([entry], [vector], model_name="text-embedding-3-small") + rows = build_embedding_rows( + [entry], [vector], model_name="text-embedding-3-small" + ) assert len(rows) == 1 assert rows[0].log_entry_id == entry.id @@ -158,3 +163,146 @@ def test_insert_failure_returns_zero_without_raising(self) -> None: assert count == 0 db.begin_nested.assert_called() + + +class TestClusterEmbeddingUpsert: + def test_row_values_sets_scope_fingerprint_and_vector(self) -> None: + vector = [0.2] * STORED_EMBEDDING_DIMS + row = cluster_embedding_row_values( + scope="incident:A", + fingerprint="abc123", + template="connection refused", + vector=vector, + model_name="text-embedding-3-small", + count=4, + ) + assert row is not None + assert row["scope"] == "incident:A" + assert row["fingerprint"] == "abc123" + assert row["template"] == "connection refused" + assert row["embedding"] == vector + assert row["model_name"] == "text-embedding-3-small" + assert row["count"] == 4 + + def test_row_values_skips_wrong_dimension(self) -> None: + assert ( + cluster_embedding_row_values( + scope="default", + fingerprint="abc", + template="msg", + vector=[0.1, 0.2], + model_name="m", + ) + is None + ) + + def test_row_values_skips_blank_template(self) -> None: + assert ( + cluster_embedding_row_values( + scope="default", + fingerprint="abc", + template=" ", + vector=[0.1] * STORED_EMBEDDING_DIMS, + model_name="m", + ) + is None + ) + + def test_upsert_statement_compiles_on_conflict_do_update(self) -> None: + from sqlalchemy.dialects import postgresql + + vector = [0.05] * STORED_EMBEDDING_DIMS + row = cluster_embedding_row_values( + scope="incident:A", + fingerprint="abc123", + template="connection refused", + vector=vector, + model_name="m", + ) + assert row is not None + stmt = cluster_embedding_upsert_statement([row]) + compiled = str(stmt.compile(dialect=postgresql.dialect())).upper() + assert "ON CONFLICT" in compiled + assert "DO UPDATE" in compiled + assert "SCOPE" in compiled + assert "FINGERPRINT" in compiled + assert "UX_CLUSTER_EMBEDDINGS_SCOPE_FINGERPRINT" in compiled + + def test_persist_upserts_when_provider_available(self) -> None: + from src.core.clustering.clusterer import ClusterData + + cluster = ClusterData( + fingerprint="abc123", + representative_message="connection refused", + count=3, + services={"api": 3}, + levels={"error": 3}, + first_seen=None, + last_seen=None, + baseline_count=0, + change_ratio=1.0, + importance_score=1.0, + ) + provider = FakeEmbeddingsProvider([[0.05] * STORED_EMBEDDING_DIMS]) + db = MagicMock() + settings = _settings(embeddings_model="text-embedding-3-small") + + count = persist_cluster_embeddings( + db, [cluster], scope="incident:A", provider=provider, settings=settings + ) + + assert count == 1 + db.execute.assert_called_once() + assert provider.calls == [["connection refused"]] + + def test_persist_disabled_provider_is_noop(self) -> None: + from src.core.clustering.clusterer import ClusterData + + cluster = ClusterData( + fingerprint="abc123", + representative_message="connection refused", + count=1, + services={}, + levels={}, + first_seen=None, + last_seen=None, + baseline_count=0, + change_ratio=1.0, + importance_score=1.0, + ) + db = MagicMock() + count = persist_cluster_embeddings( + db, + [cluster], + scope="default", + provider=DisabledEmbeddingsProvider(), + settings=_settings(), + ) + assert count == 0 + db.execute.assert_not_called() + + def test_persist_provider_raise_skips_without_raising(self) -> None: + from src.core.clustering.clusterer import ClusterData + + cluster = ClusterData( + fingerprint="abc123", + representative_message="connection refused", + count=1, + services={}, + levels={}, + first_seen=None, + last_seen=None, + baseline_count=0, + change_ratio=1.0, + importance_score=1.0, + ) + db = MagicMock() + count = persist_cluster_embeddings( + db, + [cluster], + scope="default", + provider=FakeEmbeddingsProvider([], fail=True), + settings=_settings(), + ) + assert count == 0 + db.execute.assert_not_called() diff --git a/tests/unit/test_openapi_contract.py b/tests/unit/test_openapi_contract.py index 29cdbcd..3731615 100644 --- a/tests/unit/test_openapi_contract.py +++ b/tests/unit/test_openapi_contract.py @@ -25,6 +25,7 @@ "/v1/query/timeline", "/v1/query/compare", "/v1/query/clusters", + "/v1/query/similar", "/v1/ingestions", "/v1/ingestions/lines", "/v1/ingestions/{ingestion_job_id}:pause", @@ -59,6 +60,8 @@ def test_unversioned_aliases_are_deprecated_in_schema() -> None: paths = app.openapi()["paths"] assert "/query/explain" in paths assert paths["/query/explain"]["post"].get("deprecated") is True + assert "/query/similar" in paths + assert paths["/query/similar"]["post"].get("deprecated") is True assert paths["/ingestions"]["post"].get("deprecated") is True assert paths["/config"]["get"].get("deprecated") is True diff --git a/tests/unit/test_query_schema.py b/tests/unit/test_query_schema.py index 4322020..d07e30e 100644 --- a/tests/unit/test_query_schema.py +++ b/tests/unit/test_query_schema.py @@ -15,6 +15,7 @@ ClustersResponse, CompareResponse, ExplainResponse, + SimilarResponse, TimelineResponse, explain_from_cached, short_summary, @@ -65,6 +66,7 @@ def test_published_jsonschema_matches_pydantic() -> None: "compare.v1.json": CompareResponse, "ask.v1.json": AskResponse, "clusters.v1.json": ClustersResponse, + "similar.v1.json": SimilarResponse, } for filename, model in mapping.items(): path = SCHEMA_DIR / filename @@ -148,3 +150,14 @@ def test_explain_and_unversioned_share_v1_body() -> None: assert unversioned.json() == versioned.json() ExplainResponse.model_validate(versioned.json()) assert versioned.json()["schema_version"] == "1.0" + + +def test_similar_openapi_documents_response_model() -> None: + schema = app.openapi() + similar = schema["paths"]["/v1/query/similar"]["post"] + ref = similar["responses"]["200"]["content"]["application/json"]["schema"] + assert "$ref" in ref + assert "SimilarResponse" in ref["$ref"] + components = schema["components"]["schemas"] + assert "SimilarResponse" in components + assert "SimilarMatchModel" in components diff --git a/tests/unit/test_settings.py b/tests/unit/test_settings.py index 7aa6369..c0f06d8 100644 --- a/tests/unit/test_settings.py +++ b/tests/unit/test_settings.py @@ -43,6 +43,14 @@ def test_ask_semantic_settings_from_env(monkeypatch): assert settings.ask_semantic_min_similarity == pytest.approx(0.6) +def test_similar_semantic_settings_from_env(monkeypatch): + monkeypatch.setenv("SIMILAR_SEMANTIC_MIN_SIMILARITY", "0.85") + + settings = Settings(_env_file=None) + + assert settings.similar_semantic_min_similarity == pytest.approx(0.85) + + def test_ignores_legacy_raglogs_prefix(monkeypatch): monkeypatch.setenv("RAGLOGS_LLM_PROVIDER", "ollama") monkeypatch.setenv("RAGLOGS_DB_URL", "postgresql+psycopg://legacy/db") diff --git a/tests/unit/test_similar.py b/tests/unit/test_similar.py new file mode 100644 index 0000000..e292db0 --- /dev/null +++ b/tests/unit/test_similar.py @@ -0,0 +1,455 @@ +"""Unit tests for similar-incident search (no database).""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +from src.core.embeddings.provider import DisabledEmbeddingsProvider +from src.core.embeddings.store import STORED_EMBEDDING_DIMS +from src.core.retrieval.similar import ( + QueryCluster, + SimilarMatch, + SimilarVisibility, + apply_visibility_filters, + cluster_embedding_fingerprint_statement, + cluster_embedding_semantic_statement, + collect_query_fingerprints, + find_similar_incidents, + log_entry_fingerprint_statement, + query_clusters_from_fingerprints, + render_similar_summary, + resolve_similar_visibility, + search_similar_fingerprint, +) +from src.db.models import ClusterEmbedding + +WINDOW_START = datetime(2026, 3, 12, 13, 0, 0, tzinfo=timezone.utc) +WINDOW_END = datetime(2026, 3, 12, 14, 0, 0, tzinfo=timezone.utc) +VECTOR = [0.01] * STORED_EMBEDDING_DIMS + + +def _compiled(stmt) -> str: + return str(stmt.compile(compile_kwargs={"literal_binds": False})).lower() + + +def _params(stmt) -> list[object]: + values: list[object] = [] + for value in stmt.compile(compile_kwargs={"literal_binds": False}).params.values(): + if ( + isinstance(value, (list, tuple)) + and value + and not isinstance(value[0], float) + ): + values.extend(value) + else: + values.append(value) + return values + + +def _row( + *, + scope: str = "incident:INC-1188", + fingerprint: str = "abc123", + template: str = "connection refused", + similarity: float = 0.91, + count: int = 4, +) -> MagicMock: + row = MagicMock() + row.scope = scope + row.fingerprint = fingerprint + row.template = template + row.first_seen = WINDOW_START + row.last_seen = WINDOW_END + row.count = count + row.similarity = similarity + return row + + +class FakeEmbeddingsProvider: + def __init__( + self, + vectors: list[list[float]], + *, + fail: bool = False, + available: bool = True, + ) -> None: + self.vectors = vectors + self.fail = fail + self.available = available + self.calls: list[list[str]] = [] + + def is_available(self) -> bool: + return self.available + + def embed_texts(self, texts: list[str]) -> list[list[float]]: + self.calls.append(list(texts)) + if self.fail: + raise RuntimeError("embedding backend down") + return self.vectors + + +class TestResolveSimilarVisibility: + def test_auth_off_defaults_to_cross_scope(self) -> None: + vis = resolve_similar_visibility( + auth_enabled=False, + resolved_scope="incident:A", + cross_scope_requested=None, + ) + assert vis.cross_scope is True + assert vis.visible_scope is None + + def test_auth_off_can_pin_same_scope(self) -> None: + vis = resolve_similar_visibility( + auth_enabled=False, + resolved_scope="incident:A", + cross_scope_requested=False, + ) + assert vis.cross_scope is False + assert vis.visible_scope == "incident:A" + + def test_admin_defaults_to_cross_scope(self) -> None: + vis = resolve_similar_visibility( + auth_enabled=True, + resolved_scope="incident:A", + cross_scope_requested=None, + role="admin", + ) + assert vis.cross_scope is True + assert vis.visible_scope is None + + def test_pinned_query_ignores_cross_scope_request(self) -> None: + vis = resolve_similar_visibility( + auth_enabled=True, + resolved_scope="incident:A", + cross_scope_requested=True, + role="query", + allow_scope_override=False, + ) + assert vis.cross_scope is False + assert vis.visible_scope == "incident:A" + + def test_override_query_requires_explicit_cross_scope(self) -> None: + vis = resolve_similar_visibility( + auth_enabled=True, + resolved_scope="incident:A", + cross_scope_requested=None, + role="query", + allow_scope_override=True, + ) + assert vis.cross_scope is False + vis_on = resolve_similar_visibility( + auth_enabled=True, + resolved_scope="incident:A", + cross_scope_requested=True, + role="query", + allow_scope_override=True, + ) + assert vis_on.cross_scope is True + assert vis_on.visible_scope is None + + +class TestVisibilitySQL: + def test_pinned_fingerprint_sql_filters_scope(self) -> None: + vis = SimilarVisibility(cross_scope=False, visible_scope="incident:A") + stmt = cluster_embedding_fingerprint_statement( + ["abc123"], + query_scope="incident:A", + visibility=vis, + limit=10, + ) + sql = _compiled(stmt) + params = _params(stmt) + assert "cluster_embeddings" in sql + assert "scope" in sql + assert "incident:a" in [str(p).lower() for p in params] + assert "abc123" in params + + def test_admin_cross_scope_excludes_query_scope(self) -> None: + vis = SimilarVisibility(cross_scope=True, visible_scope=None) + stmt = cluster_embedding_fingerprint_statement( + ["abc123"], + query_scope="incident:A", + visibility=vis, + limit=10, + ) + sql = _compiled(stmt) + params = _params(stmt) + assert "cluster_embeddings.scope" in sql + assert "incident:A" in params + assert "!=" in sql or "<>" in sql or " is not " in sql + + def test_semantic_sql_includes_cosine_and_scope(self) -> None: + vis = SimilarVisibility(cross_scope=False, visible_scope="incident:A") + stmt = cluster_embedding_semantic_statement( + VECTOR, + query_scope="incident:A", + query_fingerprints=["abc123"], + visibility=vis, + min_similarity=0.8, + limit=10, + ) + sql = _compiled(stmt) + where = str(stmt.whereclause).lower() + compiled = stmt.compile(compile_kwargs={"literal_binds": False}) + assert "cluster_embeddings" in sql + assert "<=>" in where + assert 0.8 in compiled.params.values() + assert "incident:A" in compiled.params.values() + assert any(p == "abc123" or p == ["abc123"] for p in compiled.params.values()) + + def test_log_entry_fallback_sql_includes_scope(self) -> None: + vis = SimilarVisibility(cross_scope=False, visible_scope="incident:A") + stmt = log_entry_fingerprint_statement( + ["abc123"], + query_scope="incident:A", + visibility=vis, + limit=10, + ) + sql = _compiled(stmt) + assert "log_entries" in sql + assert "log_entries.scope" in sql + assert "incident:A" in _params(stmt) + + def test_apply_visibility_never_omits_pinned_scope(self) -> None: + vis = SimilarVisibility(cross_scope=False, visible_scope="incident:A") + from sqlalchemy import select + + stmt = apply_visibility_filters( + select(ClusterEmbedding), + query_scope="incident:A", + query_fingerprints=["fp1"], + visibility=vis, + scope_column=ClusterEmbedding.scope, + fingerprint_column=ClusterEmbedding.fingerprint, + ) + sql = _compiled(stmt) + assert "cluster_embeddings.scope" in sql + assert "incident:A" in _params(stmt) + assert "fp1" in _params(stmt) + + +class TestFindSimilarDegradation: + def test_disabled_provider_uses_fingerprint(self) -> None: + db = MagicMock() + db.execute.return_value.all.return_value = [_row()] + result = find_similar_incidents( + db, + [QueryCluster(fingerprint="abc123", template="connection refused")], + query_scope="incident:A", + visibility=SimilarVisibility(cross_scope=True, visible_scope=None), + window_start=WINDOW_START, + window_end=WINDOW_END, + provider=DisabledEmbeddingsProvider(), + ) + assert result.retrieval_mode == "fingerprint" + assert result.matches[0].scope == "incident:INC-1188" + assert result.matches[0].similarity == 1.0 + + def test_provider_down_falls_back_without_raising(self) -> None: + db = MagicMock() + db.execute.return_value.all.return_value = [_row()] + provider = FakeEmbeddingsProvider([], fail=True, available=True) + result = find_similar_incidents( + db, + [QueryCluster(fingerprint="abc123", template="connection refused")], + query_scope="incident:A", + visibility=SimilarVisibility(cross_scope=True, visible_scope=None), + window_start=WINDOW_START, + window_end=WINDOW_END, + provider=provider, + ) + assert result.retrieval_mode == "fingerprint" + assert len(result.matches) == 1 + + def test_semantic_path_when_provider_returns_hits(self) -> None: + db = MagicMock() + db.execute.return_value.all.return_value = [_row(similarity=0.93)] + provider = FakeEmbeddingsProvider([VECTOR]) + result = find_similar_incidents( + db, + [QueryCluster(fingerprint="abc123", template="connection refused")], + query_scope="incident:A", + visibility=SimilarVisibility(cross_scope=True, visible_scope=None), + window_start=WINDOW_START, + window_end=WINDOW_END, + provider=provider, + ) + assert result.retrieval_mode == "semantic" + assert result.matches[0].similarity == 0.93 + assert provider.calls == [["connection refused"]] + + def test_empty_semantic_falls_back_to_fingerprint(self) -> None: + db = MagicMock() + db.execute.return_value.all.side_effect = [ + [], # semantic + [_row()], # fingerprint cluster_embeddings + ] + provider = FakeEmbeddingsProvider([VECTOR]) + result = find_similar_incidents( + db, + [QueryCluster(fingerprint="abc123", template="connection refused")], + query_scope="incident:A", + visibility=SimilarVisibility(cross_scope=True, visible_scope=None), + window_start=WINDOW_START, + window_end=WINDOW_END, + provider=provider, + ) + assert result.retrieval_mode == "fingerprint" + assert result.matches[0].fingerprint == "abc123" + + def test_db_errors_return_empty_fingerprint_result(self) -> None: + db = MagicMock() + db.execute.side_effect = RuntimeError("connection lost") + result = find_similar_incidents( + db, + [QueryCluster(fingerprint="abc123", template="x")], + query_scope="incident:A", + provider=DisabledEmbeddingsProvider(), + ) + assert result.retrieval_mode == "fingerprint" + assert result.matches == [] + + def test_fingerprint_falls_through_to_log_entries(self) -> None: + db = MagicMock() + db.execute.return_value.all.side_effect = [[], [_row()]] + matches = search_similar_fingerprint( + db, + [QueryCluster(fingerprint="abc123")], + query_scope="incident:A", + visibility=SimilarVisibility(cross_scope=True, visible_scope=None), + limit=10, + ) + assert len(matches) == 1 + assert db.execute.call_count == 2 + second_sql = _compiled(db.execute.call_args_list[1][0][0]) + assert "log_entries" in second_sql + + +class TestHelpers: + def test_collect_query_fingerprints_dedupes(self) -> None: + assert collect_query_fingerprints("abc", ["abc", "def", " "]) == ["abc", "def"] + + def test_query_clusters_from_fingerprints(self) -> None: + clusters = query_clusters_from_fingerprints(["a"], templates={"a": "msg"}) + assert clusters[0].template == "msg" + + def test_render_summary_empty(self) -> None: + text = render_similar_summary([], "fingerprint") + assert "No similar" in text + + def test_render_summary_with_match(self) -> None: + match = SimilarMatch( + scope="incident:INC-1188", + fingerprint="abc123", + template="x", + similarity=0.94, + first_seen=None, + last_seen=None, + count=1, + ) + text = render_similar_summary([match], "semantic") + assert "INC-1188" in text + assert "abc123" in text + + +class TestSimilarEndpointPermissions: + def _key( + self, role: str, *, scope: str, allow_scope_override: bool = False + ) -> MagicMock: + from uuid import uuid4 + + record = MagicMock() + record.id = uuid4() + record.name = "test" + record.key_prefix = "rlk_testhash" + record.role = role + record.scope = scope + record.allow_scope_override = allow_scope_override + record.revoked_at = None + return record + + def _settings(self): + from src.config.settings import Settings + + return Settings(_env_file=None, auth_enabled=True, auth_mode="api_key") + + def test_pinned_query_key_forces_same_scope(self) -> None: + from fastapi.testclient import TestClient + + from src.api.app import app + from src.core.retrieval.similar import SimilarResult + + result = SimilarResult( + query_clusters=[QueryCluster(fingerprint="abc123")], + matches=[], + retrieval_mode="fingerprint", + window_start=WINDOW_START, + window_end=WINDOW_END, + ) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + client = TestClient(app, raise_server_exceptions=False) + with ( + patch("src.config.get_settings", return_value=self._settings()), + patch( + "src.api.auth.keys.lookup_api_key", + return_value=self._key("query", scope="incident:A"), + ), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + patch( + "src.core.retrieval.similar.find_similar_incidents", + return_value=result, + ) as mock_find, + ): + resp = client.post( + "/v1/query/similar", + json={"fingerprint": "abc123", "cross_scope": True}, + headers={"Authorization": "Bearer rlk_queryrolexx"}, + ) + + assert resp.status_code == 200 + vis = mock_find.call_args.kwargs["visibility"] + assert vis.cross_scope is False + assert vis.visible_scope == "incident:A" + + def test_admin_key_cross_scope_by_default(self) -> None: + from fastapi.testclient import TestClient + + from src.api.app import app + from src.core.retrieval.similar import SimilarResult + + result = SimilarResult( + query_clusters=[QueryCluster(fingerprint="abc123")], + matches=[], + retrieval_mode="fingerprint", + window_start=WINDOW_START, + window_end=WINDOW_END, + ) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + client = TestClient(app, raise_server_exceptions=False) + with ( + patch("src.config.get_settings", return_value=self._settings()), + patch( + "src.api.auth.keys.lookup_api_key", + return_value=self._key("admin", scope="incident:A"), + ), + patch("src.db.session.get_db", side_effect=lambda: mock_db), + patch( + "src.core.retrieval.similar.find_similar_incidents", + return_value=result, + ) as mock_find, + ): + resp = client.post( + "/v1/query/similar", + json={"fingerprint": "abc123"}, + headers={"Authorization": "Bearer rlk_adminrolexx"}, + ) + + assert resp.status_code == 200 + vis = mock_find.call_args.kwargs["visibility"] + assert vis.cross_scope is True + assert vis.visible_scope is None diff --git a/tests/unit/test_v1_client.py b/tests/unit/test_v1_client.py index bacde69..f373a3c 100644 --- a/tests/unit/test_v1_client.py +++ b/tests/unit/test_v1_client.py @@ -47,6 +47,32 @@ def test_explain_targets_v1_and_returns_body() -> None: assert payload["llm"]["used"] is False +def test_similar_targets_v1() -> None: + from src.core.retrieval.similar import QueryCluster, SimilarResult + + start = datetime(2026, 3, 12, 13, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 3, 12, 14, 0, 0, tzinfo=timezone.utc) + result = SimilarResult( + query_clusters=[QueryCluster(fingerprint="abc123", template="x")], + matches=[], + retrieval_mode="fingerprint", + window_start=start, + window_end=end, + ) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + http = TestClient(app, raise_server_exceptions=False) + with patch("src.db.session.get_db", side_effect=lambda: mock_db), \ + patch("src.core.retrieval.similar.find_similar_incidents", return_value=result): + client = RaglogsClient(base_url="http://testserver", client=http) + payload = client.similar(fingerprint="abc123", since="1h") + + assert payload["schema_version"] == "1.0" + assert payload["retrieval_mode"] == "fingerprint" + assert payload["llm"]["used"] is False + + def test_health_uses_unversioned_path() -> None: http = TestClient(app, raise_server_exceptions=False) with patch("src.db.session.check_connection", return_value=False):