feat: isolate logs and jobs by resolved scope (#17)

README.md

@@ -280,6 +280,7 @@ raglogs ingest --adapter k8s ./node-logs.tar.gz
 | `--since` | Window for pull adapters, e.g. `30m`, `1h`, `24h` (default last 1h) |
 | `--from` / `--to` | Explicit ISO 8601 window bounds (pull adapters) |
 | `--resume-job` | Prior ingestion job UUID to resume pagination cursors from |
+| `--scope` | Isolation scope (CLI default `default`). Service requests require a resolvable scope. |
 
 **Loki**
 
@@ -636,14 +637,16 @@ Mint, list, and revoke HTTP API keys. The API bearer token is printed **once** a
 
 ```bash
 raglogs keys create --role query --scope default --name "ci"
+raglogs keys create --role query --scope incident:INC-9 --allow-scope-override --name "ci-override"
 raglogs keys list
 raglogs keys revoke <key-uuid>
 ```
 
 | Flag | Description |
 |---|---|
 | `--role` | `ingest`, `query`, or `admin` (default `query`) |
-| `--scope` | Stored on the key for later tenant isolation (G8). Default `default`. Not used to filter log queries yet. |
+| `--scope` | Pin the key to this isolation scope (enforced on every service read/write). Default `default`. Convention: `incident:<id>`, `service:<name>`, `env:<name>`. |
+| `--allow-scope-override` | Allow the caller to pass a request `scope` other than the key's pin. Pinned by default. |
 | `--name` | Optional label |
 
 Requires a migrated database (`raglogs init`). See [HTTP API authentication](#http-api-authentication).
@@ -945,7 +948,13 @@ A valid key with the wrong role returns **403**:
 {"error_code": "AUTH_FORBIDDEN", "message": "…"}
 ```
 
-Keys are stored argon2-hashed with a short indexed prefix. `scope` defaults to `default` and is stored for later isolation work — this release does **not** filter clusters or logs by scope. Each new key also gets a `whsec_…` webhook signing secret (shown once; `keys list` shows `whsec_****` only).
+Keys are stored argon2-hashed with a short indexed prefix. Each key is **pinned** to a `scope` (default `default`) and every service ingest/query is filtered by that scope — including baseline comparison — so one incident's logs cannot contaminate another. Mint with `--allow-scope-override` to let the caller pass a request `scope`. A service request with no resolvable scope returns **400**:
+
+```json
+{"error_code": "SCOPE_REQUIRED", "message": "…"}
+```
+
+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).
 
 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.
 
@@ -994,7 +1003,7 @@ curl -X POST http://localhost:8000/v1/ingestions/$ID:stop
 
 **Backpressure.** When pending worker jobs ≥ `INGEST_QUEUE_MAX` (default 100), `POST /v1/ingestions` and `POST /v1/ingestions/lines` return **429** with `Retry-After` (`INGEST_RETRY_AFTER_SECONDS`, default 5) and body `{"error_code":"INGEST_QUEUE_FULL","message":"..."}`. This is a queue-depth stand-in, not full API/LLM rate limiting.
 
-**Idempotency-Key.** `POST /v1/ingestions` (batch enqueue and tail create; also the deprecated `/ingestions` alias) honors an `Idempotency-Key` header (max 256 characters). A repeat within `INGEST_IDEMPOTENCY_TTL_SECONDS` (default 86400) returns the original **202** job — the same `worker_job_id` for batch, the same `ingestion_job_id` for tail — instead of starting a new one. Empty keys return **400**. GET routes ignore the header. `POST /v1/ingestions/lines` does not use the header; duplicate push/tail lines are handled by content dedup instead.
+**Idempotency-Key.** `POST /v1/ingestions` (batch enqueue and tail create; also the deprecated `/ingestions` alias) honors an `Idempotency-Key` header (max 256 characters). A repeat **in the same isolation scope** within `INGEST_IDEMPOTENCY_TTL_SECONDS` (default 86400) returns the original **202** job — the same `worker_job_id` for batch, the same `ingestion_job_id` for tail — instead of starting a new one. Reusing another scope's key returns **409** `IDEMPOTENCY_SCOPE_CONFLICT`. Empty keys return **400**. GET routes ignore the header. `POST /v1/ingestions/lines` does not use the header; duplicate push/tail lines are handled by content dedup instead.
 
 ```bash
 curl -X POST http://localhost:8000/v1/ingestions \
@@ -1003,11 +1012,11 @@ curl -X POST http://localhost:8000/v1/ingestions \
   -d '{"paths":["/var/log/app"]}'
 ```
 
-**Content dedup.** Every persist path (`ingest_files`, `ingest_from_source` including tail ticks, and push `/lines`) stores `original_line_hash` (SHA-256 of the **raw** line, distinct from the normalized fingerprint) and upserts on `(scope, source_ref, original_line_hash, timestamp)`. Re-reading the same physical lines is a no-op, so cluster counts stay stable across overlapping windows and tail/push retries. Missing `source_ref` is stored as `""` so uniqueness works (Postgres NULLs are distinct). `scope` defaults to `"default"`, or the API key's scope when a principal is present; queries are **not** filtered by scope yet. Duplicate lines are skipped, not errors.
+**Content dedup.** Every persist path (`ingest_files`, `ingest_from_source` including tail ticks, and push `/lines`) stores `original_line_hash` (SHA-256 of the **raw** line, distinct from the normalized fingerprint) and upserts on `(scope, source_ref, original_line_hash, timestamp)`. Re-reading the same physical lines is a no-op, so cluster counts stay stable across overlapping windows and tail/push retries. Missing `source_ref` is stored as `""` so uniqueness works (Postgres NULLs are distinct). `scope` defaults to `"default"` (CLI) or is resolved from the API key / request (service). Queries, ingest lists, and baseline comparison are filtered by the same scope. Duplicate lines are skipped, not errors.
 
 **Completion callbacks.** Optional `callback_url` on `POST /v1/ingestions` (http or https only; `file:` and empty hosts are rejected). When a **batch** worker job reaches a terminal state (`done` / `failed`), raglogs POSTs an HMAC-SHA256-signed JSON body to that URL. Delivery is fail-open: retries with jittered exponential backoff (`WEBHOOK_MAX_RETRIES`, default 5 extra attempts) on 5xx, 429, and connect errors; 4xx other than 429 are not retried. Failures are logged and **do not** change ingest status — poll `GET /v1/ingestions/jobs/{worker_job_id}` still works.
 
-`job_id` in the payload is the **ingestion_job_id** when ingest created a row; if the worker failed before that, it is the `worker_job_id`. `scope` is the authenticated key's scope, or `"default"` when auth is off. `counts.clusters` is `0` (clustering is not part of ingest). Worker `done` maps to `"succeeded"` (or `"partial"` when `error_count > 0`); `failed` maps to `"failed"`.
+`job_id` in the payload is the **ingestion_job_id** when ingest created a row; if the worker failed before that, it is the `worker_job_id`. `scope` is the resolved isolation scope (API key pin, request override when allowed, or `"default"` when auth is off). `counts.clusters` is `0` (clustering is not part of ingest). Worker `done` maps to `"succeeded"` (or `"partial"` when `error_count > 0`); `failed` maps to `"failed"`.
 
 ```bash
 curl -X POST http://localhost:8000/v1/ingestions \

migrations/versions/0008_scope_isolation.py

@@ -0,0 +1,53 @@
+"""scope isolation: ingestion_jobs.scope, log_entries indexes, key override flag
+
+Revision ID: 0008_scope_isolation
+Revises: 0007_ingest_idempotency
+Create Date: 2026-08-17 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0008_scope_isolation"
+down_revision: Union[str, None] = "0007_ingest_idempotency"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ingestion_jobs",
+        sa.Column("scope", sa.String(255), nullable=False, server_default="default"),
+    )
+    op.create_index(
+        "ix_log_entries_scope_timestamp",
+        "log_entries",
+        ["scope", "timestamp"],
+    )
+    op.create_index(
+        "ix_log_entries_scope_service_environment_fingerprint",
+        "log_entries",
+        ["scope", "service", "environment", "fingerprint"],
+    )
+    op.add_column(
+        "api_keys",
+        sa.Column(
+            "allow_scope_override",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.false(),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("api_keys", "allow_scope_override")
+    op.drop_index(
+        "ix_log_entries_scope_service_environment_fingerprint",
+        table_name="log_entries",
+    )
+    op.drop_index("ix_log_entries_scope_timestamp", table_name="log_entries")
+    op.drop_column("ingestion_jobs", "scope")

src/api/app.py

@@ -3,12 +3,13 @@
 from pathlib import Path
 import re
 
-from fastapi import APIRouter, FastAPI
-from fastapi.responses import ORJSONResponse
+from fastapi import APIRouter, FastAPI, Request
+from fastapi.responses import JSONResponse, ORJSONResponse
 from fastapi.routing import APIRoute
 from fastapi.staticfiles import StaticFiles
 
 from src.api.auth.middleware import AuthMiddleware
+from src.api.auth.scope import ScopeResolutionError, scope_error_response
 from src.api.deprecation import DeprecationHeaderMiddleware
 from src.api.routes import ask, clusters, compare_windows, config, explain, health, ingestions, timeline, ui
 
@@ -63,6 +64,13 @@ def generate(route: APIRoute) -> str:
 app.add_middleware(AuthMiddleware)
 app.add_middleware(DeprecationHeaderMiddleware)
 
+
+@app.exception_handler(ScopeResolutionError)
+async def handle_scope_resolution_error(
+    request: Request, exc: ScopeResolutionError
+) -> JSONResponse:
+    return scope_error_response(exc)
+
 app.include_router(health.router, tags=["health"])
 app.include_router(ui.router, tags=["ui"])
 

src/api/auth/keys.py

@@ -32,6 +32,7 @@ class ApiKeyInfo:
     revoked_at: datetime | None
     created_at: datetime | None
     webhook_secret_preview: str | None = None
+    allow_scope_override: bool = False
 
 
 @dataclass(frozen=True)
@@ -46,6 +47,7 @@ class ApiKeyRecord:
     scope: str
     revoked_at: datetime | None
     created_at: datetime | None = None
+    allow_scope_override: bool = False
 
 
 def generate_api_key() -> str:
@@ -127,6 +129,7 @@ def lookup_api_key(token: str) -> ApiKeyRecord | None:
                 scope=row.scope,
                 revoked_at=row.revoked_at,
                 created_at=row.created_at,
+                allow_scope_override=bool(getattr(row, "allow_scope_override", False)),
             )
             for row in rows
         ]
@@ -138,6 +141,7 @@ def create_api_key(
     role: str,
     scope: str = DEFAULT_SCOPE,
     name: str | None = None,
+    allow_scope_override: bool = False,
 ) -> tuple[str, str, ApiKeyInfo]:
     """Persist a hashed key and return (api_key, webhook_secret, metadata).
 
@@ -158,6 +162,7 @@ def create_api_key(
         role=role,
         scope=scope,
         name=name,
+        allow_scope_override=allow_scope_override,
     )
     return plaintext, webhook_secret, record
 
@@ -174,6 +179,7 @@ def _key_info(row: Any) -> ApiKeyInfo:
         webhook_secret_preview=mask_webhook_secret(
             getattr(row, "webhook_secret", None)
         ),
+        allow_scope_override=bool(getattr(row, "allow_scope_override", False)),
     )
 
 
@@ -184,6 +190,7 @@ def _persist_key(
     role: str,
     scope: str,
     name: str | None,
+    allow_scope_override: bool = False,
 ) -> ApiKeyInfo:
     from src.db.models import ApiKey
     from src.db.session import get_db
@@ -195,6 +202,7 @@ def _persist_key(
         role=role,
         scope=scope,
         webhook_secret=webhook_secret,
+        allow_scope_override=allow_scope_override,
     )
     with get_db() as db:
         db.add(row)

src/api/auth/middleware.py

@@ -24,6 +24,7 @@ class AuthPrincipal:
     auth_method: str
     key_id: str | None = None
     subject: str | None = None
+    allow_scope_override: bool = False
 
 
 def _auth_error(status_code: int, error_code: str, message: str) -> JSONResponse:
@@ -62,6 +63,7 @@ def _authenticate_token(token: str, settings: Any) -> AuthPrincipal | None:
             scope=principal.scope,
             auth_method="oidc",
             subject=principal.subject,
+            allow_scope_override=False,
         )
 
     if mode not in ("api_key", "both"):
