diff --git a/backend/ee/onyx/configs/license_enforcement_config.py b/backend/ee/onyx/configs/license_enforcement_config.py index 68c05ddc7d0..51d44429012 100644 --- a/backend/ee/onyx/configs/license_enforcement_config.py +++ b/backend/ee/onyx/configs/license_enforcement_config.py @@ -78,7 +78,7 @@ "/manage/admin/standard-answer": Tier.ENTERPRISE, "/admin/token-rate-limits": Tier.ENTERPRISE, "/admin/hooks": Tier.ENTERPRISE, # outbound webhooks - "/admin/log-export": Tier.ENTERPRISE, # container-local log download + "/admin/log-export": Tier.ENTERPRISE, # deployment-wide log export "/analytics": Tier.ENTERPRISE, # non-admin analytics (e.g. assistant stats) "/evals": Tier.ENTERPRISE, "/scim": Tier.ENTERPRISE, # SCIM protocol diff --git a/backend/ee/onyx/server/log_export/api.py b/backend/ee/onyx/server/log_export/api.py index 57d3118ad30..0cf5b374f8a 100644 --- a/backend/ee/onyx/server/log_export/api.py +++ b/backend/ee/onyx/server/log_export/api.py @@ -8,10 +8,7 @@ from fastapi.responses import StreamingResponse from starlette.background import BackgroundTask -from ee.onyx.server.log_export.collection import ( - build_log_zip, - get_default_log_directories, -) +from ee.onyx.server.log_export.collection import get_default_log_directories from ee.onyx.server.log_export.models import ( LogExportManifest, LogExportStartResponse, @@ -46,24 +43,18 @@ router = APIRouter() -API_SERVER_SCOPE_NOTE = ( - "Scope: this export contains log files from the api_server container only. " - "Logs from background workers and other services are not included; use " - "'docker logs ' or 'kubectl logs ' to retrieve those." -) - class _ExpiringLock: """Non-blocking lock whose hold expires after a TTL. - Guards against leaked holds: release hooks tied to the response lifecycle - are skipped by Starlette on some exit paths (a body iterator raising, or - client disconnects under ASGI >= 2.4), so a plain ``threading.Lock`` could - stay held until process restart. Expiry bounds any such leak. + Guards against leaked holds: when no code path releases (nobody polls the + status endpoint, or the poll lands on another replica), a plain + ``threading.Lock`` would stay held until process restart. Expiry bounds any + such leak. ``try_acquire`` returns a token; ``release`` is a no-op unless the token - belongs to the current hold, so a stale holder (or a duplicate call from a - second cleanup hook) can never release a successor's hold. + belongs to the current hold, so a stale holder (or a duplicate release) can + never free a successor's hold. """ def __init__( @@ -97,79 +88,6 @@ def held(self) -> bool: return self._held_until is not None and self._clock() < self._held_until -# Serializes exports process-wide: each one burns seconds of CPU on compression -# and holds a temp file until streaming ends, and concurrent exports of the same -# logs are pure waste. The TTL comfortably exceeds build time plus a slow -# streaming session (nginx's ``proxy_read_timeout`` defaults to 300s of idle). -_EXPORT_LOCK_TTL_SECONDS = 15 * 60 -_EXPORT_LOCK = _ExpiringLock(ttl_seconds=_EXPORT_LOCK_TTL_SECONDS) - - -@router.get("/admin/log-export/download") -def download_api_server_logs( - _: User = Depends(require_permission(Permission.FULL_ADMIN_PANEL_ACCESS)), -) -> StreamingResponse: - if MULTI_TENANT: - raise OnyxError( - OnyxErrorCode.SINGLE_TENANT_ONLY, - "Log export is only available on self-hosted deployments.", - ) - - token = _EXPORT_LOCK.try_acquire() - if token is None: - raise OnyxError( - OnyxErrorCode.RATE_LIMITED, - "A log export is already in progress. Try again once it completes.", - ) - - handed_off = False - try: - # The archive is fully materialized before streaming, so its exact size - # is known and an explicit Content-Length can be sent. - built = build_log_zip(get_default_log_directories(), API_SERVER_SCOPE_NOTE) - zip_buffer = built.zip_buffer - - def cleanup() -> None: - # Wired to both the generator's ``finally`` and the response - # background task because neither alone covers every exit path: - # Starlette skips background tasks when the body iterator raises - # (and on client disconnects under ASGI >= 2.4), while a generator - # ``finally`` never runs if the generator is closed before its first - # iteration. Double invocation is safe: ``close`` tolerates repeats - # and ``release`` ignores stale or duplicate tokens. - try: - zip_buffer.close() - finally: - _EXPORT_LOCK.release(token) - - def iter_zip() -> Generator[bytes, None, None]: - try: - while chunk := zip_buffer.read(STANDARD_CHUNK_SIZE): - yield chunk - finally: - cleanup() - - timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d_%H-%M-%S") - response = StreamingResponse( - content=iter_zip(), - media_type="application/zip", - headers={ - "Content-Disposition": ( - f"attachment; filename=onyx_api_server_logs_{timestamp}.zip" - ), - "Content-Length": str(built.size_bytes), - }, - background=BackgroundTask(cleanup), - ) - handed_off = True - return response - finally: - # Once the response exists, its cleanup hooks own the release; until - # then, any exit (including BaseException) releases here. - if not handed_off: - _EXPORT_LOCK.release(token) - - API_SERVER_WORKER_NAME = "api_server" # One collector task per worker type, each routed to a queue that worker @@ -302,8 +220,6 @@ def start_log_export( _ASYNC_EXPORT_LOCK.release(token) -# Declared after the sync ``/admin/log-export/download`` route above so that -# literal path keeps matching before ``{export_id}``. @router.get("/admin/log-export/{export_id}") def get_log_export_status( export_id: str, @@ -352,8 +268,11 @@ def download_log_export( def cleanup() -> None: # Wired to both the generator's ``finally`` and the response background - # task, matching ``download_api_server_logs``; ``close`` tolerates - # repeats. No lock is involved here. + # task because neither alone covers every exit path: Starlette skips + # background tasks when the body iterator raises (and on client + # disconnects under ASGI >= 2.4), while a generator ``finally`` never + # runs if the generator is closed before its first iteration. Double + # invocation is safe: ``close`` tolerates repeats. zip_buffer.close() def iter_zip() -> Generator[bytes, None, None]: diff --git a/backend/onyx/background/celery/tasks/docfetching/tasks.py b/backend/onyx/background/celery/tasks/docfetching/tasks.py index 6e67e33ce2a..0cce6699cc1 100644 --- a/backend/onyx/background/celery/tasks/docfetching/tasks.py +++ b/backend/onyx/background/celery/tasks/docfetching/tasks.py @@ -53,6 +53,7 @@ from onyx.redis.redis_connector import RedisConnector from onyx.server.metrics.connector_health_metrics import on_index_attempt_status_change from onyx.utils.logger import setup_logger +from onyx.utils.os_reaper import reap_children_before_exit from onyx.utils.variable_functionality import global_version from shared_configs.configs import SENTRY_CELERY_TRACES_SAMPLE_RATE, SENTRY_DSN @@ -272,6 +273,8 @@ def _docfetching_task( cc_pair_id, search_settings_id, ) + # os._exit bypasses the drain in _initializer's finally, so reap here. + reap_children_before_exit() os._exit(0) # ensure process exits cleanly diff --git a/backend/onyx/background/indexing/job_client.py b/backend/onyx/background/indexing/job_client.py index 47969d9652b..9394227c2de 100644 --- a/backend/onyx/background/indexing/job_client.py +++ b/backend/onyx/background/indexing/job_client.py @@ -16,6 +16,11 @@ from onyx.configs.constants import POSTGRES_CELERY_WORKER_INDEXING_CHILD_APP_NAME from onyx.db.engine.sql_engine import SqlEngine from onyx.utils.logger import setup_logger +from onyx.utils.os_reaper import ( + become_child_subreaper, + install_sigterm_drain, + reap_children_before_exit, +) from shared_configs.configs import POSTGRES_DEFAULT_SCHEMA, TENANT_ID_PREFIX from shared_configs.contextvars import CURRENT_TENANT_ID_CONTEXTVAR @@ -55,6 +60,12 @@ def _initializer( kwargs = {} logger.info("Initializing spawned worker child process.") + + # adopt orphans (e.g. Chromium helpers) instead of leaking zombies to PID 1, + # and drain them even when a watchdog cancels this child with SIGTERM + become_child_subreaper() + install_sigterm_drain() + # 1. Get tenant_id from args or fallback to default tenant_id = POSTGRES_DEFAULT_SCHEMA for arg in reversed(args): @@ -94,6 +105,11 @@ def _initializer( finally: CURRENT_TENANT_ID_CONTEXTVAR.reset(token) + # os._exit entrypoints skip this finally and drain themselves + reaped = reap_children_before_exit() + if reaped: + logger.info("Spawned worker child reaped %s orphaned processes.", reaped) + def _run_in_process( func: Callable, diff --git a/backend/onyx/document_index/opensearch/client.py b/backend/onyx/document_index/opensearch/client.py index 8d0ef88c684..9f854035868 100644 --- a/backend/onyx/document_index/opensearch/client.py +++ b/backend/onyx/document_index/opensearch/client.py @@ -160,11 +160,27 @@ def __init__( _VERSION_CONFLICT_ERROR_TYPE = "version_conflict_engine_exception" # Raised by a search whose PIT has expired/been deleted; we re-open and retry. _SEARCH_CONTEXT_MISSING_ERROR_TYPE = "search_context_missing" +# Rejection by an index/cluster block, e.g. the read_only_allow_delete block +# OpenSearch applies when disk usage crosses the flood-stage watermark. +_CLUSTER_BLOCK_ERROR_TYPE = "cluster_block_exception" # Chunks per PIT-scan page. A port doc-batch is small (INDEX_BATCH_SIZE docs), so # one page covers a batch; paging still protects against a pathological doc. _PIT_SCAN_PAGE_SIZE = 1000 +def is_cluster_block_error(e: Exception) -> bool: + """True when a request was rejected by an index/cluster block rather than a + problem with the request itself.""" + return isinstance(e, TransportError) and _CLUSTER_BLOCK_ERROR_TYPE in str(e.error) + + +class OpenSearchIndexWriteBlockedError(Exception): + """An existing index rejected a metadata write because of a block (e.g. + read_only_allow_delete applied at the disk flood-stage watermark). The + index is still fully readable — callers that can serve degraded may catch + this. Never raised for a missing index or a blocked index creation.""" + + class OpenSearchServerSideTimeout(Exception): """ A server-side timeout occurred when searching an OpenSearch index. diff --git a/backend/onyx/document_index/opensearch/opensearch_document_index.py b/backend/onyx/document_index/opensearch/opensearch_document_index.py index b0e476181d7..d61cbff043e 100644 --- a/backend/onyx/document_index/opensearch/opensearch_document_index.py +++ b/backend/onyx/document_index/opensearch/opensearch_document_index.py @@ -39,7 +39,9 @@ OpenSearchClient, OpenSearchDocumentMissingError, OpenSearchIndexClient, + OpenSearchIndexWriteBlockedError, SearchHit, + is_cluster_block_error, ) from onyx.document_index.opensearch.cluster_settings import OPENSEARCH_CLUSTER_SETTINGS from onyx.document_index.opensearch.constants import OpenSearchSearchType @@ -308,10 +310,24 @@ def __init__( and VERIFY_CREATE_OPENSEARCH_INDEX_ON_INIT_MT and index_name not in _verified_index_names_for_current_process ): - self.verify_and_create_index_if_necessary( - embedding_dim=embedding_dim, embedding_precision=embedding_precision - ) - _verified_index_names_for_current_process.add(index_name) + try: + self.verify_and_create_index_if_necessary( + embedding_dim=embedding_dim, embedding_precision=embedding_precision + ) + except OpenSearchIndexWriteBlockedError as e: + # Existing index, still readable — don't fail the caller. Not + # cached as verified, so a later init retries the mapping + # refresh once the block clears. + logger.error( + "Index %s is write-blocked; continuing without the mapping " + "refresh. Search still works, but indexing will fail until " + "the block is cleared (usually by freeing disk space below " + "the flood-stage watermark). Error: %s", + index_name, + e, + ) + else: + _verified_index_names_for_current_process.add(index_name) def verify_and_create_index_if_necessary( self, @@ -372,6 +388,15 @@ def verify_and_create_index_if_necessary( try: self._client.put_mapping(expected_mappings) except Exception as e: + if is_cluster_block_error(e): + # The index exists and is readable; only this metadata + # write was rejected. Raise the targeted type so + # callers that can serve degraded can catch exactly + # this case (never a missing index / blocked create). + raise OpenSearchIndexWriteBlockedError( + f"Index {self._index_name} is write-blocked; the mapping " + "refresh was rejected." + ) from e logger.error( "Failed to update mappings for index %s. This likely means a field type was changed which requires reindexing. Error: %s", self._index_name, diff --git a/backend/onyx/setup.py b/backend/onyx/setup.py index 467d64c1c1a..485933d8756 100644 --- a/backend/onyx/setup.py +++ b/backend/onyx/setup.py @@ -47,6 +47,7 @@ from onyx.document_index.interfaces_new import DocumentIndex from onyx.document_index.opensearch.client import ( OpenSearchClient, + OpenSearchIndexWriteBlockedError, wait_for_opensearch_with_timeout, ) from onyx.document_index.opensearch.opensearch_document_index import set_cluster_state @@ -235,6 +236,22 @@ def setup_document_indices( ) document_index_setup_success = True break + except OpenSearchIndexWriteBlockedError as e: + # The index exists but is write-blocked (typically the + # read_only_allow_delete block applied at the disk flood-stage + # watermark). It is still readable, so start up degraded rather + # than crash-loop until the block clears. A missing index or + # blocked creation raises a different error and still fails. + logger.error( + "Document index %s is write-blocked; continuing startup without " + "the mapping refresh. Search still works, but indexing will fail " + "until the block is cleared (usually by freeing disk space below " + "the flood-stage watermark). Error: %s", + document_index.__class__.__name__, + e, + ) + document_index_setup_success = True + break except Exception: logger.exception( "Document index %s setup did not succeed. The relevant service may not be ready yet. Retrying in %s seconds.", diff --git a/backend/onyx/utils/os_reaper.py b/backend/onyx/utils/os_reaper.py new file mode 100644 index 00000000000..778482a30d1 --- /dev/null +++ b/backend/onyx/utils/os_reaper.py @@ -0,0 +1,149 @@ +"""Reap orphaned child processes in spawned connector workers (Linux). + +Chromium helpers orphaned by Playwright teardown re-parent to PID 1 (the +celery worker, which never wait()s) and accumulate as zombies. Spawned +connector children mark themselves as the subreaper and drain them instead. + +Linux-only by nature, not as a shortcut: the pathology exists only where +PID 1 is a non-reaping worker (our containers). On macOS orphans re-parent +to launchd, which reaps them. + +Only call the drains from a process whose subprocess usage is sequential and +fully owned — a blanket waitpid(-1) steals exit statuses from concurrent +waiters. +""" + +import os +import signal +import sys +import time + +from onyx.utils.logger import setup_logger + +logger = setup_logger() + +_PR_SET_CHILD_SUBREAPER = 36 + + +def become_child_subreaper() -> bool: + """Re-parent orphaned descendants to this process instead of PID 1.""" + if sys.platform != "linux": + return False + + try: + import ctypes + + libc = ctypes.CDLL(None, use_errno=True) + if libc.prctl(_PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) != 0: + logger.warning( + "become_child_subreaper: prctl failed errno=%s", ctypes.get_errno() + ) + return False + except Exception: + logger.warning("become_child_subreaper: prctl unavailable", exc_info=True) + return False + + return True + + +def reap_exited_children() -> int: + """Reap already-exited (zombie) children without blocking; returns count.""" + if sys.platform != "linux": + return 0 + + reaped = 0 + while True: + try: + pid, _ = os.waitpid(-1, os.WNOHANG) + except ChildProcessError: + break # no children at all + except OSError: + logger.warning("reap_exited_children: waitpid failed", exc_info=True) + break + + if pid == 0: + break # children exist but none have exited + + reaped += 1 + + return reaped + + +def _live_child_pids() -> list[int]: + """Direct children (adopted orphans included) that are still running.""" + me = os.getpid() + pids = [] + for entry in os.listdir("/proc"): + if not entry.isdigit(): + continue + try: + with open(f"/proc/{entry}/stat") as f: + data = f.read() + # comm may contain spaces; state and ppid follow the closing paren + state, ppid = data[data.rindex(")") + 2 :].split()[:2] + if int(ppid) == me and state != "Z": + pids.append(int(entry)) + except (OSError, IndexError, ValueError): + continue + return pids + + +def reap_children_before_exit(grace_seconds: float = 2.0) -> int: + """Exit-path drain: reap exited children, give still-running ones a short + grace to finish, then SIGKILL the rest and reap them. A child left alive + at exit would re-parent to PID 1 and zombify there when it dies.""" + if sys.platform != "linux": + return 0 + + reaped = reap_exited_children() + deadline = time.monotonic() + grace_seconds + while _live_child_pids() and time.monotonic() < deadline: + time.sleep(0.05) + reaped += reap_exited_children() + + # iterate: killing a child re-parents ITS children to us (we are the + # subreaper), so new live children can appear mid-kill; bounded hard stop + kill_deadline = time.monotonic() + 5.0 + killed = 0 + while time.monotonic() < kill_deadline: + live = _live_child_pids() + if not live: + break + for pid in live: + try: + os.kill(pid, signal.SIGKILL) + killed += 1 + except OSError: + pass + time.sleep(0.05) + reaped += reap_exited_children() + + if killed: + logger.warning( + "reap_children_before_exit: SIGKILLed %s straggler children", killed + ) + + reaped += reap_exited_children() + return reaped + + +def install_sigterm_drain() -> None: + """Make SIGTERM drain orphans before the process dies (Linux, main thread). + + Watchdogs cancel spawned children with SIGTERM; the default disposition + kills the process instantly, stranding adopted zombies and live Chromium + descendants on PID 1. Exits 143 after draining.""" + if sys.platform != "linux": + return + + def _drain_and_exit(signum: int, frame: object) -> None: # noqa: ARG001 + reap_children_before_exit() + os._exit(128 + signal.SIGTERM) + + try: + signal.signal(signal.SIGTERM, _drain_and_exit) + except (ValueError, OSError): + # not the main thread / exotic embedding — child startup must not fail + logger.warning( + "install_sigterm_drain: could not install handler", exc_info=True + ) diff --git a/backend/tests/external_dependency_unit/document_index/test_verify_write_blocked.py b/backend/tests/external_dependency_unit/document_index/test_verify_write_blocked.py new file mode 100644 index 00000000000..c2228c712fb --- /dev/null +++ b/backend/tests/external_dependency_unit/document_index/test_verify_write_blocked.py @@ -0,0 +1,123 @@ +"""External dependency tests for behavior when the OpenSearch index carries a +write block, as OpenSearch applies at the disk flood-stage watermark. + +Regression tests for api-server pods crash-looping on startup while the index +was read_only_allow_delete-blocked: the mapping refresh is a metadata write, so +it is rejected while the block is active. verify_and_create_index_if_necessary +still raises (callers like embedding-model swaps must not silently continue); +the tolerant call sites — startup's setup_document_indices and the multitenant +DocumentIndex init — catch the block error and proceed degraded. +""" + +from collections.abc import Generator +from unittest.mock import patch + +import pytest + +from onyx.db.enums import EmbeddingPrecision +from onyx.document_index.interfaces_new import TenantState +from onyx.document_index.opensearch import ( + opensearch_document_index as opensearch_document_index_module, +) +from onyx.document_index.opensearch.client import ( + OpenSearchIndexClient, + OpenSearchIndexWriteBlockedError, + is_cluster_block_error, +) +from onyx.document_index.opensearch.opensearch_document_index import ( + OpenSearchDocumentIndex, +) +from onyx.indexing.models import IndexingSetting +from onyx.setup import setup_document_indices +from shared_configs.configs import POSTGRES_DEFAULT_SCHEMA_STANDARD_VALUE +from tests.external_dependency_unit.document_index.conftest import EMBEDDING_DIM + +_WRITE_BLOCK_SETTING = "index.blocks.read_only_allow_delete" + + +@pytest.fixture +def write_blocked_index( + opensearch_index: OpenSearchDocumentIndex, + test_index_name: str, +) -> Generator[OpenSearchDocumentIndex, None, None]: + """Applies the flood-stage write block to the test index for the duration + of the test. Clearing the block is always permitted, so cleanup works even + while the block is active.""" + client = OpenSearchIndexClient(index_name=test_index_name) + client.update_settings({_WRITE_BLOCK_SETTING: True}) + try: + yield opensearch_index + finally: + client.update_settings({_WRITE_BLOCK_SETTING: None}) + + +def test_verify_raises_typed_error_under_write_block( + write_blocked_index: OpenSearchDocumentIndex, +) -> None: + """verify_and_create_index_if_necessary keeps raising under the block (a + caller such as an embedding-model swap must not silently continue). The + existing-index refresh raises the targeted type — never raised for a + missing index or blocked creation — chained from the block rejection.""" + with pytest.raises(OpenSearchIndexWriteBlockedError) as exc_info: + write_blocked_index.verify_and_create_index_if_necessary( + embedding_dim=EMBEDDING_DIM, + embedding_precision=EmbeddingPrecision.FLOAT, + ) + + cause = exc_info.value.__cause__ + assert isinstance(cause, Exception) + assert is_cluster_block_error(cause) + + +def test_setup_document_indices_succeeds_under_write_block( + write_blocked_index: OpenSearchDocumentIndex, +) -> None: + """Startup must survive an existing, readable index that is merely + write-blocked instead of crash-looping.""" + index_setting = IndexingSetting.model_construct(model_dim=EMBEDDING_DIM) + + assert setup_document_indices( + document_indices=[write_blocked_index], + index_setting=index_setting, + num_attempts=1, + ) + + +def test_mt_init_survives_write_block_and_is_not_cached( + write_blocked_index: OpenSearchDocumentIndex, # noqa: ARG001 + test_index_name: str, +) -> None: + """Multitenant __init__ tolerates the block without caching the index as + verified, so the mapping refresh is retried once the block clears.""" + verified_names = ( + opensearch_document_index_module._verified_index_names_for_current_process + ) + mt_tenant_state = TenantState( + tenant_id=POSTGRES_DEFAULT_SCHEMA_STANDARD_VALUE, multitenant=True + ) + try: + with patch.object( + opensearch_document_index_module, + "VERIFY_CREATE_OPENSEARCH_INDEX_ON_INIT_MT", + True, + ): + OpenSearchDocumentIndex( + tenant_state=mt_tenant_state, + index_name=test_index_name, + embedding_dim=EMBEDDING_DIM, + embedding_precision=EmbeddingPrecision.FLOAT, + ) + assert test_index_name not in verified_names + + OpenSearchIndexClient(index_name=test_index_name).update_settings( + {_WRITE_BLOCK_SETTING: None} + ) + OpenSearchDocumentIndex( + tenant_state=mt_tenant_state, + index_name=test_index_name, + embedding_dim=EMBEDDING_DIM, + embedding_precision=EmbeddingPrecision.FLOAT, + ) + assert test_index_name in verified_names + finally: + verified_names.discard(test_index_name) diff --git a/backend/tests/integration/tests/log_export/test_log_export.py b/backend/tests/integration/tests/log_export/test_log_export.py index 2722518455a..467a046cd18 100644 --- a/backend/tests/integration/tests/log_export/test_log_export.py +++ b/backend/tests/integration/tests/log_export/test_log_export.py @@ -21,41 +21,6 @@ reason="Log export is an enterprise feature", ) class TestLogExport: - def test_admin_can_download_log_zip(self, admin_user: DATestUser) -> None: - response = client.get( - f"{API_SERVER_URL}/admin/log-export/download", - headers=admin_user.headers, - ) - assert response.status_code == 200 - assert response.headers["Content-Type"] == "application/zip" - assert "attachment" in response.headers["Content-Disposition"] - - # The environment may or may not write file logs, so assert the export - # structure rather than the presence of specific log files. - with ZipFile(BytesIO(response.content)) as zip_file: - names = zip_file.namelist() - assert "README.txt" in names - readme = zip_file.read("README.txt").decode("utf-8") - assert "api_server" in readme - assert "WARNING" in readme - - def test_sequential_downloads_both_succeed(self, admin_user: DATestUser) -> None: - # The export lock must be released once a download completes; a wedged - # lock would 429 every subsequent request until the server restarts. - for _ in range(2): - response = client.get( - f"{API_SERVER_URL}/admin/log-export/download", - headers=admin_user.headers, - ) - assert response.status_code == 200 - - def test_non_admin_cannot_download_log_zip(self, basic_user: DATestUser) -> None: - response = client.get( - f"{API_SERVER_URL}/admin/log-export/download", - headers=basic_user.headers, - ) - assert response.status_code == 403 - def test_async_export_flow(self, admin_user: DATestUser) -> None: # Start an export. response = client.post( diff --git a/backend/tests/unit/conftest.py b/backend/tests/unit/conftest.py new file mode 100644 index 00000000000..485f49a1a38 --- /dev/null +++ b/backend/tests/unit/conftest.py @@ -0,0 +1,34 @@ +"""Unit-suite conftest. + +Unit tests assume OSS resolution unless they opt into EE via the shared +``enable_ee`` fixture (see ``backend/tests/conftest.py``). +""" + +from collections.abc import Generator + +import pytest + +from onyx.utils.variable_functionality import ( + fetch_versioned_implementation, + global_version, +) + + +@pytest.fixture(autouse=True) +def _reset_leaked_ee_state() -> Generator[None, None, None]: + """Undoes EE state leaked into the process by import side effects. + + ``set_is_ee_based_on_env_variable()`` runs at module level in ``onyx.main`` + and every ``background/celery/versioned_apps`` module, and flips the + process-global EE flag whenever license enforcement is on (its default). A + unit test whose import chain reaches one of those modules therefore silently + switches every later test in the worker to EE resolution, breaking + OSS-asserting tests order-dependently. Runs before ``enable_ee`` (autouse + fixtures are instantiated first), so opting in still works. + """ + if global_version.is_ee_version(): + global_version.unset_ee() + # Entries resolved while the flag was flipped point at EE + # implementations; drop them along with the flag. + fetch_versioned_implementation.cache_clear() + yield diff --git a/backend/tests/unit/ee/onyx/server/log_export/test_download_endpoint.py b/backend/tests/unit/ee/onyx/server/log_export/test_download_endpoint.py deleted file mode 100644 index 000eaf6494a..00000000000 --- a/backend/tests/unit/ee/onyx/server/log_export/test_download_endpoint.py +++ /dev/null @@ -1,225 +0,0 @@ -import asyncio -from collections.abc import MutableMapping, Sequence -from pathlib import Path -from typing import Any - -import pytest -from starlette.requests import ClientDisconnect - -from ee.onyx.server.log_export import api as log_export_api -from ee.onyx.server.log_export.api import _ExpiringLock, download_api_server_logs -from ee.onyx.server.log_export.collection import BuiltLogZip -from onyx.error_handling.error_codes import OnyxErrorCode -from onyx.error_handling.exceptions import OnyxError - - -def _use_tmp_log_dir(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: - (tmp_path / "onyx_debug.log").write_text("a log line\n") - monkeypatch.setattr( - log_export_api, "get_default_log_directories", lambda: [tmp_path] - ) - - -def _http_scope(spec_version: str) -> dict[str, Any]: - return { - "type": "http", - "asgi": {"version": "3.0", "spec_version": spec_version}, - "method": "GET", - "path": "/admin/log-export/download", - "headers": [], - } - - -async def _never_receive() -> dict[str, Any]: - await asyncio.Event().wait() - raise AssertionError("Unreachable.") - - -def test_expiring_lock_ttl_steal_and_stale_release() -> None: - now = [0.0] - lock = _ExpiringLock(ttl_seconds=60.0, clock=lambda: now[0]) - - first = lock.try_acquire() - assert first is not None - assert lock.try_acquire() is None - - # Expiry lets a new holder steal the hold. - now[0] = 61.0 - second = lock.try_acquire() - assert second is not None - - # The stale holder's release must not free the new hold. - lock.release(first) - assert lock.held() - - lock.release(second) - assert not lock.held() - # A duplicate release stays a no-op. - lock.release(second) - assert not lock.held() - - -def test_rejected_while_export_in_progress() -> None: - assert not log_export_api._EXPORT_LOCK.held(), "Lock leaked from another test." - token = log_export_api._EXPORT_LOCK.try_acquire() - assert token is not None - try: - with pytest.raises(OnyxError) as exc_info: - download_api_server_logs() - assert exc_info.value.error_code == OnyxErrorCode.RATE_LIMITED - finally: - log_export_api._EXPORT_LOCK.release(token) - - -def test_lock_released_when_build_fails(monkeypatch: pytest.MonkeyPatch) -> None: - assert not log_export_api._EXPORT_LOCK.held(), "Lock leaked from another test." - - def failing_build(*args: object, **kwargs: object) -> None: # noqa: ARG001 - raise OSError("Disk exploded.") - - monkeypatch.setattr(log_export_api, "build_log_zip", failing_build) - - with pytest.raises(OSError): - download_api_server_logs() - - assert not log_export_api._EXPORT_LOCK.held() - - -def test_lock_released_even_if_buffer_close_fails( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - assert not log_export_api._EXPORT_LOCK.held(), "Lock leaked from another test." - - _use_tmp_log_dir(monkeypatch, tmp_path) - real_build = log_export_api.build_log_zip - - def build_with_broken_close( - log_directories: Sequence[Path], scope_note: str - ) -> BuiltLogZip: - built = real_build(log_directories, scope_note) - - def broken_close() -> None: - raise OSError("Close failed.") - - built.zip_buffer.close = broken_close # ty: ignore[invalid-assignment] - return built - - monkeypatch.setattr(log_export_api, "build_log_zip", build_with_broken_close) - - response = download_api_server_logs() - assert response.background is not None - with pytest.raises(OSError): - asyncio.run(response.background()) - - assert not log_export_api._EXPORT_LOCK.held() - - -def test_cleanup_releases_lock_without_body_iteration( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - """ - Covers the client-disconnected-before-first-chunk path: the response's - background task alone must release the lock, without the body generator - ever running. - """ - assert not log_export_api._EXPORT_LOCK.held(), "Lock leaked from another test." - - _use_tmp_log_dir(monkeypatch, tmp_path) - - response = download_api_server_logs() - - assert log_export_api._EXPORT_LOCK.held(), ( - "Lock must be held while the response is pending." - ) - assert response.background is not None - - asyncio.run(response.background()) - - assert not log_export_api._EXPORT_LOCK.held() - - -def test_lock_released_when_iterator_raises_mid_stream( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - """ - Starlette skips background tasks when the body iterator raises, so the - release must come from the generator's own ``finally``. Drives the real - ``StreamingResponse.__call__`` rather than invoking hooks by hand. - """ - assert not log_export_api._EXPORT_LOCK.held(), "Lock leaked from another test." - - _use_tmp_log_dir(monkeypatch, tmp_path) - real_build = log_export_api.build_log_zip - - def build_with_broken_read( - log_directories: Sequence[Path], scope_note: str - ) -> BuiltLogZip: - built = real_build(log_directories, scope_note) - - def broken_read(size: int = -1) -> bytes: # noqa: ARG001 - raise OSError("Read failed.") - - built.zip_buffer.read = broken_read # ty: ignore[invalid-assignment] - return built - - monkeypatch.setattr(log_export_api, "build_log_zip", build_with_broken_read) - - response = download_api_server_logs() - - async def send(_message: MutableMapping[str, Any]) -> None: - return None - - with pytest.raises(OSError): - asyncio.run(response(_http_scope("2.3"), _never_receive, send)) - - assert not log_export_api._EXPORT_LOCK.held() - - -def test_lock_released_on_client_disconnect_mid_stream( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - assert not log_export_api._EXPORT_LOCK.held(), "Lock leaked from another test." - - _use_tmp_log_dir(monkeypatch, tmp_path) - - response = download_api_server_logs() - - async def receive() -> dict[str, Any]: - return {"type": "http.disconnect"} - - async def blocked_send(_message: MutableMapping[str, Any]) -> None: - # Simulates a stalled transport so the stream cannot finish before the - # disconnect message is observed. - await asyncio.Event().wait() - - asyncio.run(response(_http_scope("2.3"), receive, blocked_send)) - - assert not log_export_api._EXPORT_LOCK.held() - - -def test_leaked_hold_recovers_via_ttl_under_asgi_2_4( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - """ - Under ASGI >= 2.4 semantics a failing ``send`` makes Starlette raise - ``ClientDisconnect`` before running background tasks; the generator never - started either, so no prompt hook runs and only the TTL recovers the lock. - """ - now = [0.0] - test_lock = _ExpiringLock(ttl_seconds=60.0, clock=lambda: now[0]) - monkeypatch.setattr(log_export_api, "_EXPORT_LOCK", test_lock) - _use_tmp_log_dir(monkeypatch, tmp_path) - - response = download_api_server_logs() - - async def failing_send(_message: MutableMapping[str, Any]) -> None: - raise OSError("Transport closed.") - - with pytest.raises(ClientDisconnect): - asyncio.run(response(_http_scope("2.4"), _never_receive, failing_send)) - - # No prompt hook ran; the hold leaks until the TTL expires. - assert test_lock.held() - assert test_lock.try_acquire() is None - now[0] = 61.0 - assert test_lock.try_acquire() is not None diff --git a/backend/tests/unit/ee/onyx/server/log_export/test_expiring_lock.py b/backend/tests/unit/ee/onyx/server/log_export/test_expiring_lock.py new file mode 100644 index 00000000000..c99ac649b18 --- /dev/null +++ b/backend/tests/unit/ee/onyx/server/log_export/test_expiring_lock.py @@ -0,0 +1,25 @@ +from ee.onyx.server.log_export.api import _ExpiringLock + + +def test_expiring_lock_ttl_steal_and_stale_release() -> None: + now = [0.0] + lock = _ExpiringLock(ttl_seconds=60.0, clock=lambda: now[0]) + + first = lock.try_acquire() + assert first is not None + assert lock.try_acquire() is None + + # Expiry lets a new holder steal the hold. + now[0] = 61.0 + second = lock.try_acquire() + assert second is not None + + # The stale holder's release must not free the new hold. + lock.release(first) + assert lock.held() + + lock.release(second) + assert not lock.held() + # A duplicate release stays a no-op. + lock.release(second) + assert not lock.held() diff --git a/backend/tests/unit/ee/onyx/server/middleware/test_tier_gate.py b/backend/tests/unit/ee/onyx/server/middleware/test_tier_gate.py index 32f5a1edbff..0363892027e 100644 --- a/backend/tests/unit/ee/onyx/server/middleware/test_tier_gate.py +++ b/backend/tests/unit/ee/onyx/server/middleware/test_tier_gate.py @@ -123,9 +123,7 @@ async def test_business_blocked_from_log_export( ) -> None: mock_get_tier.return_value = Tier.BUSINESS middleware, call_next = middleware_harness - response = await middleware( - _make_request("/api/admin/log-export/download"), call_next - ) + response = await middleware(_make_request("/api/admin/log-export"), call_next) assert response.status_code == 402 @@ -136,9 +134,7 @@ async def test_enterprise_passes_log_export( ) -> None: mock_get_tier.return_value = Tier.ENTERPRISE middleware, call_next = middleware_harness - response = await middleware( - _make_request("/api/admin/log-export/download"), call_next - ) + response = await middleware(_make_request("/api/admin/log-export"), call_next) assert response.status_code == 200 diff --git a/backend/tests/unit/onyx/utils/test_os_reaper.py b/backend/tests/unit/onyx/utils/test_os_reaper.py new file mode 100644 index 00000000000..3cb33b269ce --- /dev/null +++ b/backend/tests/unit/onyx/utils/test_os_reaper.py @@ -0,0 +1,204 @@ +"""Validate subreaper adoption + zombie draining (Linux-only). + +Each scenario runs in an isolated subprocess so the subreaper flag and the +waitpid(-1) drains never touch the pytest process. Harnesses end with a +kill-everything finally so a failed assertion can't leave a sleeping +grandchild holding the captured stdout/stderr pipes (which would turn the +real failure into a subprocess timeout).""" + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +_BACKEND_DIR = Path(__file__).parents[4] + +_CLEANUP = """ +def _kill_live_children(): + import os + import signal + + from onyx.utils.os_reaper import _live_child_pids, reap_exited_children + + for pid in _live_child_pids(): + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + reap_exited_children() +""" + +_HARNESS = ( + _CLEANUP + + """ +import os +import sys +import time + +from onyx.utils.os_reaper import become_child_subreaper, reap_exited_children + +try: + assert become_child_subreaper(), "prctl(PR_SET_CHILD_SUBREAPER) failed" + + child_pid = os.fork() + if child_pid == 0: + # child: orphan a grandchild while it is still running, then exit + grandchild_pid = os.fork() + if grandchild_pid == 0: + time.sleep(0.2) # outlive the parent so re-parenting happens while alive + os._exit(0) + os._exit(0) + + os.waitpid(child_pid, 0) # reap the direct child; only the orphan remains + + # the orphaned grandchild re-parents to us (the subreaper) and zombifies + # on exit; without the flag it would re-parent to PID 1 instead + deadline = time.monotonic() + 10 + reaped = 0 + while reaped == 0 and time.monotonic() < deadline: + reaped = reap_exited_children() + time.sleep(0.05) + + assert reaped == 1, f"expected to reap exactly the orphaned grandchild, got {reaped}" + assert reap_exited_children() == 0, "no children should remain" + print("ok") +finally: + _kill_live_children() +""" +) + +_EXIT_HARNESS = ( + _CLEANUP + + """ +import os +import sys +import time + +from onyx.utils.os_reaper import ( + _live_child_pids, + become_child_subreaper, + reap_children_before_exit, +) + +try: + assert become_child_subreaper(), "prctl(PR_SET_CHILD_SUBREAPER) failed" + + # orphan a grandchild that stays RUNNING well past the drain + child_pid = os.fork() + if child_pid == 0: + grandchild_pid = os.fork() + if grandchild_pid == 0: + time.sleep(60) + os._exit(0) + os._exit(0) + + os.waitpid(child_pid, 0) + + deadline = time.monotonic() + 10 + while not _live_child_pids() and time.monotonic() < deadline: + time.sleep(0.05) # wait for the orphan to re-parent to us + assert _live_child_pids(), "orphan never re-parented to the subreaper" + + reaped = reap_children_before_exit(grace_seconds=0.2) + assert reaped >= 1, f"expected the running orphan to be killed and reaped, got {reaped}" + assert not _live_child_pids(), "no running children should remain" + print("ok") +finally: + _kill_live_children() +""" +) + +_SIGTERM_HARNESS = ( + _CLEANUP + + """ +import os +import signal +import sys +import time + +from onyx.utils.os_reaper import ( + _live_child_pids, + become_child_subreaper, + install_sigterm_drain, +) + +try: + assert become_child_subreaper() + install_sigterm_drain() + + # orphan a long-running grandchild, print its pid for the outer test + child_pid = os.fork() + if child_pid == 0: + grandchild_pid = os.fork() + if grandchild_pid == 0: + time.sleep(60) + os._exit(0) + print(grandchild_pid, flush=True) + os._exit(0) + + os.waitpid(child_pid, 0) + deadline = time.monotonic() + 10 + while not _live_child_pids() and time.monotonic() < deadline: + time.sleep(0.05) + assert _live_child_pids(), "orphan never re-parented" + + os.kill(os.getpid(), signal.SIGTERM) # watchdog cancellation + time.sleep(30) # never reached: the handler drains and exits 143 +finally: + _kill_live_children() +""" +) + + +def _run_harness(code: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + timeout=30, + env={"PYTHONPATH": str(_BACKEND_DIR)}, + ) + + +@pytest.mark.skipif( + sys.platform != "linux", reason="prctl/waitpid semantics are Linux-only" +) +def test_subreaper_adopts_and_drains_orphaned_grandchild() -> None: + result = _run_harness(_HARNESS) + assert result.returncode == 0, result.stderr + assert "ok" in result.stdout + + +@pytest.mark.skipif( + sys.platform != "linux", reason="prctl/waitpid semantics are Linux-only" +) +def test_exit_drain_kills_and_reaps_running_orphan() -> None: + result = _run_harness(_EXIT_HARNESS) + assert result.returncode == 0, result.stderr + assert "ok" in result.stdout + + +@pytest.mark.skipif( + sys.platform != "linux", reason="prctl/waitpid semantics are Linux-only" +) +def test_sigterm_cancellation_drains_running_orphan() -> None: + result = _run_harness(_SIGTERM_HARNESS) + assert result.returncode == 143, result.stderr + orphan_pid = int(result.stdout.split()[0]) + # the drain must have killed the orphan — not left it running for PID 1 + with pytest.raises(OSError): + os.kill(orphan_pid, 0) + + +def test_reap_exited_children_noop_without_children() -> None: + """The drain must be safe to call from a process with no children at all.""" + harness = ( + "from onyx.utils.os_reaper import reap_exited_children\n" + "assert reap_exited_children() == 0\n" + "print('ok')\n" + ) + result = _run_harness(harness) + assert result.returncode == 0, result.stderr + assert "ok" in result.stdout diff --git a/web/lib/opal/src/components/table/components.tsx b/web/lib/opal/src/components/table/components.tsx index 473676acf7d..f3e9f9a8f00 100644 --- a/web/lib/opal/src/components/table/components.tsx +++ b/web/lib/opal/src/components/table/components.tsx @@ -514,7 +514,6 @@ export function Table(props: DataTableProps) { sortableId={rowId} selected={row.getIsSelected()} data-clickable={onRowClick ? true : undefined} - role={onRowClick ? "button" : undefined} tabIndex={onRowClick ? 0 : undefined} aria-label={ onRowClick ? getRowLabel?.(row.original) : undefined diff --git a/web/src/app/craft/hooks/loadSessionRestore.test.ts b/web/src/app/craft/hooks/loadSessionRestore.test.ts index 5089431915f..c1f5c21a133 100644 --- a/web/src/app/craft/hooks/loadSessionRestore.test.ts +++ b/web/src/app/craft/hooks/loadSessionRestore.test.ts @@ -22,7 +22,9 @@ function sleepingSession(): unknown { }; } -function runningSession(nextjsPort: number | null = null): unknown { +function runningSession( + nextjsPort: number | null = null +): Record { return { id: SESSION_ID, status: "active", @@ -469,6 +471,34 @@ describe("loadSession restore status", () => { expect(session?.activeTurnLocalOwner).toBe(false); }); + it("preserves stale-skill state while loading a pre-provisioned turn", async () => { + mockedApi.fetchSession.mockResolvedValue({ + ...runningSession(), + skills_stale: true, + } as never); + useBuildSessionStore.getState().createSession(SESSION_ID, { + status: "running", + messages: [ + { + id: "local-user", + type: "user", + content: "hello", + timestamp: new Date(), + }, + ], + skillsStale: false, + isLoaded: false, + }); + + await useBuildSessionStore + .getState() + .loadSession(SESSION_ID, { force: true }); + + expect( + useBuildSessionStore.getState().sessions.get(SESSION_ID)?.skillsStale + ).toBe(false); + }); + it("clears stale turn metadata when active turn lookup says no turn is running", async () => { mockedApi.fetchSession.mockResolvedValue(runningSession() as never); mockedApi.fetchActiveTurn.mockResolvedValue(null as never); @@ -568,6 +598,22 @@ describe("loadSession preferPersisted (interrupt reconciliation)", () => { expect(session?.activeTurnLocalOwner).toBe(false); }); + it("reconciles stale skills when an interrupted turn settles", async () => { + seedInterruptedSession(); + mockedApi.fetchSession.mockResolvedValue({ + ...runningSession(), + skills_stale: true, + } as never); + + await useBuildSessionStore + .getState() + .loadSession(SESSION_ID, { force: true, preferPersisted: true }); + + expect( + useBuildSessionStore.getState().sessions.get(SESSION_ID)?.skillsStale + ).toBe(true); + }); + it("keeps the stale local transcript without preferPersisted (the bug)", async () => { seedInterruptedSession(); diff --git a/web/src/app/craft/hooks/useBuildSessionStore.ts b/web/src/app/craft/hooks/useBuildSessionStore.ts index a02d5163e0b..468062ea779 100644 --- a/web/src/app/craft/hooks/useBuildSessionStore.ts +++ b/web/src/app/craft/hooks/useBuildSessionStore.ts @@ -1611,7 +1611,9 @@ export const useBuildSessionStore = create()((set, get) => ({ sandbox, agentProvider: sessionData.agent_provider, agentModel: sessionData.agent_model, - skillsStale: sessionData.skills_stale, + // Persisted loads reconcile stale state. Optimistic welcome loads keep + // their live local state until the turn settles. + ...(useDbMessages && { skillsStale: sessionData.skills_stale }), origin: sessionData.origin, activeTurnId: resolvedActiveTurnId, activeTurnIndex: resolvedActiveTurnIndex, diff --git a/web/src/app/ee/admin/performance/usage/FeedbackChart.tsx b/web/src/app/ee/admin/performance/usage/FeedbackChart.tsx deleted file mode 100644 index 8cb98063c4e..00000000000 --- a/web/src/app/ee/admin/performance/usage/FeedbackChart.tsx +++ /dev/null @@ -1,75 +0,0 @@ -import SvgSimpleLoader from "@opal/icons/simple-loader"; -import { getDatesList, useQueryAnalytics } from "../lib"; -import { Text } from "@opal/components"; -import Title from "@/components/ui/title"; - -import { DateRangePickerValue } from "@/components/dateRangeSelectors/AdminDateRangeSelector"; -import CardSection from "@/components/admin/CardSection"; -import { AreaChartDisplay } from "@/components/ui/areaChart"; - -export function FeedbackChart({ - timeRange, -}: { - timeRange: DateRangePickerValue; -}) { - const { - data: queryAnalyticsData, - isLoading: isQueryAnalyticsLoading, - error: queryAnalyticsError, - } = useQueryAnalytics(timeRange); - - let chart; - if (isQueryAnalyticsLoading) { - chart = ( -
- -
- ); - } else if ( - !queryAnalyticsData || - queryAnalyticsData[0] === undefined || - queryAnalyticsError - ) { - chart = ( -
-

Failed to fetch feedback data...

-
- ); - } else { - const initialDate = timeRange.from || new Date(queryAnalyticsData[0].date); - const dateRange = getDatesList(initialDate); - - const dateToQueryAnalytics = new Map( - queryAnalyticsData.map((queryAnalyticsEntry) => [ - queryAnalyticsEntry.date, - queryAnalyticsEntry, - ]) - ); - - chart = ( - { - const queryAnalyticsForDate = dateToQueryAnalytics.get(dateStr); - return { - Day: dateStr, - "Positive Feedback": queryAnalyticsForDate?.total_likes || 0, - "Negative Feedback": queryAnalyticsForDate?.total_dislikes || 0, - }; - })} - categories={["Positive Feedback", "Negative Feedback"]} - index="Day" - colors={["indigo", "fuchsia"]} - yAxisWidth={60} - /> - ); - } - - return ( - - Feedback - Thumbs Up / Thumbs Down over time - {chart} - - ); -} diff --git a/web/src/app/ee/admin/performance/usage/OnyxBotChart.tsx b/web/src/app/ee/admin/performance/usage/OnyxBotChart.tsx deleted file mode 100644 index 7c7e0ae6b23..00000000000 --- a/web/src/app/ee/admin/performance/usage/OnyxBotChart.tsx +++ /dev/null @@ -1,76 +0,0 @@ -import SvgSimpleLoader from "@opal/icons/simple-loader"; -import { getDatesList, useOnyxBotAnalytics } from "../lib"; -import { DateRangePickerValue } from "@/components/dateRangeSelectors/AdminDateRangeSelector"; -import { Text } from "@opal/components"; -import Title from "@/components/ui/title"; -import CardSection from "@/components/admin/CardSection"; -import { AreaChartDisplay } from "@/components/ui/areaChart"; - -export function OnyxBotChart({ - timeRange, -}: { - timeRange: DateRangePickerValue; -}) { - const { - data: onyxBotAnalyticsData, - isLoading: isOnyxBotAnalyticsLoading, - error: onyxBotAnalyticsError, - } = useOnyxBotAnalytics(timeRange); - - let chart; - if (isOnyxBotAnalyticsLoading) { - chart = ( -
- -
- ); - } else if ( - !onyxBotAnalyticsData || - onyxBotAnalyticsData[0] == undefined || - onyxBotAnalyticsError - ) { - chart = ( -
-

Failed to fetch feedback data...

-
- ); - } else { - const initialDate = - timeRange.from || new Date(onyxBotAnalyticsData[0].date); - const dateRange = getDatesList(initialDate); - - const dateToOnyxBotAnalytics = new Map( - onyxBotAnalyticsData.map((onyxBotAnalyticsEntry) => [ - onyxBotAnalyticsEntry.date, - onyxBotAnalyticsEntry, - ]) - ); - - chart = ( - { - const onyxBotAnalyticsForDate = dateToOnyxBotAnalytics.get(dateStr); - return { - Day: dateStr, - "Total Queries": onyxBotAnalyticsForDate?.total_queries || 0, - "Automatically Resolved": - onyxBotAnalyticsForDate?.auto_resolved || 0, - }; - })} - categories={["Total Queries", "Automatically Resolved"]} - index="Day" - colors={["indigo", "fuchsia"]} - yAxisWidth={60} - /> - ); - } - - return ( - - Slack Channel - Total Queries vs Auto Resolved - {chart} - - ); -} diff --git a/web/src/app/ee/admin/performance/usage/PersonaMessagesChart.tsx b/web/src/app/ee/admin/performance/usage/PersonaMessagesChart.tsx deleted file mode 100644 index 0c64e0aacc8..00000000000 --- a/web/src/app/ee/admin/performance/usage/PersonaMessagesChart.tsx +++ /dev/null @@ -1,235 +0,0 @@ -import SvgSimpleLoader from "@opal/icons/simple-loader"; -import { X, Search } from "lucide-react"; -import { - getDatesList, - usePersonaMessages, - usePersonaUniqueUsers, -} from "../lib"; -import { DateRangePickerValue } from "@/components/dateRangeSelectors/AdminDateRangeSelector"; -import { Text } from "@opal/components"; -import Title from "@/components/ui/title"; -import CardSection from "@/components/admin/CardSection"; -import { AreaChartDisplay } from "@/components/ui/areaChart"; -import { - Select, - SelectContent, - SelectItem, - SelectTrigger, - SelectValue, -} from "@/components/ui/select"; -import { useState, useMemo, useEffect } from "react"; -import { Agent } from "@/lib/agents/types"; - -export function PersonaMessagesChart({ - availablePersonas, - timeRange, -}: { - availablePersonas: Agent[]; - timeRange: DateRangePickerValue; -}) { - const [selectedPersonaId, setSelectedPersonaId] = useState< - number | undefined - >(undefined); - const [searchQuery, setSearchQuery] = useState(""); - const [highlightedIndex, setHighlightedIndex] = useState(-1); - - const { - data: personaMessagesData, - isLoading: isPersonaMessagesLoading, - error: personaMessagesError, - } = usePersonaMessages(selectedPersonaId, timeRange); - - const { - data: personaUniqueUsersData, - isLoading: isPersonaUniqueUsersLoading, - error: personaUniqueUsersError, - } = usePersonaUniqueUsers(selectedPersonaId, timeRange); - - const isLoading = isPersonaMessagesLoading || isPersonaUniqueUsersLoading; - const hasError = personaMessagesError || personaUniqueUsersError; - - const filteredPersonaList = useMemo(() => { - if (!availablePersonas) return []; - return availablePersonas.filter((persona) => - persona.name.toLowerCase().includes(searchQuery.toLowerCase()) - ); - }, [availablePersonas, searchQuery]); - - const handleKeyDown = (e: React.KeyboardEvent) => { - e.stopPropagation(); - - switch (e.key) { - case "ArrowDown": - e.preventDefault(); - setHighlightedIndex((prev) => - prev < filteredPersonaList.length - 1 ? prev + 1 : prev - ); - break; - case "ArrowUp": - e.preventDefault(); - setHighlightedIndex((prev) => (prev > 0 ? prev - 1 : prev)); - break; - case "Enter": - if ( - highlightedIndex >= 0 && - highlightedIndex < filteredPersonaList.length - ) { - const filteredPersona = filteredPersonaList[highlightedIndex]; - if (filteredPersona !== undefined) { - setSelectedPersonaId(filteredPersona.id); - setSearchQuery(""); - setHighlightedIndex(-1); - } - } - break; - case "Escape": - setSearchQuery(""); - setHighlightedIndex(-1); - break; - } - }; - - // Reset highlight when search query changes - useEffect(() => { - setHighlightedIndex(-1); - }, [searchQuery]); - - const chartData = useMemo(() => { - if ( - !personaMessagesData?.length || - !personaUniqueUsersData?.length || - selectedPersonaId === undefined - ) { - return null; - } - - const initialDate = - timeRange.from || - new Date( - Math.min( - ...personaMessagesData.map((entry) => new Date(entry.date).getTime()) - ) - ); - const dateRange = getDatesList(initialDate); - - // Create maps for messages and unique users data - const messagesMap = new Map( - personaMessagesData.map((entry) => [entry.date, entry]) - ); - const uniqueUsersMap = new Map( - personaUniqueUsersData.map((entry) => [entry.date, entry]) - ); - - return dateRange.map((dateStr) => { - const messageData = messagesMap.get(dateStr); - const uniqueUserData = uniqueUsersMap.get(dateStr); - return { - Day: dateStr, - Messages: messageData?.total_messages || 0, - "Unique Users": uniqueUserData?.unique_users || 0, - }; - }); - }, [ - personaMessagesData, - personaUniqueUsersData, - timeRange.from, - selectedPersonaId, - ]); - - let content; - if (isLoading) { - content = ( -
- -
- ); - } else if (!availablePersonas || hasError) { - content = ( -
-

Failed to fetch data...

-
- ); - } else if (selectedPersonaId === undefined) { - content = ( -
-

Select an agent to view analytics

-
- ); - } else if (!personaMessagesData?.length) { - content = ( -
-

- No data found for selected agent in the specified time range -

-
- ); - } else if (chartData) { - content = ( - - ); - } - - return ( - - Agent Analytics -
- - Messages and unique users per day for the selected agent - -
- setSearchQuery(e.target.value)} - onClick={(e) => e.stopPropagation()} - onMouseDown={(e) => e.stopPropagation()} - onKeyDown={handleKeyDown} - /> - {searchQuery && ( - { - setSearchQuery(""); - setHighlightedIndex(-1); - }} - /> - )} -
- {filteredPersonaList.map((persona, index) => ( - setHighlightedIndex(index)} - > - {persona.name} - - ))} - - -
- - {content} - - ); -} diff --git a/web/src/app/ee/admin/performance/usage/QueryPerformanceChart.tsx b/web/src/app/ee/admin/performance/usage/QueryPerformanceChart.tsx deleted file mode 100644 index 25572ed2b7f..00000000000 --- a/web/src/app/ee/admin/performance/usage/QueryPerformanceChart.tsx +++ /dev/null @@ -1,105 +0,0 @@ -"use client"; - -import { DateRangePickerValue } from "@/components/dateRangeSelectors/AdminDateRangeSelector"; -import { getDatesList, useQueryAnalytics, useUserAnalytics } from "../lib"; -import SvgSimpleLoader from "@opal/icons/simple-loader"; -import { AreaChartDisplay } from "@/components/ui/areaChart"; -import Title from "@/components/ui/title"; -import { Text } from "@opal/components"; -import CardSection from "@/components/admin/CardSection"; - -export function QueryPerformanceChart({ - timeRange, -}: { - timeRange: DateRangePickerValue; -}) { - const { - data: queryAnalyticsData, - isLoading: isQueryAnalyticsLoading, - error: queryAnalyticsError, - } = useQueryAnalytics(timeRange); - const { - data: userAnalyticsData, - isLoading: isUserAnalyticsLoading, - error: userAnalyticsError, - } = useUserAnalytics(timeRange); - - let chart; - if (isQueryAnalyticsLoading || isUserAnalyticsLoading) { - chart = ( -
- -
- ); - } else if ( - !queryAnalyticsData || - queryAnalyticsData[0] === undefined || - !userAnalyticsData || - queryAnalyticsError || - userAnalyticsError - ) { - chart = ( -
-

Failed to fetch query data...

-
- ); - } else { - const initialDate = timeRange.from || new Date(queryAnalyticsData[0].date); - const dateRange = getDatesList(initialDate); - - const dateToQueryAnalytics = new Map( - queryAnalyticsData.map((queryAnalyticsEntry) => [ - queryAnalyticsEntry.date, - queryAnalyticsEntry, - ]) - ); - const dateToUserAnalytics = new Map( - userAnalyticsData.map((userAnalyticsEntry) => [ - userAnalyticsEntry.date, - userAnalyticsEntry, - ]) - ); - - chart = ( - { - const queryAnalyticsForDate = dateToQueryAnalytics.get(dateStr); - const userAnalyticsForDate = dateToUserAnalytics.get(dateStr); - return { - Day: dateStr, - Queries: queryAnalyticsForDate?.total_queries || 0, - "Unique Users": userAnalyticsForDate?.total_active_users || 0, - }; - })} - categories={["Queries", "Unique Users"]} - index="Day" - colors={["indigo", "fuchsia"]} - yAxisFormatter={(number: number) => - new Intl.NumberFormat("en-US", { - notation: "standard", - maximumFractionDigits: 0, - }).format(number) - } - xAxisFormatter={(dateStr: string) => { - const date = new Date(dateStr); - return date.toLocaleDateString("en-US", { - month: "short", - day: "numeric", - }); - }} - yAxisWidth={60} - allowDecimals={false} - /> - ); - } - - return ( - - Usage - Usage over time - {chart} - - ); -} diff --git a/web/src/app/ee/admin/performance/usage/UsageReports.tsx b/web/src/app/ee/admin/performance/usage/UsageReports.tsx deleted file mode 100644 index 3dca7e390a2..00000000000 --- a/web/src/app/ee/admin/performance/usage/UsageReports.tsx +++ /dev/null @@ -1,454 +0,0 @@ -"use client"; - -import { format } from "date-fns"; -import { errorHandlingFetcher } from "@/lib/fetcher"; - -import { FiDownload } from "react-icons/fi"; -import { - Table, - TableBody, - TableCell, - TableHead, - TableHeader, - TableRow, -} from "@/components/ui/table"; -import { Text } from "@opal/components"; -import Title from "@/components/ui/title"; -import { Spacer } from "@opal/components"; -import Button from "@/refresh-components/buttons/Button"; -import { Button as OpalButton } from "@opal/components"; -import useSWR from "swr"; -import { SWR_KEYS } from "@/lib/swr-keys"; -import React, { useState } from "react"; -import { UsageReport } from "./types"; -import SvgSimpleLoader from "@opal/icons/simple-loader"; -import Link from "next/link"; -import { humanReadableFormat, humanReadableFormatWithTime } from "@opal/time"; -import { ErrorCallout } from "@/components/ErrorCallout"; -import { PageSelector } from "@/components/PageSelector"; -import { Divider } from "@opal/components"; -import { DateRangePickerValue } from "../../../../../components/dateRangeSelectors/AdminDateRangeSelector"; -import { Popover } from "@opal/components"; -import Calendar from "@/refresh-components/Calendar"; -import { cn } from "@opal/utils"; -import { Spinner } from "@/components/Spinner"; -import { SvgCalendar, SvgDownloadCloud } from "@opal/icons"; - -function GenerateReportInput({ - onReportGenerated, - isWaitingForReport, -}: { - onReportGenerated: () => void; - isWaitingForReport: boolean; -}) { - const [dateRange, setDateRange] = useState( - undefined - ); - const [isLoading, setIsLoading] = useState(false); - - const [errorOccurred, setErrorOccurred] = useState(null); - - const requestReport = async () => { - setIsLoading(true); - setErrorOccurred(null); - try { - let period_from: string | null = null; - let period_to: string | null = null; - - if (dateRange?.selectValue != "allTime" && dateRange?.from) { - period_from = dateRange?.from?.toISOString(); - period_to = dateRange?.to?.toISOString() ?? new Date().toISOString(); - } - - const res = await fetch("/api/admin/usage-report", { - method: "POST", - credentials: "include", - headers: { - "Content-Type": "application/json", - }, - body: JSON.stringify({ - period_from: period_from, - period_to: period_to, - }), - }); - - if (!res.ok) { - throw Error(`Received an error: ${res.statusText}`); - } - - // Trigger refresh of the reports list - onReportGenerated(); - } catch (e) { - setErrorOccurred(e as Error); - } finally { - setIsLoading(false); - } - }; - - const today = new Date(); - - const lastWeek = new Date(); - lastWeek.setDate(today.getDate() - 7); - - const lastMonth = new Date(); - lastMonth.setMonth(today.getMonth() - 1); - - const lastYear = new Date(); - lastYear.setFullYear(today.getFullYear() - 1); - - return ( -
- Generate Usage Reports - Generate usage statistics for users in the workspace. - -
- - - {/* TODO(@raunakab): migrate to opal Button once className/iconClassName is resolved */} - - - - - range?.from && - setDateRange({ - from: range.from, - to: range.to ?? range.from, - selectValue: "custom", - }) - } - numberOfMonths={2} - disabled={(date) => date > new Date()} - /> -
- { - setDateRange({ - from: lastWeek, - to: new Date(), - selectValue: "lastWeek", - }); - }} - > - Last 7 days - - { - setDateRange({ - from: lastMonth, - to: new Date(), - selectValue: "lastMonth", - }); - }} - > - Last 30 days - - { - setDateRange({ - from: lastYear, - to: new Date(), - selectValue: "lastYear", - }); - }} - > - Last year - - { - setDateRange({ - from: new Date(1970, 0, 1), - to: new Date(), - selectValue: "allTime", - }); - }} - > - All time - -
- - -
- requestReport()} - > - {isWaitingForReport ? "Generating..." : "Generate Report"} - -

