Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,10 @@ Base URL: `http://localhost:8000`
| `POST` | `/v1/alerts/events/{id}/ack` | Mark an alert as seen |
| `POST` | `/v1/alerts/evaluate` | Evaluate all rules now |
| `POST` | `/v1/otlp/v1/traces` | Standard OTLP/HTTP trace receiver |
| `GET` | `/v1/privacy/projects` | Capture mode per project + what is stored |
| `PUT` | `/v1/privacy/projects/{id}` | Set a project's capture mode |
| `DELETE` | `/v1/privacy/projects/{id}` | Back to the path defaults |
| `POST` | `/v1/privacy/projects/{id}/enforce` | Rewrite stored data to the mode (`confirm`) |
| `GET` | `/v1/datasets` | List test-case datasets |
| `POST` | `/v1/datasets` | Create a dataset |
| `GET` | `/v1/datasets/{id}` | A dataset with its cases |
Expand Down Expand Up @@ -281,6 +285,18 @@ Scores of the previous rubric are replaced when the turn is re-scored. A backgro
backfills history. Evaluations are idempotent (the id derives from the trace and the rubric). The
Evaluations page shows health by model, both axes, the grade and the per-criterion breakdown.

### Capture privacy per project

Each project has a capture mode, set on the **Privacy** page or with `PUT /v1/privacy/projects/{id}`:
`metadata` (no inputs/outputs; content-like attributes dropped), `redacted` (everything kept, but
emails, phone numbers, card and ID numbers, tokens and keys are masked) or `full`. It applies on
every ingestion path (SDK, Manitos, OTLP). Without a project mode each path keeps its default (SDK
and Manitos store what they send, OTLP stores metadata only); `OBSERVATORY_CAPTURE_DEFAULT`
overrides those defaults. The page also reports, from the latest spans, how many hold an input, an
output or sensitive patterns, so a mode can be checked instead of trusted, and **Apply to stored
data** rewrites what is already stored (irreversible, asks for confirmation). Redaction is
pattern-based: it does not detect names or addresses. Details: [docs/privacy.md](docs/privacy.md).

### Standard OpenTelemetry (OTLP) ingestion

Point any stock OpenTelemetry exporter at `OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:8100/v1/otlp`
Expand Down
35 changes: 35 additions & 0 deletions backend/alembic/versions/20261018_project_settings.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
"""add project settings (capture mode)

Revision ID: 20261018_settings
Revises: 20261017_runs
Create Date: 2026-10-18
"""

from collections.abc import Sequence

import sqlalchemy as sa

from alembic import op

revision: str = "20261018_settings"
down_revision: str | None = "20261017_runs"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
"""Create the project settings table."""

op.create_table(
"project_settings",
sa.Column("project_id", sa.String(length=128), nullable=False),
sa.Column("capture_mode", sa.String(length=10), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True),
sa.PrimaryKeyConstraint("project_id"),
)


def downgrade() -> None:
"""Drop the project settings table."""

op.drop_table("project_settings")
25 changes: 16 additions & 9 deletions backend/app/api/manitos_ingest.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.capture import apply_capture, capture_mode
from app.db import get_session
from app.models.ingestion import IngestionReceipt
from app.models.trace import Span, Trace
Expand Down Expand Up @@ -235,20 +236,23 @@ def _datetime_equal(left: datetime | None, right: datetime | None) -> bool:
return abs(left.timestamp() - right.timestamp()) < 0.000001


def _span_values(span: ManitOSSpan) -> dict[str, Any]:
def _span_values(span: ManitOSSpan, mode: str = "full") -> dict[str, Any]:
inp, out, attributes, meta = apply_capture(
mode, inp=span.input, out=span.output, attributes=span.attributes, meta=span.metadata
)
return {
"name": span.name,
"span_type": span.span_type,
"start_time": _as_datetime(span.start_time),
"end_time": _as_datetime(span.end_time),
"status": span.status,
"inp": span.input,
"out": span.output,
"inp": inp,
"out": out,
"tokens_input": span.tokens_input,
"tokens_output": span.tokens_output,
"cost_usd": span.cost_usd,
"meta": span.metadata,
"attributes": span.attributes,
"meta": meta,
"attributes": attributes,
}