@@ -76,6 +78,7 @@ def _authenticate_token(token: str, settings: Any) -> AuthPrincipal | None:
         scope=record.scope,
         auth_method="api_key",
         key_id=str(record.id),
+        allow_scope_override=getattr(record, "allow_scope_override", False) is True,
     )
 
 

src/api/auth/roles.py

@@ -1,7 +1,7 @@
 """Route → allowed role mapping.
 
-`admin` is included in every non-exempt set. Scope is stored on keys for later
-G8 isolation; this module does not filter log queries by scope.
+`admin` is included in every non-exempt set. Scope isolation (G8) is enforced
+in ``src.api.auth.scope`` and query filters, not in this role map.
 
 A leading `/v1` or `/v2` (any `/v<digits>`) is stripped before matching, so
 `POST /v1/ingestions` uses the same roles as `POST /ingestions`.

src/api/auth/scope.py

@@ -0,0 +1,137 @@
+"""Resolve the isolation scope for an HTTP request (G8).
+
+Service requests always end with a non-empty scope, or ``400 SCOPE_REQUIRED``.
+Pinned API keys (and OIDC principals) cannot switch to another scope.
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import Request
+from starlette.responses import JSONResponse
+
+from src.api.auth.middleware import AuthPrincipal
+from src.db.models import DEFAULT_LOG_SCOPE
+
+ERROR_SCOPE_REQUIRED = "SCOPE_REQUIRED"
+ERROR_SCOPE_MISMATCH = "SCOPE_MISMATCH"
+
+
+class ScopeResolutionError(Exception):
+    """Raised when a service request has no resolvable scope, or a pinned mismatch."""
+
+    def __init__(self, status_code: int, error_code: str, message: str) -> None:
+        self.status_code = status_code
+        self.error_code = error_code
+        self.message = message
+        super().__init__(message)
+
+
+def scope_error_response(exc: ScopeResolutionError) -> JSONResponse:
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"error_code": exc.error_code, "message": exc.message},
+    )
+
+
+def normalize_requested_scope(value: Optional[str]) -> Optional[str]:
+    """Strip whitespace. Empty / None means "not provided"."""
+    if value is None:
+        return None
+    stripped = value.strip()
+    return stripped or None
+
+
+def resolve_scope(
+    *,
+    requested_scope: Optional[str],
+    auth_enabled: bool,
+    principal: Optional[AuthPrincipal],
+) -> str:
+    """Return the scope to stamp and filter with.
+
+    Auth off: explicit request scope, else ``default`` (unauthenticated tests).
+    Pinned key / OIDC: always the principal scope; a different request scope
+    is ``403 SCOPE_MISMATCH``. Override-allowed keys use the request scope
+    when present, else the key scope. Empty after resolution → ``400``.
+    """
+    requested = normalize_requested_scope(requested_scope)
+
+    if not auth_enabled:
+        resolved = requested or DEFAULT_LOG_SCOPE
+        if not resolved:
+            raise ScopeResolutionError(
+                400,
+                ERROR_SCOPE_REQUIRED,
+                "A non-empty scope is required",
+            )
+        return resolved
+
+    if principal is None:
+        raise ScopeResolutionError(
+            400,
+            ERROR_SCOPE_REQUIRED,
+            "A resolvable scope is required for this request",
+        )
+
+    pinned = bool(
+        principal.auth_method == "oidc" or not principal.allow_scope_override
+    )
+    principal_scope = normalize_requested_scope(principal.scope)
+
+    if pinned:
+        if not principal_scope:
+            raise ScopeResolutionError(
+                400,
+                ERROR_SCOPE_REQUIRED,
+                "API key has no scope; mint a key with --scope",
+            )
+        if requested is not None and requested != principal_scope:
+            raise ScopeResolutionError(
+                403,
+                ERROR_SCOPE_MISMATCH,
+                "Request scope does not match the API key's pinned scope",
+            )
+        return principal_scope
+
+    resolved = requested or principal_scope
+    if not resolved:
+        raise ScopeResolutionError(
+            400,
+            ERROR_SCOPE_REQUIRED,
+            "Pass scope on the request or mint a key with a default scope",
+        )
+    return resolved
+
+
+def requested_scope_from_http(
+    request: Request,
+    body_scope: Optional[str] = None,
+) -> Optional[str]:
+    """Prefer a JSON-body ``scope``; otherwise the ``scope`` query parameter."""
+    from_body = normalize_requested_scope(body_scope)
+    if from_body is not None:
+        return from_body
+    return normalize_requested_scope(request.query_params.get("scope"))
+
+
+def bind_request_scope(
+    request: Request,
+    body_scope: Optional[str] = None,
+) -> str:
+    """Resolve scope, store it on ``request.state``, and return it."""
+    from src.config import get_settings
+
+    settings = get_settings()
+    principal = getattr(request.state, "auth_principal", None)
+    if principal is not None and not isinstance(principal, AuthPrincipal):
+        principal = None
+
+    resolved = resolve_scope(
+        requested_scope=requested_scope_from_http(request, body_scope),
+        auth_enabled=bool(settings.auth_enabled),
+        principal=principal,
+    )
+    request.state.resolved_scope = resolved
+    return resolved