mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-15 10:14:32 +00:00
fix(admin): second review pass
Correctness - /api/remote never recorded source.created, so URL, GitHub and connector sources had a source.deleted with no matching creation. All three creation paths now go through one _audit_source_created helper. - The prompt-cache rate divided cached tokens by a whole bucket's prompt tokens. A bucket is a day and mixes calls whose provider reports a cache breakdown with calls whose provider does not, so filtering buckets in the client could not separate them and the rate was understated by however much traffic ran on a non-reporting provider. The denominator is now computed in SQL over the reporting rows. - The outcome pill matched values nothing writes. Guardrails emit triggered / not_evaluated and the device feed emits dispatched; the map had blocked / denied / allowed, so a guardrail that fired rendered neutral grey -- the one signal the merged feed exists to surface. Fixtures were seeding the fictional values, so the tests passed on it too. - Stream duration_ms timed the consumer. stream_token_usage is a generator, so start-to-exhaustion includes the agent loop's tool handling and the SSE client's pace; a slow browser recorded ~30s for a sub-second call. It now accumulates only the time spent inside next(). Safety - Activity filters failed open: an unknown facet or unparseable timestamp was dropped, and no filter means every row, so a typo widened an audit view and on the export streamed the full history. Both are now a 400. - The search term was interpolated into an ILIKE pattern, so "100%" matched everything and "q1_report" matched more than it should. Escaped. - 0034 set actor_id NOT NULL with no default. A previous-release process inserting mid-rollout would raise, and in admin/routes.py that insert shares the request transaction, so a role grant beside it would roll back too. Noise and dead code - The per-user panel is a security panel: data-plane events file under the actor, so an active account's routine deletes pushed a denied login out of the 20-row window. It now excludes them; the Activity tab shows everything. - device_audit_log had no created_at-leading index, so the merged feed sequentially scanned that branch every page (migration 0036). - conversation.deleted_all no longer records when nothing was deleted, and agent.updated no longer records an empty field list. - Dropped by_model from /admin/usage (no consumer; an extra aggregate per page load), the duplicate filter surface on AuthEventsRepository that nothing called, and the unreachable FLOW_LABELS.schedule entry. - Type hints on record_event's conn and the remaining unannotated helpers.
This commit is contained in:
1 parent
2ea9d66789
commit
25c82003d7
33 files changed
+635
-206
No files matched your search
@@ -76,6 +76,15 @@ def upgrade() -> None:
|
||||
|
||||
# Backfilled every row above, so the NOT NULL is safe to assert now. New
|
||||
# rows always carry an actor (the repository derives one from ``user_id``).
|
||||
#
|
||||
# The DEFAULT is what makes this safe during a rolling deploy: a
|
||||
# still-running previous-release process inserts (user_id, event, ip,
|
||||
# user_agent, metadata) with no actor, and those inserts share the
|
||||
# request's transaction in admin/routes.py -- a NotNullViolation there
|
||||
# would 500 the deactivate or role grant and roll its write back with it.
|
||||
# 'unknown' is the same sentinel ``record_event`` writes for an
|
||||
# unauthenticated actor, so the column stays readable either way.
|
||||
op.execute("ALTER TABLE auth_events ALTER COLUMN actor_id SET DEFAULT 'unknown';")
|
||||
op.execute("ALTER TABLE auth_events ALTER COLUMN actor_id SET NOT NULL;")
|
||||
|
||||
op.execute(
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
"""0036 device audit feed index — created_at for the merged activity feed.
|
||||
|
||||
``device_audit_log`` carried only ``(device_id, created_at DESC)`` and
|
||||
``(user_id, created_at DESC)``. Both serve a per-device or per-user lookup;
|
||||
neither can serve the merged admin activity feed, which orders by
|
||||
``created_at DESC`` across every user and device, so that branch of the union
|
||||
was sequentially scanned on every page.
|
||||
|
||||
``auth_events`` got the equivalent index in 0034 and ``guardrail_events`` has
|
||||
had ``ix_guardrail_events_created`` since 0029; this closes the third journal.
|
||||
|
||||
Idempotent both ways.
|
||||
|
||||
Revision ID: 0036_device_audit_created_idx
|
||||
Revises: 0035_token_usage_latency
|
||||
"""
|
||||
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
|
||||
|
||||
revision: str = "0036_device_audit_created_idx"
|
||||
down_revision: Union[str, None] = "0035_token_usage_latency"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute(
|
||||
"CREATE INDEX IF NOT EXISTS device_audit_created_idx "
|
||||
"ON device_audit_log (created_at DESC);"
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute("DROP INDEX IF EXISTS device_audit_created_idx;")
|
||||
@@ -43,23 +43,38 @@ _EXPORT_CHUNK = 1_000
|
||||
_MAX_SEARCH_LENGTH = 200
|
||||
|
||||
|
||||
class _InvalidFilter(ValueError):
|
||||
"""A filter argument the caller got wrong. Surfaces as a 400."""
|
||||
|
||||
|
||||
def _facet_list(name: str, allowed: tuple[str, ...]) -> Optional[list[str]]:
|
||||
"""Parse a repeated/comma-separated query arg, dropping unknown values.
|
||||
"""Parse a repeated/comma-separated query arg into known facet values.
|
||||
|
||||
Rejects unknown values rather than dropping them. Dropping would leave the
|
||||
request with no filter on that facet, and "no filter" means *everything* --
|
||||
so a typo or a stale chip would silently widen an audit view instead of
|
||||
narrowing it, and on the export would stream the whole history.
|
||||
|
||||
Args:
|
||||
name: Query parameter name.
|
||||
allowed: The permitted values; anything else is ignored rather than
|
||||
rejected, so an old bookmark degrades instead of erroring.
|
||||
allowed: The permitted values.
|
||||
|
||||
Returns:
|
||||
The distinct valid values in the order given, or None when absent.
|
||||
The distinct values in the order given, or None when absent.
|
||||
|
||||
Raises:
|
||||
_InvalidFilter: If any supplied value is not in ``allowed``.
|
||||
"""
|
||||
raw: list[str] = []
|
||||
for value in request.args.getlist(name):
|
||||
raw.extend(part.strip() for part in value.split(","))
|
||||
raw.extend(part.strip() for part in value.split(",") if part.strip())
|
||||
seen: list[str] = []
|
||||
for value in raw:
|
||||
if value in allowed and value not in seen:
|
||||
if value not in allowed:
|
||||
raise _InvalidFilter(
|
||||
f"Unknown {name}: {value!r}. Expected one of {', '.join(allowed)}."
|
||||
)
|
||||
if value not in seen:
|
||||
seen.append(value)
|
||||
return seen or None
|
||||
|
||||
@@ -76,14 +91,21 @@ def _event_list() -> Optional[list[str]]:
|
||||
|
||||
|
||||
def _time_arg(name: str) -> Optional[datetime]:
|
||||
"""Parse an ISO-8601 timestamp query arg, or None when absent/unparseable."""
|
||||
"""Parse an ISO-8601 timestamp query arg, or None when absent.
|
||||
|
||||
Raises:
|
||||
_InvalidFilter: If present but unparseable. Silently ignoring it would
|
||||
drop the time bound and return the full history.
|
||||
"""
|
||||
raw = (request.args.get(name) or "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
except ValueError as exc:
|
||||
raise _InvalidFilter(
|
||||
f"{name} must be an ISO-8601 timestamp, got {raw!r}."
|
||||
) from exc
|
||||
# A naive bound is read as UTC, matching how the feed stores timestamps.
|
||||
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
||||
|
||||
@@ -122,7 +144,10 @@ class AdminActivityResource(Resource):
|
||||
def get(self):
|
||||
"""Merged audit feed, newest first, with filters and a total."""
|
||||
page, page_size, offset = _page()
|
||||
filters = _filters()
|
||||
try:
|
||||
filters = _filters()
|
||||
except _InvalidFilter as exc:
|
||||
return make_response(jsonify({"success": False, "message": str(exc)}), 400)
|
||||
with db_readonly() as conn:
|
||||
repo = ActivityRepository(conn)
|
||||
total = repo.count(**filters)
|
||||
@@ -183,7 +208,10 @@ class AdminActivityExportResource(Resource):
|
||||
jsonify({"success": False, "message": "Invalid format"}), 400
|
||||
)
|
||||
limit = max(1, min(_EXPORT_MAX_ROWS, _int_arg("max_rows", _EXPORT_MAX_ROWS)))
|
||||
filters = _filters()
|
||||
try:
|
||||
filters = _filters()
|
||||
except _InvalidFilter as exc:
|
||||
return make_response(jsonify({"success": False, "message": str(exc)}), 400)
|
||||
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||||
generator = _csv_rows if fmt == "csv" else _ndjson_rows
|
||||
return Response(
|
||||
|
||||
@@ -38,6 +38,9 @@ admin_ns = Namespace("admin", description="Admin-only management endpoints", pat
|
||||
_DEFAULT_PAGE_SIZE = 25
|
||||
_MAX_PAGE_SIZE = 100
|
||||
|
||||
# Event namespaces the per-user security panel leaves out; see the drill-down.
|
||||
_DATA_PLANE_PREFIXES = ("source.", "agent.", "conversation.")
|
||||
|
||||
|
||||
def _int_arg(name: str, default: int) -> int:
|
||||
try:
|
||||
@@ -135,8 +138,12 @@ class AdminUserResource(Resource):
|
||||
},
|
||||
"roles": sorted({"user", *roles_repo.role_names_for(user_id)}),
|
||||
"grants": roles_repo.list_for(user_id),
|
||||
# Identity and access only. Data-plane events file under the
|
||||
# actor, so an active user's routine deletes would push a
|
||||
# denied login or a role grant out of a 20-row window.
|
||||
# The Activity tab, filtered by user, shows everything.
|
||||
"recent_events": AuthEventsRepository(conn).list_recent(
|
||||
user_id, limit=20
|
||||
user_id, limit=20, exclude_events_like=_DATA_PLANE_PREFIXES
|
||||
),
|
||||
"counts": AdminStatsRepository(conn).user_counts(user_id),
|
||||
}
|
||||
@@ -347,7 +354,6 @@ class AdminUsageResource(Resource):
|
||||
total = usage_repo.sum_tokens_in_range(
|
||||
start=start, end=datetime.now(timezone.utc)
|
||||
)
|
||||
by_model = usage_repo.tokens_by_model(start=start)
|
||||
latency = stats_repo.latency_summary(since=start)
|
||||
top_users = stats_repo.top_token_users(since=start, limit=10)
|
||||
return make_response(
|
||||
@@ -362,7 +368,6 @@ class AdminUsageResource(Resource):
|
||||
# Summed from the series, not from ``by_model``: the latter
|
||||
# drops rows with no model_id, so it would undercount.
|
||||
"total_cost": round(sum(row["cost"] for row in series), 4),
|
||||
"by_model": by_model,
|
||||
"latency": latency,
|
||||
"top_users": top_users,
|
||||
}
|
||||
|
||||
@@ -25,6 +25,7 @@ import logging
|
||||
from typing import Any, Optional
|
||||
|
||||
from flask import has_request_context, request
|
||||
from sqlalchemy import Connection
|
||||
|
||||
from docsgpt.storage.db.repositories.auth_events import AuthEventsRepository
|
||||
|
||||
@@ -32,7 +33,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def record_event(
|
||||
conn,
|
||||
conn: Connection,
|
||||
event: str,
|
||||
*,
|
||||
actor: Optional[str],
|
||||
@@ -42,8 +43,9 @@ def record_event(
|
||||
"""Append one audit event, best-effort, inside the caller's transaction.
|
||||
|
||||
Args:
|
||||
conn: Open database connection, inside the action's transaction so the
|
||||
audit row commits atomically with the change it describes.
|
||||
conn: Open SQLAlchemy ``Connection`` -- not a Session -- inside the
|
||||
action's transaction, so the audit row commits atomically with the
|
||||
change it describes and ``begin_nested`` gives it a real SAVEPOINT.
|
||||
event: Dotted event name (``source.deleted``, ``agent.created``).
|
||||
actor: Who performed the action; ``"unknown"`` when unauthenticated.
|
||||
target: The user acted upon, when the event is about a user. Data-plane
|
||||
|
||||
@@ -1294,7 +1294,8 @@ class UpdateAgent(Resource):
|
||||
updated = bool(result)
|
||||
else:
|
||||
updated = agents_repo.update(pg_agent_id, user, update_fields)
|
||||
if updated:
|
||||
# An update that changed no field is not an event.
|
||||
if updated and update_fields:
|
||||
record_event(
|
||||
conn,
|
||||
"agent.updated",
|
||||
|
||||
@@ -73,12 +73,15 @@ class DeleteAllConversations(Resource):
|
||||
try:
|
||||
with db_session() as conn:
|
||||
deleted = ConversationsRepository(conn).delete_all_for_user(user_id)
|
||||
record_event(
|
||||
conn,
|
||||
"conversation.deleted_all",
|
||||
actor=user_id,
|
||||
deleted=deleted,
|
||||
)
|
||||
# Nothing deleted is not an event; the endpoint is idempotent
|
||||
# and a row here would render as a destructive action.
|
||||
if deleted:
|
||||
record_event(
|
||||
conn,
|
||||
"conversation.deleted_all",
|
||||
actor=user_id,
|
||||
deleted=deleted,
|
||||
)
|
||||
except Exception as err:
|
||||
current_app.logger.error(
|
||||
f"Error deleting all conversations: {err}", exc_info=True
|
||||
|
||||
@@ -5,6 +5,7 @@ import os
|
||||
import tempfile
|
||||
import uuid
|
||||
import zipfile
|
||||
from typing import Optional
|
||||
|
||||
from flask import current_app, jsonify, make_response, request
|
||||
from flask_restx import fields, Namespace, Resource
|
||||
@@ -179,6 +180,40 @@ def _source_archive_limits() -> ZipExtractionLimits:
|
||||
)
|
||||
|
||||
|
||||
def _audit_source_created(
|
||||
*,
|
||||
source_id: str,
|
||||
user: Optional[str],
|
||||
name: Optional[str],
|
||||
source_type: str,
|
||||
task_id: Optional[str],
|
||||
) -> None:
|
||||
"""Record ``source.created`` for an accepted ingestion job.
|
||||
|
||||
Audited at the request, not in the worker: enqueuing is the user's action.
|
||||
The ingest may still fail, which the source's own status records.
|
||||
|
||||
Every path that mints a ``source_id`` and dispatches must call this --
|
||||
``source.deleted`` is recorded for remote sources too, and a trail showing
|
||||
a deletion with no matching creation is worse than no trail.
|
||||
"""
|
||||
try:
|
||||
with db_session() as conn:
|
||||
record_event(
|
||||
conn,
|
||||
"source.created",
|
||||
actor=user,
|
||||
source_id=source_id,
|
||||
name=name,
|
||||
type=source_type,
|
||||
task_id=task_id,
|
||||
)
|
||||
except Exception as err:
|
||||
current_app.logger.warning(
|
||||
"Could not audit source.created for %s: %s", source_id, err, exc_info=True
|
||||
)
|
||||
|
||||
|
||||
@sources_upload_ns.route("/upload")
|
||||
class UploadFile(Resource):
|
||||
@api.expect(
|
||||
@@ -397,26 +432,13 @@ class UploadFile(Resource):
|
||||
return make_response(jsonify({"success": False}), 400)
|
||||
# Predetermined id matches the dedup-claim row; loser GET sees same.
|
||||
response_task_id = predetermined_task_id or task.id
|
||||
# Audited here, not in the worker: this is the user action. The
|
||||
# ingest may still fail, which the source's own status records.
|
||||
try:
|
||||
with db_session() as conn:
|
||||
record_event(
|
||||
conn,
|
||||
"source.created",
|
||||
actor=user,
|
||||
source_id=str(source_uuid),
|
||||
name=job_name,
|
||||
type="local",
|
||||
task_id=response_task_id,
|
||||
)
|
||||
except Exception as err:
|
||||
current_app.logger.warning(
|
||||
"Could not audit upload for source %s: %s",
|
||||
source_uuid,
|
||||
err,
|
||||
exc_info=True,
|
||||
)
|
||||
_audit_source_created(
|
||||
source_id=str(source_uuid),
|
||||
user=user,
|
||||
name=job_name,
|
||||
source_type="local",
|
||||
task_id=response_task_id,
|
||||
)
|
||||
# ``source_uuid`` was minted above and passed to the worker as
|
||||
# ``source_id``; the worker uses it verbatim for every SSE event,
|
||||
# so the frontend can correlate inbound ``source.ingest.*`` to
|
||||
@@ -568,6 +590,13 @@ class UploadRemote(Resource):
|
||||
connector_kwargs["task_id"] = predetermined_task_id
|
||||
task = ingest_connector_task.apply_async(**connector_kwargs)
|
||||
response_task_id = predetermined_task_id or task.id
|
||||
_audit_source_created(
|
||||
source_id=str(source_uuid),
|
||||
user=user,
|
||||
name=data["name"],
|
||||
source_type=data["source"],
|
||||
task_id=response_task_id,
|
||||
)
|
||||
# ``source_uuid`` was minted above and passed to the
|
||||
# worker as ``source_id``; the worker uses it verbatim
|
||||
# for every SSE event, so the frontend can correlate
|
||||
@@ -601,6 +630,13 @@ class UploadRemote(Resource):
|
||||
_release_claim(scoped_key)
|
||||
return make_response(jsonify({"success": False}), 400)
|
||||
response_task_id = predetermined_task_id or task.id
|
||||
_audit_source_created(
|
||||
source_id=str(source_uuid),
|
||||
user=user,
|
||||
name=data["name"],
|
||||
source_type=data["source"],
|
||||
task_id=response_task_id,
|
||||
)
|
||||
response_payload = {
|
||||
"success": True,
|
||||
"task_id": response_task_id,
|
||||
|
||||
@@ -65,3 +65,21 @@ def row_to_dict(row: Any) -> dict:
|
||||
out["_id"] = out["id"]
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def like_escape(term: str) -> str:
|
||||
"""Escape LIKE/ILIKE metacharacters so ``term`` matches literally.
|
||||
|
||||
A search box takes a substring, not a pattern. Interpolated raw, ``%``
|
||||
matches everything and ``_`` matches any single character, so searching
|
||||
for ``100%`` or ``q1_report`` silently returns the wrong rows.
|
||||
|
||||
Callers must pair this with ``ESCAPE '\\'`` on the comparison.
|
||||
|
||||
Args:
|
||||
term: The user-supplied substring.
|
||||
|
||||
Returns:
|
||||
``term`` with ``\\``, ``%`` and ``_`` backslash-escaped.
|
||||
"""
|
||||
return term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
@@ -250,9 +250,11 @@ token_usage_table = Table(
|
||||
# Added in ``0033_quotas``. USD cost of the call at write time; 0 for
|
||||
# unpriced and bring-your-own models.
|
||||
Column("cost", Numeric(12, 8), nullable=False, server_default="0"),
|
||||
# Added in ``0035_token_usage_latency``. Wall-clock for the call, and time
|
||||
# to the first streamed chunk. ``ttft_ms`` is NULL for non-streaming calls
|
||||
# and for streams that failed before yielding -- "no first token", not 0.
|
||||
# Added in ``0035_token_usage_latency``. Time spent waiting on the
|
||||
# provider (a stream excludes consumer backpressure; see ``usage.py``),
|
||||
# and time to the first streamed chunk. ``ttft_ms`` is NULL for
|
||||
# non-streaming calls and for streams that failed before yielding --
|
||||
# "no first token", not 0.
|
||||
Column("duration_ms", Integer),
|
||||
Column("ttft_ms", Integer),
|
||||
)
|
||||
|
||||
@@ -25,6 +25,7 @@ from typing import Any, Iterator, Optional, Sequence
|
||||
from sqlalchemy import Connection, text
|
||||
|
||||
from docsgpt.audit_events import category_case_sql
|
||||
from docsgpt.storage.db.base_repository import like_escape
|
||||
|
||||
|
||||
# Columns every branch of the union produces, in order.
|
||||
@@ -197,12 +198,18 @@ class ActivityRepository:
|
||||
clauses.append("created_at <= :until")
|
||||
params["until"] = until
|
||||
if search:
|
||||
# ``search`` is a literal substring, not a pattern: an operator
|
||||
# looking for "100%" or "q1_report" must not have those characters
|
||||
# read as wildcards.
|
||||
clauses.append(
|
||||
"(actor_id ILIKE :search OR target_id ILIKE :search "
|
||||
"OR event ILIKE :search OR ip ILIKE :search "
|
||||
"OR COALESCE(outcome, '') ILIKE :search OR detail::text ILIKE :search)"
|
||||
"(actor_id ILIKE :search ESCAPE '\\' "
|
||||
"OR target_id ILIKE :search ESCAPE '\\' "
|
||||
"OR event ILIKE :search ESCAPE '\\' "
|
||||
"OR ip ILIKE :search ESCAPE '\\' "
|
||||
"OR COALESCE(outcome, '') ILIKE :search ESCAPE '\\' "
|
||||
"OR detail::text ILIKE :search ESCAPE '\\')"
|
||||
)
|
||||
params["search"] = f"%{search}%"
|
||||
params["search"] = f"%{like_escape(search)}%"
|
||||
# Always a WHERE, even with no filters, so the keyset cursor below can
|
||||
# append " AND ..." without sniffing the string it was handed.
|
||||
where = "WHERE " + " AND ".join(clauses or ["TRUE"])
|
||||
@@ -259,7 +266,10 @@ class ActivityRepository:
|
||||
return [dict(row._mapping) for row in result.fetchall()]
|
||||
|
||||
def _page_after(
|
||||
self, cursor: Optional[tuple], limit: int, filters: dict
|
||||
self,
|
||||
cursor: Optional[tuple[datetime, str, str]],
|
||||
limit: int,
|
||||
filters: dict,
|
||||
) -> list[dict]:
|
||||
"""One keyset page strictly older than ``cursor``.
|
||||
|
||||
@@ -332,7 +342,7 @@ class ActivityRepository:
|
||||
Row dicts carrying :data:`ACTIVITY_COLUMNS`.
|
||||
"""
|
||||
emitted = 0
|
||||
cursor: Optional[tuple] = None
|
||||
cursor: Optional[tuple[datetime, str, str]] = None
|
||||
while emitted < max_rows:
|
||||
page = self._page_after(
|
||||
cursor, min(chunk_size, max_rows - emitted), filters
|
||||
|
||||
@@ -17,7 +17,7 @@ from docsgpt.storage.db.base_repository import row_to_dict
|
||||
from docsgpt.storage.db.repositories.token_usage import TokenUsageRepository
|
||||
|
||||
|
||||
def _round_ms(value) -> Optional[int]:
|
||||
def _round_ms(value: Optional[float]) -> Optional[int]:
|
||||
"""Round a percentile to whole milliseconds, preserving NULL as None."""
|
||||
return int(round(float(value))) if value is not None else None
|
||||
|
||||
|
||||
@@ -88,18 +88,44 @@ class AuthEventsRepository:
|
||||
)
|
||||
return row_to_dict(result.fetchone())
|
||||
|
||||
def list_recent(self, user_id: str, limit: int = 50) -> list[dict]:
|
||||
"""Return the newest events for ``user_id``, newest first."""
|
||||
def list_recent(
|
||||
self,
|
||||
user_id: str,
|
||||
limit: int = 50,
|
||||
*,
|
||||
exclude_events_like: Optional[Sequence[str]] = None,
|
||||
) -> list[dict]:
|
||||
"""Return the newest events for ``user_id``, newest first.
|
||||
|
||||
Args:
|
||||
user_id: The subject whose trail to read.
|
||||
limit: How many rows to return.
|
||||
exclude_events_like: Event-name prefixes to drop, e.g.
|
||||
``("source.", "agent.")``. Data-plane events are filed under
|
||||
the actor, so on an active account they would otherwise push
|
||||
a denied login or a role grant out of a short window within
|
||||
minutes.
|
||||
|
||||
Returns:
|
||||
Matching rows, newest first.
|
||||
"""
|
||||
clauses = ["user_id = :user_id"]
|
||||
params: dict = {"user_id": user_id, "limit": limit}
|
||||
for index, prefix in enumerate(exclude_events_like or ()):
|
||||
key = f"excl_{index}"
|
||||
clauses.append(f"event NOT LIKE :{key}")
|
||||
params[key] = f"{prefix}%"
|
||||
where = " AND ".join(clauses)
|
||||
result = self._conn.execute(
|
||||
text(
|
||||
"""
|
||||
f"""
|
||||
SELECT * FROM auth_events
|
||||
WHERE user_id = :user_id
|
||||
WHERE {where}
|
||||
ORDER BY created_at DESC
|
||||
LIMIT :limit
|
||||
"""
|
||||
),
|
||||
{"user_id": user_id, "limit": limit},
|
||||
params,
|
||||
)
|
||||
return [row_to_dict(row) for row in result.fetchall()]
|
||||
|
||||
@@ -108,41 +134,18 @@ class AuthEventsRepository:
|
||||
event: Optional[str],
|
||||
user_id: Optional[str],
|
||||
since: Any,
|
||||
*,
|
||||
events: Optional[Sequence[str]] = None,
|
||||
actor_id: Optional[str] = None,
|
||||
until: Any = None,
|
||||
search: Optional[str] = None,
|
||||
) -> tuple[str, dict]:
|
||||
clauses: list[str] = []
|
||||
params: dict = {}
|
||||
if event:
|
||||
clauses.append("event = :event")
|
||||
params["event"] = event
|
||||
if events:
|
||||
clauses.append("event = ANY(:events)")
|
||||
params["events"] = list(events)
|
||||
if user_id:
|
||||
clauses.append("(user_id = :user_id OR target_id = :user_id)")
|
||||
params["user_id"] = user_id
|
||||
if actor_id:
|
||||
clauses.append("actor_id = :actor_id")
|
||||
params["actor_id"] = actor_id
|
||||
if since is not None:
|
||||
clauses.append("created_at >= :since")
|
||||
params["since"] = since
|
||||
if until is not None:
|
||||
clauses.append("created_at <= :until")
|
||||
params["until"] = until
|
||||
if search:
|
||||
# Substring match across the human-meaningful columns plus the
|
||||
# serialized metadata, so "ci-deploy-key" finds the PAT it named.
|
||||
clauses.append(
|
||||
"(user_id ILIKE :search OR actor_id ILIKE :search "
|
||||
"OR target_id ILIKE :search OR event ILIKE :search "
|
||||
"OR ip ILIKE :search OR metadata::text ILIKE :search)"
|
||||
)
|
||||
params["search"] = f"%{search}%"
|
||||
where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
|
||||
return where, params
|
||||
|
||||
@@ -150,25 +153,18 @@ class AuthEventsRepository:
|
||||
self,
|
||||
*,
|
||||
event: Optional[str] = None,
|
||||
events: Optional[Sequence[str]] = None,
|
||||
user_id: Optional[str] = None,
|
||||
actor_id: Optional[str] = None,
|
||||
since=None,
|
||||
until=None,
|
||||
search: Optional[str] = None,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> list[dict]:
|
||||
"""Global audit feed (admin), newest first; see :meth:`_filter_clauses`."""
|
||||
where, params = self._filter_clauses(
|
||||
event,
|
||||
user_id,
|
||||
since,
|
||||
events=events,
|
||||
actor_id=actor_id,
|
||||
until=until,
|
||||
search=search,
|
||||
)
|
||||
"""Single-journal audit feed for ``GET /api/admin/audit``, newest first.
|
||||
|
||||
The admin UI reads the merged feed instead; richer filtering lives on
|
||||
:class:`~docsgpt.storage.db.repositories.activity.ActivityRepository`
|
||||
rather than being implemented twice here.
|
||||
"""
|
||||
where, params = self._filter_clauses(event, user_id, since)
|
||||
params.update({"limit": int(limit), "offset": int(offset)})
|
||||
result = self._conn.execute(
|
||||
text(
|
||||
@@ -183,37 +179,14 @@ class AuthEventsRepository:
|
||||
self,
|
||||
*,
|
||||
event: Optional[str] = None,
|
||||
events: Optional[Sequence[str]] = None,
|
||||
user_id: Optional[str] = None,
|
||||
actor_id: Optional[str] = None,
|
||||
since=None,
|
||||
until=None,
|
||||
search: Optional[str] = None,
|
||||
) -> int:
|
||||
"""Total matching the same filters as :meth:`list_all` (for pagination)."""
|
||||
where, params = self._filter_clauses(
|
||||
event,
|
||||
user_id,
|
||||
since,
|
||||
events=events,
|
||||
actor_id=actor_id,
|
||||
until=until,
|
||||
search=search,
|
||||
)
|
||||
where, params = self._filter_clauses(event, user_id, since)
|
||||
return int(
|
||||
self._conn.execute(
|
||||
text(f"SELECT count(*) FROM auth_events {where}"), params
|
||||
).scalar()
|
||||
or 0
|
||||
)
|
||||
|
||||
def event_names(self) -> list[str]:
|
||||
"""Distinct event names present in the table, alphabetically.
|
||||
|
||||
Feeds the admin filter's event picker, so operators pick from what the
|
||||
instance actually recorded instead of typing a name from memory.
|
||||
"""
|
||||
result = self._conn.execute(
|
||||
text("SELECT DISTINCT event FROM auth_events ORDER BY event")
|
||||
)
|
||||
return [row[0] for row in result.fetchall()]
|
||||
@@ -306,7 +306,18 @@ class TokenUsageRepository:
|
||||
COALESCE(SUM(tu.cost), 0) AS cost,
|
||||
-- NULL cache bins mean "provider did not report", so
|
||||
-- they stay out of the sum rather than reading as 0.
|
||||
SUM(tu.cached_tokens) AS cached_tokens
|
||||
SUM(tu.cached_tokens) AS cached_tokens,
|
||||
-- The matching denominator: prompt tokens from the
|
||||
-- rows that reported. Dividing ``cached_tokens`` by
|
||||
-- ``prompt_tokens`` instead would understate the hit
|
||||
-- rate by however much traffic ran on a provider that
|
||||
-- reports nothing, since those rows are in one sum but
|
||||
-- not the other.
|
||||
COALESCE(
|
||||
SUM(tu.prompt_tokens)
|
||||
FILTER (WHERE tu.cached_tokens IS NOT NULL),
|
||||
0
|
||||
) AS cache_eligible_prompt_tokens
|
||||
{group_select}
|
||||
FROM token_usage tu
|
||||
{join}
|
||||
@@ -328,6 +339,9 @@ class TokenUsageRepository:
|
||||
if row._mapping["cached_tokens"] is not None
|
||||
else None
|
||||
),
|
||||
"cache_eligible_prompt_tokens": int(
|
||||
row._mapping["cache_eligible_prompt_tokens"]
|
||||
),
|
||||
**(
|
||||
{"group_key": row._mapping["group_key"]}
|
||||
if group_by is not None
|
||||
|
||||
+23
-4
@@ -305,12 +305,28 @@ def stream_token_usage(func):
|
||||
batch = []
|
||||
started_at = time.monotonic()
|
||||
first_chunk_at: float | None = None
|
||||
# Time spent waiting on the provider, accumulated across ``next()``
|
||||
# calls. The wall clock cannot be used here: this is a generator, so
|
||||
# every ``yield`` suspends until the consumer comes back, and the span
|
||||
# from start to exhaustion includes the agent loop's tool handling and
|
||||
# the SSE client's backpressure. A slow browser would otherwise record
|
||||
# 30s for a 900ms call, and latency_summary would mix that with true
|
||||
# non-streaming durations under one p50.
|
||||
provider_seconds = 0.0
|
||||
error: BaseException | None = None
|
||||
try:
|
||||
result = func(self, model, messages, stream, tools, **kwargs)
|
||||
for r in result:
|
||||
stream_iter = iter(result)
|
||||
while True:
|
||||
pull_started = time.monotonic()
|
||||
try:
|
||||
r = next(stream_iter)
|
||||
except StopIteration:
|
||||
provider_seconds += time.monotonic() - pull_started
|
||||
break
|
||||
provider_seconds += time.monotonic() - pull_started
|
||||
if first_chunk_at is None:
|
||||
first_chunk_at = time.monotonic()
|
||||
first_chunk_at = pull_started + provider_seconds
|
||||
batch.append(r)
|
||||
yield r
|
||||
except Exception as exc:
|
||||
@@ -320,7 +336,7 @@ def stream_token_usage(func):
|
||||
error = exc
|
||||
raise
|
||||
finally:
|
||||
duration_ms = int((time.monotonic() - started_at) * 1000)
|
||||
duration_ms = int(provider_seconds * 1000)
|
||||
# NULL, not 0, when the stream failed before yielding: "no first
|
||||
# token" must not read as an instant one in a p50.
|
||||
ttft_ms = (
|
||||
@@ -343,7 +359,10 @@ def stream_token_usage(func):
|
||||
model,
|
||||
prompt_tokens=call_usage["prompt_tokens"],
|
||||
completion_tokens=call_usage["generated_tokens"],
|
||||
latency_ms=duration_ms,
|
||||
# The log line has always meant end-to-end wall clock
|
||||
# for the streamed response; only the persisted column
|
||||
# isolates provider time.
|
||||
latency_ms=int((time.monotonic() - started_at) * 1000),
|
||||
cached_tokens=call_usage.get("cached_tokens"),
|
||||
cache_write_tokens=call_usage.get("cache_write_tokens"),
|
||||
error=error,
|
||||
|
||||
Reference in new issue
Block a user