Expand Down Expand Up @@ -316,9 +320,12 @@ async def ingest_manitos_trace(


async def apply_envelope(
session: AsyncSession, payload: ManitOSIngestRequest
session: AsyncSession, payload: ManitOSIngestRequest, default_capture: str = "full"
) -> ManitOSIngestResponse:
"""The ingestion itself (also used by the OTLP receiver, which builds envelopes)."""
"""The ingestion itself (also used by the OTLP receiver, which builds envelopes).

``default_capture`` is the path's capture mode when the project has none of its own."""
mode = await capture_mode(session, payload.project_id, default_capture)

digest = _payload_hash(payload)
receipt = await _existing_receipt(
Expand Down Expand Up @@ -369,7 +376,7 @@ async def apply_envelope(
"start_time": _as_datetime(payload.trace.start_time),
"end_time": _as_datetime(payload.trace.end_time),
"status": payload.trace.status,
"meta": payload.trace.metadata,
"meta": apply_capture(mode, meta=payload.trace.metadata)[3],
}
if trace is None:
trace = Trace(id=payload.trace.id, **trace_values)
Expand Down Expand Up @@ -411,7 +418,7 @@ async def apply_envelope(
materialized: dict[str, Span] = {}
values_by_id: dict[str, dict[str, Any]] = {}
for incoming in payload.spans:
values = _span_values(incoming)
values = _span_values(incoming, mode)
values_by_id[incoming.id] = values
existing = existing_by_id.get(incoming.id)
if existing is None:
Expand Down
9 changes: 7 additions & 2 deletions backend/app/api/otlp.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@
response_body,
translate,
)
from app.otlp import (
capture_mode as otlp_capture_mode,
)
from app.ratelimit import limiter

router = APIRouter()
Expand Down Expand Up @@ -86,7 +89,9 @@ async def receive_traces(
try:
body = decompress(await request.body(), request.headers.get("content-encoding"))
data = request_from_protobuf(body) if protobuf else request_from_json(body)
translated = translate(data, project_id=request.headers.get("x-observer-project"))
translated = translate(
data, project_id=request.headers.get("x-observer-project"), mode="full"
)
except OtlpError as exc:
return _plain(exc.status, str(exc))

Expand All @@ -98,7 +103,7 @@ async def receive_traces(
await _link_and_widen(session, envelope)
payload = ManitOSIngestRequest.model_validate(envelope)
async with session.begin_nested(): # a rejected trace leaves no receipt behind
await apply_envelope(session, payload)
await apply_envelope(session, payload, default_capture=otlp_capture_mode())
except ValidationError as exc:
rejected += count
errors.append(f"trace rejected: {exc.errors()[0]['msg']}"[:200])
Expand Down
177 changes: 177 additions & 0 deletions backend/app/api/privacy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,177 @@
"""Capture privacy per project (``/v1/privacy``): the mode, a report of what is stored, enforcement.

The report answers "what did this project actually store?" from the latest spans, so a mode change
can be checked rather than trusted.
"""

from __future__ import annotations

from typing import Any, Literal

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.capture import (
MODES,
apply_capture,
content_like_keys,
default_mode,
has_sensitive_patterns,
)
from app.db import get_session
from app.models.settings import ProjectSetting
from app.models.trace import Span, Trace

router = APIRouter()

AUDIT_SAMPLE = 2000
MAX_PROJECTS = 50
ENFORCE_BATCH = 500

Mode = Literal["metadata", "redacted", "full"]


class ModeIn(BaseModel):
capture_mode: Mode


class ConfirmIn(BaseModel):
confirm: bool = False


async def audit_project(session: AsyncSession, project_id: str) -> dict[str, Any]:
"""What the latest spans of a project actually hold."""
rows = (
await session.execute(
select(Span.inp, Span.out, Span.attributes, Span.meta)
.join(Trace, Trace.id == Span.trace_id)
.where(Trace.project_id == project_id)
.order_by(Span.start_time.desc())
.limit(AUDIT_SAMPLE)
)
).all()
with_input = sum(1 for r in rows if r[0])
with_output = sum(1 for r in rows if r[1])
return {
"sampled_spans": len(rows),
"with_input": with_input,
"with_output": with_output,
"with_content_like_attributes": sum(1 for r in rows if content_like_keys(r[2])),
"with_sensitive_patterns": sum(1 for r in rows if has_sensitive_patterns(*r)),
}


def _entry(project_id: str, setting: str | None) -> dict[str, Any]:
return {
"project_id": project_id,
"capture_mode": setting,
"source": "setting" if setting else "default",
"default_mode": default_mode("full"),
}


@router.get("/projects")
async def list_projects(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""Every known project with its capture mode and what its latest spans hold."""
settings = {
pid: mode
for pid, mode in (
await session.execute(select(ProjectSetting.project_id, ProjectSetting.capture_mode))
).all()
}
seen = (
(await session.execute(select(Trace.project_id).where(Trace.project_id.is_not(None))))
.scalars()
.all()
)
ids = sorted(set(settings) | set(seen))[:MAX_PROJECTS]
projects = []
for pid in ids:
entry = _entry(pid, settings.get(pid))
entry["report"] = await audit_project(session, pid)
projects.append(entry)
return {
"modes": list(MODES),
"note": (
"Without a project mode each ingestion path keeps its default: the SDK and the Manitos "
"exporter store what they send, OTLP stores metadata only."
),
"projects": projects,
}


@router.put("/projects/{project_id}")
async def set_mode(
project_id: str, body: ModeIn, session: AsyncSession = Depends(get_session)
) -> dict[str, Any]:
if not 0 < len(project_id) <= 128:
raise HTTPException(status_code=422, detail="invalid project id")
row = await session.get(ProjectSetting, project_id)
if row is None:
session.add(ProjectSetting(project_id=project_id, capture_mode=body.capture_mode))
else:
row.capture_mode = body.capture_mode
await session.flush()
return _entry(project_id, body.capture_mode)


@router.delete("/projects/{project_id}")
async def clear_mode(
project_id: str, session: AsyncSession = Depends(get_session)
) -> dict[str, Any]:
"""Back to the ingestion paths' defaults."""
await session.execute(delete(ProjectSetting).where(ProjectSetting.project_id == project_id))
return _entry(project_id, None)


@router.post("/projects/{project_id}/enforce")
async def enforce_mode(
project_id: str, body: ConfirmIn, session: AsyncSession = Depends(get_session)
) -> dict[str, Any]:
"""Re-apply the project's mode to what is already stored. **Irreversible** for ``metadata``
(stored input/output are removed) and ``redacted`` (matches are replaced): needs ``confirm``."""
row = await session.get(ProjectSetting, project_id)
if row is None:
raise HTTPException(status_code=409, detail="set a capture mode for this project first")
if not body.confirm:
raise HTTPException(status_code=422, detail="this rewrites stored data: send confirm=true")
mode = str(row.capture_mode)
scanned = changed = 0
if mode != "full":
span_ids = (
(
await session.execute(
select(Span.id)
.join(Trace, Trace.id == Span.trace_id)
.where(Trace.project_id == project_id)
)
)
.scalars()
.all()
)
for i in range(0, len(span_ids), ENFORCE_BATCH):
chunk = span_ids[i : i + ENFORCE_BATCH]
for span in (await session.execute(select(Span).where(Span.id.in_(chunk)))).scalars():
scanned += 1
inp, out, attributes, meta = apply_capture(
mode, inp=span.inp, out=span.out, attributes=span.attributes, meta=span.meta
)
if (inp, out, attributes, meta) != (span.inp, span.out, span.attributes, span.meta):
span.inp, span.out, span.attributes, span.meta = inp, out, attributes, meta
changed += 1
traces = (
await session.execute(select(Trace).where(Trace.project_id == project_id))
).scalars()
for trace in traces:
_, _, _, meta = apply_capture(mode, meta=trace.meta)
if meta != trace.meta:
trace.meta = meta
await session.flush()
return {
"project_id": project_id,
"capture_mode": mode,
"spans_scanned": scanned,
"spans_changed": changed,
}
Loading
Loading