- {isWaitingForReport - ? "A report is currently being generated. Please wait..." - : 'Report generation runs in the background. Check the "Previous Reports" section below to download when ready.'} -

- {errorOccurred && ( - - )} -
- ); -} - -const USAGE_REPORT_URL = SWR_KEYS.usageReport; - -function UsageReportsTable({ - refreshTrigger, - isWaitingForReport, - onNewReportDetected, -}: { - refreshTrigger: number; - isWaitingForReport: boolean; - onNewReportDetected: () => void; -}) { - const [page, setPage] = useState(1); - const NUM_IN_PAGE = 10; - const [previousReportCount, setPreviousReportCount] = useState( - null - ); - - const { - data: usageReportsMetadata, - error: usageReportsError, - isLoading: usageReportsIsLoading, - mutate, - } = useSWR(USAGE_REPORT_URL, errorHandlingFetcher, { - refreshInterval: isWaitingForReport ? 3000 : 0, // Poll every 3 seconds when waiting - }); - - // Refresh when refreshTrigger changes - React.useEffect(() => { - if (refreshTrigger > 0) { - mutate(); - } - }, [refreshTrigger, mutate]); - - // Detect when a new report appears - React.useEffect(() => { - if (usageReportsMetadata && previousReportCount !== null) { - if (usageReportsMetadata.length > previousReportCount) { - onNewReportDetected(); - } - } - if (usageReportsMetadata) { - setPreviousReportCount(usageReportsMetadata.length); - } - }, [usageReportsMetadata, previousReportCount, onNewReportDetected]); - - const paginatedReports = usageReportsMetadata - ? usageReportsMetadata - .slice(0) - .reverse() - .slice(NUM_IN_PAGE * (page - 1), NUM_IN_PAGE * page) - : []; - - const totalPages = usageReportsMetadata - ? Math.ceil(usageReportsMetadata.length / NUM_IN_PAGE) - : 0; - - return ( -
- Previous Reports - {usageReportsIsLoading && !isWaitingForReport ? ( -
- -
- ) : usageReportsError ? ( - - ) : ( - <> - - - - Report - Period - Generated By - Time Generated - Download - - - - - {paginatedReports.map((r) => ( - - - {r.report_name.split("_")[1]?.substring(0, 8) || - r.report_name.substring(0, 8)} - - - {r.period_from - ? `${humanReadableFormat( - r.period_from - )} - ${humanReadableFormat(r.period_to!)}` - : "All time"} - - {r.requestor ?? "Auto generated"} - - {humanReadableFormatWithTime(r.time_created)} - - - - - - - - ))} - -
-
-
- { - setPage(newPage); - window.scrollTo({ - top: 0, - left: 0, - behavior: "smooth", - }); - }} - /> -
-
- - )} -
- ); -} - -export default function UsageReports() { - const [refreshTrigger, setRefreshTrigger] = useState(0); - const [isWaitingForReport, setIsWaitingForReport] = useState(false); - const [timeoutMessage, setTimeoutMessage] = useState(null); - const timeoutRef = React.useRef(null); - - const handleReportGenerated = () => { - setRefreshTrigger((prev) => prev + 1); - setIsWaitingForReport(true); - setTimeoutMessage(null); - - // Clear any existing timeout - if (timeoutRef.current) { - clearTimeout(timeoutRef.current); - } - - // Set a 15 second timeout - timeoutRef.current = setTimeout(() => { - setIsWaitingForReport(false); - setTimeoutMessage( - "Report generation is taking longer than expected. The report will continue generating in the background. Please check back in a few minutes." - ); - timeoutRef.current = null; - }, 15000); - }; - - const handleNewReportDetected = () => { - setIsWaitingForReport(false); - setTimeoutMessage(null); - // Clear the timeout if report completed before timeout - if (timeoutRef.current) { - clearTimeout(timeoutRef.current); - timeoutRef.current = null; - } - }; - - // Cleanup on unmount - React.useEffect(() => { - return () => { - if (timeoutRef.current) { - clearTimeout(timeoutRef.current); - } - }; - }, []); - - return ( - <> - {isWaitingForReport && } - <> - - {timeoutMessage && ( -
-
-
- - - -
-
-
- - Report Generation In Progress - -
- -
- {timeoutMessage} -
-
-
-
- )} - - - - - ); -} diff --git a/web/src/app/ee/admin/performance/usage/page.tsx b/web/src/app/ee/admin/performance/usage/page.tsx index 168a957ccd3..ea6075692f9 100644 --- a/web/src/app/ee/admin/performance/usage/page.tsx +++ b/web/src/app/ee/admin/performance/usage/page.tsx @@ -1,45 +1,37 @@ "use client"; import { AdminDateRangeSelector } from "@/components/dateRangeSelectors/AdminDateRangeSelector"; -import { OnyxBotChart } from "@/app/ee/admin/performance/usage/OnyxBotChart"; -import { FeedbackChart } from "@/app/ee/admin/performance/usage/FeedbackChart"; -import { QueryPerformanceChart } from "@/app/ee/admin/performance/usage/QueryPerformanceChart"; -import { PersonaMessagesChart } from "@/app/ee/admin/performance/usage/PersonaMessagesChart"; import { useTimeRange } from "@/app/ee/admin/performance/lib"; -import UsageReports from "@/app/ee/admin/performance/usage/UsageReports"; import PerUserUsagePanel from "@/views/admin/PerUserUsagePanel"; -import { Divider } from "@opal/components"; -import { useAdminAgents } from "@/lib/agents/hooks"; import { ADMIN_ROUTES } from "@/lib/admin-routes"; +import { Divider } from "@opal/components"; import { SettingsLayouts } from "@opal/layouts"; import TokenRateLimitsPanel from "@/app/admin/token-rate-limits/TokenRateLimitsPanel"; const route = ADMIN_ROUTES.USAGE; -export default function AnalyticsPage() { +export default function UsagePage() { const [timeRange, setTimeRange] = useTimeRange(); - const { agents } = useAdminAgents(); return ( - - + + - setTimeRange(value as any)} - /> - - - - setTimeRange(value as any)} + /> + } /> - - - - diff --git a/web/src/lib/usage/userUsage.ts b/web/src/lib/usage/userUsage.ts index 57b824b661a..3fa6693d014 100644 --- a/web/src/lib/usage/userUsage.ts +++ b/web/src/lib/usage/userUsage.ts @@ -54,17 +54,3 @@ export function useUsageExport(range?: { from: Date; to: Date } | undefined) { return { usage: data, isLoading, error, refetch: mutate }; } - -/** Clears a user's usage across active enforcement windows. */ -export async function resetUserUsage(userEmail: string): Promise { - const response = await fetch(SWR_KEYS.adminUsageReset, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ user_email: userEmail }), - }); - - if (!response.ok) { - const data = await response.json().catch(() => null); - throw new Error(data?.detail || data?.error_code || response.statusText); - } -} diff --git a/web/src/sections/usage/SpendByUserTable.test.tsx b/web/src/sections/usage/SpendByUserTable.test.tsx index 8f355162689..e60807f6eba 100644 --- a/web/src/sections/usage/SpendByUserTable.test.tsx +++ b/web/src/sections/usage/SpendByUserTable.test.tsx @@ -22,7 +22,7 @@ test("opens a user from the keyboard", () => { /> ); - const row = screen.getByRole("button", { + const row = screen.getByRole("row", { name: "View usage details for ada@example.com", }); row.focus(); diff --git a/web/src/views/admin/PerUserUsagePanel.tsx b/web/src/views/admin/PerUserUsagePanel.tsx index 314cd477e26..c3f3c8c4282 100644 --- a/web/src/views/admin/PerUserUsagePanel.tsx +++ b/web/src/views/admin/PerUserUsagePanel.tsx @@ -1,312 +1,240 @@ "use client"; -import { useEffect, useMemo, useState } from "react"; -import { Button, Card, InputTypeIn, MessageCard, Text } from "@opal/components"; -import { SvgChevronLeft, SvgChevronRight, SvgX } from "@opal/icons"; -import { PageLoader, toast } from "@opal/layouts"; -import { - resetUserUsage, - useUsageExport, - UsageExportUser, -} from "@/lib/usage/userUsage"; - -const PAGE_SIZE = 10; - -type SortKey = - | "email" - | "input_tokens" - | "output_tokens" - | "cache_read_tokens" - | "cost_cents"; -type SortDir = "asc" | "desc"; - -function formatTokens(n: number): string { - return n.toLocaleString(); -} - -function formatCost(cents: number): string { - return `$${(cents / 100).toFixed(2)}`; -} - -function sortValue(user: UsageExportUser, key: SortKey): number | string { - if (key === "email") return user.email.toLowerCase(); - return user.totals[key]; +import React, { useMemo, useState } from "react"; +import { Card, MessageCard, Text } from "@opal/components"; +import { SvgX } from "@opal/icons"; +import { PageLoader, Section } from "@opal/layouts"; +import type { DateRange } from "@/components/dateRangeSelectors/AdminDateRangeSelector"; +import { formatCalendarDay } from "@/lib/dateUtils"; +import { useUsageExport } from "@/lib/usage/userUsage"; +import { formatCost, formatTokens } from "@/lib/utils"; +import SpendByUserTable from "@/sections/usage/SpendByUserTable"; +import UserUsageDetailModal from "@/sections/usage/UserUsageDetailModal"; + +function formatDate(value: string): string { + return formatCalendarDay(value, { withYear: true }); } -interface UsageRowProps { - user: UsageExportUser; - onReset: () => void; -} - -function UsageRow({ user, onReset }: UsageRowProps) { - const [resetting, setResetting] = useState(false); - const totals = user.totals; - - async function handleReset() { - setResetting(true); - try { - await resetUserUsage(user.email); - toast.success(`Reset usage for ${user.email}.`); - onReset(); - } catch (error) { - const message = error instanceof Error ? error.message : "unknown error"; - toast.error(`Failed to reset usage: ${message}`); - } finally { - setResetting(false); - } - } - +function SummaryMetric({ + label, + value, + detail, +}: { + label: string; + value: string; + detail: string; +}) { return ( -
-
- {user.email} -
-
- - {formatTokens(totals.input_tokens)} - -
-
- - {formatTokens(totals.output_tokens)} - -
-
- - {formatTokens(totals.cache_read_tokens)} +
+ + {label} + + + {value} + + + + {detail} -
-
- {formatCost(totals.cost_cents)} -
- +
); } -interface SortHeaderProps { - label: string; - sortKey: SortKey; - activeKey: SortKey; - dir: SortDir; - onSort: (key: SortKey) => void; - align: "left" | "right"; -} - -function SortHeader({ - label, - sortKey, - activeKey, - dir, - onSort, - align, -}: SortHeaderProps) { - const active = activeKey === sortKey; - const indicator = active ? (dir === "desc" ? " ↓" : " ↑") : ""; - return ( -
onSort(sortKey)} - onKeyDown={(e) => { - if (e.key === "Enter" || e.key === " ") { - e.preventDefault(); - onSort(sortKey); - } - }} - className={`cursor-pointer select-none ${ - align === "right" ? "w-24 text-right" : "flex-1" - }`} - > - - {`${label}${indicator}`} - -
- ); +interface PerUserUsagePanelProps { + timeRange?: DateRange; + headerRight?: React.ReactNode; } -/** Searchable, sortable admin per-user usage totals. */ -export default function PerUserUsagePanel() { - const { usage, isLoading, error, refetch } = useUsageExport(); - const [page, setPage] = useState(0); - const [query, setQuery] = useState(""); - const [sortKey, setSortKey] = useState("cost_cents"); - const [sortDir, setSortDir] = useState("desc"); +export default function PerUserUsagePanel({ + timeRange, + headerRight, +}: PerUserUsagePanelProps) { + const { usage, isLoading, error } = useUsageExport(timeRange); + const [selectedEmail, setSelectedEmail] = useState(null); const users = usage?.users ?? []; + const selectedUser = + users.find((user) => user.email === selectedEmail) ?? null; - const visible = useMemo(() => { - const q = query.trim().toLowerCase(); - const filtered = q - ? users.filter((u) => u.email.toLowerCase().includes(q)) - : users; - return [...filtered].sort((a, b) => { - const av = sortValue(a, sortKey); - const bv = sortValue(b, sortKey); - const cmp = - typeof av === "string" && typeof bv === "string" - ? av.localeCompare(bv) - : (av as number) - (bv as number); - return sortDir === "asc" ? cmp : -cmp; - }); - }, [users, query, sortKey, sortDir]); - - const pageCount = Math.max(1, Math.ceil(visible.length / PAGE_SIZE)); - - // Jump back to the first page whenever the filter or sort reshapes the list. - useEffect(() => { - setPage(0); - }, [query, sortKey, sortDir]); + const totalCostCents = useMemo( + () => users.reduce((total, user) => total + user.totals.cost_cents, 0), + [users] + ); + const totalTokens = useMemo( + () => + users.reduce( + (total, user) => + total + user.totals.input_tokens + user.totals.output_tokens, + 0 + ), + [users] + ); + const activeUsers = users.filter( + (user) => + user.totals.input_tokens > 0 || + user.totals.output_tokens > 0 || + user.totals.cache_read_tokens > 0 || + user.totals.cost_cents > 0 + ).length; + const topSpender = users.reduce<(typeof users)[number] | null>( + (top, user) => + user.totals.cost_cents > 0 && + (top === null || user.totals.cost_cents > top.totals.cost_cents) + ? user + : top, + null + ); - // Clamp the page when the list shrinks (e.g. a reset drops a user off). - useEffect(() => { - if (page > pageCount - 1) setPage(pageCount - 1); - }, [page, pageCount]); + const header = ( + // sm:flex-row / sm:items-center / sm:justify-between have no Section equivalent, kept as a raw div +
+
+ Usage this period + + {usage + ? `${formatDate(usage.start)} – ${formatDate(usage.end)} · Costs are calculated from recorded model usage.` + : "Per-user spend and token usage for the selected period."} + +
+ {headerRight} +
+ ); - function handleSort(key: SortKey) { - if (key === sortKey) { - setSortDir((d) => (d === "asc" ? "desc" : "asc")); - } else { - setSortKey(key); - // Numeric columns lead high→low (leaderboard); email reads A→Z. - setSortDir(key === "email" ? "asc" : "desc"); - } + if (isLoading) { + return ( +
+ {header} + +
+ ); } - - if (isLoading) return ; if (error) { return ( - +
+ {header} + +
); } - const pageUsers = visible.slice( - page * PAGE_SIZE, - page * PAGE_SIZE + PAGE_SIZE - ); - return ( - -
- Per-user usage - - Tokens (input, output, cache reads) and cost per user over the report - window. Click a column to rank by it, or search by email. Reset clears - usage from every currently active limit window. - - - setQuery(e.target.value)} - /> +
+ {header} + + +
+
+ +
+
+ +
+
+ +
+
+ +
+
+ + +
+
+ Spend by user + + Filter by model or flow, and click a user for their full breakdown. + +
{users.length === 0 ? ( - - No usage recorded yet. - - ) : visible.length === 0 ? ( - - {`No users match "${query}".`} - + + + No usage recorded for this period. + + ) : ( - <> -
-
-
- - - - - -
-
- {pageUsers.map((user) => ( - - ))} -
-
- - {pageCount > 1 && ( -
-
- )} - + )} -
- +
+ + {selectedUser && ( + { + if (!open) setSelectedEmail(null); + }} + /> + )} +
); } diff --git a/web/tests/e2e/admin/per_user_usage.spec.ts b/web/tests/e2e/admin/per_user_usage.spec.ts index cef7babeef3..bbcbd45e144 100644 --- a/web/tests/e2e/admin/per_user_usage.spec.ts +++ b/web/tests/e2e/admin/per_user_usage.spec.ts @@ -4,15 +4,14 @@ import { TEST_ADMIN_CREDENTIALS } from "@tests/e2e/constants"; import { AdminUsagePage } from "@tests/e2e/pages/AdminUsagePage"; /** - * Admin per-user usage table + Reset. Real e2e (no mocking): the admin sends a - * chat to accrue usage, then the Usage Statistics page must list that usage per - * user, and the Reset action must clear it. Requires a working LLM provider in - * the e2e environment so the chat records token usage. + * Admin per-user usage table. Real e2e (no mocking): the admin sends a chat to + * accrue usage, then the Usage page must list that usage per user. Requires a + * working LLM provider in the e2e environment so the chat records token usage. */ test.use({ storageState: "admin_auth.json" }); -test.describe("admin per-user usage table + reset", () => { - test("usage shows per user and Reset clears it", async ({ page }) => { +test.describe("admin per-user usage table", () => { + test("usage shows per user", async ({ page }) => { // 1) Accrue usage by sending a chat as the admin. const chat = new ChatPage(page); await chat.goto(); @@ -22,6 +21,6 @@ test.describe("admin per-user usage table + reset", () => { const usage = new AdminUsagePage(page); await usage.goto(); - await usage.resetUser(TEST_ADMIN_CREDENTIALS.email); + await usage.expectUser(TEST_ADMIN_CREDENTIALS.email); }); }); diff --git a/web/tests/e2e/pages/AdminUsagePage.ts b/web/tests/e2e/pages/AdminUsagePage.ts index e25f0d0d73c..d3d85c97f4a 100644 --- a/web/tests/e2e/pages/AdminUsagePage.ts +++ b/web/tests/e2e/pages/AdminUsagePage.ts @@ -1,14 +1,14 @@ import { expect, type Locator, type Page } from "@playwright/test"; -const RESET_ENDPOINT = "/api/admin/usage/reset"; - export class AdminUsagePage { readonly page: Page; readonly usageRows: Locator; + readonly userSearchInput: Locator; constructor(page: Page) { this.page = page; - this.usageRows = page.locator('[data-testid^="usage-row-"]'); + this.usageRows = page.locator('tr[aria-label^="View usage details for "]'); + this.userSearchInput = page.getByLabel("Search users by email"); } async goto(): Promise { @@ -16,19 +16,13 @@ export class AdminUsagePage { await expect(this.usageRows.first()).toBeVisible({ timeout: 15_000 }); } - async resetUser(email: string): Promise { - const row = this.page.getByTestId(`usage-row-${email}`); - await expect(row).toBeVisible(); - const responsePromise = this.page.waitForResponse( - (response) => - response.url().includes(RESET_ENDPOINT) && - response.request().method() === "POST" - ); - - await row.getByRole("button", { name: "Reset" }).click(); - expect((await responsePromise).ok()).toBeTruthy(); - await expect(this.page.getByText(`Reset usage for ${email}.`)).toBeVisible({ - timeout: 10_000, + async expectUser(email: string): Promise { + // The table only renders the first page (by spend) without filtering, so + // narrow to this user via search before asserting their row is visible. + await this.userSearchInput.fill(email); + const row = this.page.getByRole("row", { + name: `View usage details for ${email}`, }); + await expect(row).toBeVisible(); } }