diff --git a/CLAUDE.md b/CLAUDE.md index ddd4502d8f..b848eddd4e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -94,7 +94,7 @@ src/synthorg/ config/ # YAML company config loading and validation core/ # Shared domain models, base classes, resilience config engine/ # Orchestration, execution loops, task engine, coordination, checkpoint recovery, approval/review gates, stagnation detection, context budget, compaction, hybrid loop, workspace/ (git worktree isolation, merge orchestration, semantic conflict detection) - hr/ # Hiring, firing, onboarding, agent registry, performance tracking, activity timeline, career history, promotion/demotion + hr/ # Hiring, firing, onboarding, agent registry, performance tracking, activity timeline, activity event types, cost event redaction, career history, promotion/demotion memory/ # Pluggable MemoryBackend, retrieval pipeline, org memory, consolidation persistence/ # Pluggable PersistenceBackend, SQLite, settings + user repositories observability/ # Structured logging, correlation tracking, redaction, third-party logger taming, events/ diff --git a/docs/design/agents.md b/docs/design/agents.md index 67437138fe..0cc22a39a3 100644 --- a/docs/design/agents.md +++ b/docs/design/agents.md @@ -305,7 +305,7 @@ Performance data is exposed via three API sub-routes on `/api/v1/agents/{name}`: | Sub-route | Response model | Description | |-----------|---------------|-------------| | `GET /performance` | `AgentPerformanceSummary` | Flat summary: tasks completed (total/7d/30d), success rate, cost per task, quality/collaboration scores, trend direction, plus raw window metrics and trend results | -| `GET /activity` | `PaginatedResponse[ActivityEvent]` | Paginated chronological timeline merging lifecycle events, task metrics, cost records, tool invocations, and delegation records (most recent first) | +| `GET /activity` | `PaginatedResponse[ActivityEvent]` | Paginated chronological timeline merging lifecycle events, task metrics, cost records, tool invocations, and delegation records (most recent first). Supports typed `ActivityEventType` enum filtering (invalid values return 400). Cost events are redacted for read-only roles. Response includes `degraded_sources` field for partial data detection | | `GET /history` | `ApiResponse[tuple[CareerEvent, ...]]` | Career-relevant lifecycle events (hired, fired, promoted, demoted, onboarded) in chronological order | The framework tracks detailed per-agent metrics: diff --git a/docs/design/operations.md b/docs/design/operations.md index 5f2bdd198d..bce95d076b 100644 --- a/docs/design/operations.md +++ b/docs/design/operations.md @@ -1096,9 +1096,9 @@ future CLI tool are thin clients that call the API -- they contain no business l | `/api/v1/company` | CRUD company config | | `/api/v1/agents` | List, hire, fire, modify agents | | `GET /api/v1/agents/{name}/performance` | Agent performance metrics summary | -| `GET /api/v1/agents/{name}/activity` | Paginated agent activity timeline (lifecycle, task, cost, tool, delegation events) | +| `GET /api/v1/agents/{name}/activity` | Paginated agent activity timeline (lifecycle, task, cost, tool, delegation events); `degraded_sources` included in `PaginatedResponse` contract | | `GET /api/v1/agents/{name}/history` | Agent career history events | -| `GET /api/v1/activities` | Org-wide activity feed (merges all agents, filterable by type/agent/time window) | +| `GET /api/v1/activities` | Org-wide activity feed (merges all agents, enum-validated type filtering, cost event redaction for read-only roles, degraded source reporting) | | `/api/v1/departments` | Department management | | `/api/v1/projects` | Project CRUD | | `/api/v1/tasks` | Task management | diff --git a/src/synthorg/api/controllers/activities.py b/src/synthorg/api/controllers/activities.py index b849cd567e..4e16852a73 100644 --- a/src/synthorg/api/controllers/activities.py +++ b/src/synthorg/api/controllers/activities.py @@ -3,18 +3,21 @@ import asyncio from datetime import UTC, datetime, timedelta from enum import IntEnum -from typing import TYPE_CHECKING, Annotated +from typing import TYPE_CHECKING, Annotated, Any if TYPE_CHECKING: from collections.abc import Awaitable -from litestar import Controller, get + from synthorg.hr.models import AgentLifecycleEvent + +from litestar import Controller, Request, get from litestar.datastructures import State # noqa: TC002 from litestar.params import Parameter +from synthorg.api.auth.models import AuthenticatedUser from synthorg.api.dto import PaginatedResponse from synthorg.api.errors import ServiceUnavailableError -from synthorg.api.guards import require_read_access +from synthorg.api.guards import has_write_role, require_read_access from synthorg.api.pagination import PaginationLimit, PaginationOffset, paginate from synthorg.api.state import AppState # noqa: TC001 from synthorg.budget.cost_record import CostRecord # noqa: TC001 @@ -23,7 +26,9 @@ from synthorg.hr.activity import ( ActivityEvent, merge_activity_timeline, + redact_cost_events, ) +from synthorg.hr.enums import ActivityEventType # noqa: TC001 from synthorg.hr.performance.models import TaskMetricRecord # noqa: TC001 from synthorg.observability import get_logger from synthorg.observability.events.api import ( @@ -37,6 +42,13 @@ # Safety cap for unbounded lifecycle event queries. _MAX_LIFECYCLE_EVENTS = 10_000 +# Degraded source names -- used in responses and tests. +_SRC_PERFORMANCE_TRACKER = "performance_tracker" +_SRC_COST_TRACKER = "cost_tracker" +_SRC_TOOL_INVOCATION_TRACKER = "tool_invocation_tracker" +_SRC_DELEGATION_RECORD_STORE = "delegation_record_store" +_SRC_BUDGET_CONFIG = "budget_config" + class ActivityWindowHours(IntEnum): """Allowed time windows for the activity feed.""" @@ -46,6 +58,223 @@ class ActivityWindowHours(IntEnum): WEEK = 168 +def _extract_task_result( + task: asyncio.Task[tuple[tuple[Any, ...], bool]] | None, + source_name: str, + degraded: list[str], +) -> tuple[Any, ...]: + """Extract a completed task's data, appending to degraded if needed.""" + if task is None or task.cancelled(): + degraded.append(source_name) + return () + if task.exception() is not None: + degraded.append(source_name) + return () + data, is_degraded = task.result() + if is_degraded: + degraded.append(source_name) + return data + + +async def _run_async_fetchers( + app_state: AppState, + agent_id: str | None, + since: datetime, + now: datetime, + degraded: list[str], +) -> tuple[ + tuple[CostRecord, ...], + tuple[ToolInvocationRecord, ...], + tuple[DelegationRecord, ...], + tuple[DelegationRecord, ...], +]: + """Run cost, tool, and delegation fetchers concurrently. + + Completed tasks have their results extracted; failed or cancelled + tasks are individually marked as degraded rather than blanket-marking + all sources. + + Args: + app_state: Application state with service references. + agent_id: Optional agent filter. + since: Start of the time window. + now: End of the time window. + degraded: Mutable list to append degraded source names to. + + Returns: + ``(cost_records, tool_invocations, sent, received)`` tuples. + """ + cost_task: asyncio.Task[tuple[tuple[CostRecord, ...], bool]] | None = None + tool_task: asyncio.Task[tuple[tuple[ToolInvocationRecord, ...], bool]] | None = None + del_task: asyncio.Task[tuple[Any, ...]] | None = None + try: + async with asyncio.TaskGroup() as tg: + cost_task = tg.create_task( + _fetch_cost_records(app_state, agent_id, since, now), + ) + tool_task = tg.create_task( + _fetch_tool_invocations(app_state, agent_id, since, now), + ) + del_task = tg.create_task( + _fetch_delegation_records(app_state, agent_id, since, now), + ) + except ExceptionGroup as eg: + fatal = eg.subgroup((MemoryError, RecursionError)) + if fatal is not None: + logger.error( + API_REQUEST_ERROR, + endpoint="activities", + detail="fatal error in async fetchers", + exc_info=True, + ) + raise fatal.exceptions[0] from eg + svc = eg.subgroup(ServiceUnavailableError) + if svc is not None: + logger.warning( + API_REQUEST_ERROR, + endpoint="activities", + detail="service unavailable in async fetchers", + exc_info=True, + ) + raise svc.exceptions[0] from eg + failed_sources = [ + src + for src, task in [ + (_SRC_COST_TRACKER, cost_task), + (_SRC_TOOL_INVOCATION_TRACKER, tool_task), + (_SRC_DELEGATION_RECORD_STORE, del_task), + ] + if task is None or task.cancelled() or task.exception() is not None + ] + logger.warning( + API_REQUEST_ERROR, + endpoint="activities", + error_count=len(eg.exceptions), + failed_sources=failed_sources, + exc_info=True, + ) + + cost_records: tuple[CostRecord, ...] = _extract_task_result( + cost_task, + _SRC_COST_TRACKER, + degraded, + ) + tool_invocations: tuple[ToolInvocationRecord, ...] = _extract_task_result( + tool_task, + _SRC_TOOL_INVOCATION_TRACKER, + degraded, + ) + + if ( + del_task is not None + and not del_task.cancelled() + and del_task.exception() is None + ): + del_result = del_task.result() + sent, received, del_deg = del_result[0], del_result[1], del_result[2] + if del_deg: + degraded.append(_SRC_DELEGATION_RECORD_STORE) + else: + if del_task is not None: + degraded.append(_SRC_DELEGATION_RECORD_STORE) + sent, received = (), () + + return cost_records, tool_invocations, sent, received + + +async def _resolve_currency( + app_state: AppState, + degraded: list[str], +) -> str: + """Resolve the display currency from budget config. + + Falls back to ``DEFAULT_CURRENCY`` on any transient error and + appends the source name to ``degraded``. + + Args: + app_state: Application state with config resolver. + degraded: Mutable list to append degraded source names to. + + Returns: + ISO 4217 currency code. + """ + try: + budget_cfg = await app_state.config_resolver.get_budget_config() + except MemoryError, RecursionError: + logger.error( + API_REQUEST_ERROR, + endpoint="activities", + source=_SRC_BUDGET_CONFIG, + detail="fatal error", + exc_info=True, + ) + raise + except Exception: + logger.warning( + API_REQUEST_ERROR, + endpoint="activities", + detail="budget config unavailable, using default currency", + exc_info=True, + ) + degraded.append(_SRC_BUDGET_CONFIG) + return DEFAULT_CURRENCY + else: + return budget_cfg.currency + + +async def _build_timeline( + app_state: AppState, + lifecycle_events: tuple[AgentLifecycleEvent, ...], + agent_id: str | None, + since: datetime, + now: datetime, +) -> tuple[tuple[ActivityEvent, ...], list[str]]: + """Fetch non-lifecycle data sources, merge with lifecycle events. + + Args: + app_state: Application state with service references. + lifecycle_events: Pre-fetched lifecycle events. + agent_id: Optional agent filter. + since: Start of the time window. + now: End of the time window (current time). + + Returns: + ``(timeline, degraded_sources)`` where ``degraded_sources`` + lists the names of data sources that failed. + """ + degraded: list[str] = [] + + task_metrics, tm_degraded = await _fetch_task_metrics( + app_state, + agent_id, + since, + now, + ) + if tm_degraded: + degraded.append(_SRC_PERFORMANCE_TRACKER) + + cost_records, tool_invocations, sent, received = await _run_async_fetchers( + app_state, + agent_id, + since, + now, + degraded, + ) + + currency = await _resolve_currency(app_state, degraded) + + timeline = merge_activity_timeline( + lifecycle_events=lifecycle_events, + task_metrics=task_metrics, + cost_records=cost_records, + tool_invocations=tool_invocations, + delegation_records_sent=sent, + delegation_records_received=received, + currency=currency, + ) + return timeline, list(dict.fromkeys(degraded)) + + class ActivityController(Controller): """Org-wide activity feed (REST fallback for WebSocket).""" @@ -56,14 +285,14 @@ class ActivityController(Controller): @get() async def list_activities( # noqa: PLR0913 self, + request: Request[Any, Any, Any], state: State, offset: PaginationOffset = 0, limit: PaginationLimit = 50, event_type: Annotated[ - str | None, + ActivityEventType | None, Parameter( query="type", - max_length=64, description="Filter by event_type", ), ] = None, @@ -87,15 +316,18 @@ async def list_activities( # noqa: PLR0913 data sources degrade gracefully when unavailable. Args: + request: Incoming HTTP request (used for role-based redaction). state: Application state. offset: Pagination offset. limit: Page size. - event_type: Filter by event_type (e.g. ``"hired"``). + event_type: Filter by ``ActivityEventType`` (e.g. ``"hired"``). + Invalid values are rejected with 400. agent_id: Filter events for a specific agent. last_n_hours: Time window in hours (24, 48, or 168). Returns: - Paginated activity events. + Paginated activity events. The ``degraded_sources`` field + lists any data sources that failed gracefully. """ app_state: AppState = state.app_state now = datetime.now(UTC) @@ -107,63 +339,26 @@ async def list_activities( # noqa: PLR0913 limit=_MAX_LIFECYCLE_EVENTS, ) - task_metrics = _fetch_task_metrics(app_state, agent_id, since, now) - try: - async with asyncio.TaskGroup() as tg: - cost_task = tg.create_task( - _fetch_cost_records(app_state, agent_id, since, now), - ) - tool_task = tg.create_task( - _fetch_tool_invocations(app_state, agent_id, since, now), - ) - del_task = tg.create_task( - _fetch_delegation_records(app_state, agent_id, since, now), - ) - except ExceptionGroup as eg: - fatal = eg.subgroup((MemoryError, RecursionError)) - if fatal is not None: - raise fatal from eg - svc = eg.subgroup(ServiceUnavailableError) - if svc is not None: - raise svc from eg - logger.warning( - API_REQUEST_ERROR, - endpoint="activities", - error_count=len(eg.exceptions), - exc_info=True, - ) - cost_task = tool_task = del_task = None # type: ignore[assignment] - cost_records = cost_task.result() if cost_task is not None else () - tool_invocations = tool_task.result() if tool_task is not None else () - if del_task is not None: - sent, received = del_task.result() - else: - sent, received = (), () - - try: - budget_cfg = await app_state.config_resolver.get_budget_config() - currency = budget_cfg.currency - except Exception: - logger.warning( - API_REQUEST_ERROR, - endpoint="activities", - detail="budget config unavailable, using default currency", - exc_info=True, - ) - currency = DEFAULT_CURRENCY - timeline = merge_activity_timeline( - lifecycle_events=lifecycle_events, - task_metrics=task_metrics, - cost_records=cost_records, - tool_invocations=tool_invocations, - delegation_records_sent=sent, - delegation_records_received=received, - currency=currency, + timeline, degraded = await _build_timeline( + app_state, + lifecycle_events, + agent_id, + since, + now, ) if event_type is not None: timeline = tuple(e for e in timeline if e.event_type == event_type) + # Redact cost details unless the user has a write role. + # Fail-closed: redact by default if auth identity is missing + # (e.g. misconfigured excluded path, test stub without scope["user"]). + auth_user = request.scope.get("user") + if not ( + isinstance(auth_user, AuthenticatedUser) and has_write_role(auth_user.role) + ): + timeline = redact_cost_events(timeline) + page, meta = paginate(timeline, offset=offset, limit=limit) logger.debug( @@ -174,28 +369,53 @@ async def list_activities( # noqa: PLR0913 last_n_hours=last_n_hours, ) - return PaginatedResponse(data=page, pagination=meta) + return PaginatedResponse( + data=page, + pagination=meta, + degraded_sources=tuple(degraded), + ) # ── Data source fetchers (graceful degradation) ────────────────── -def _fetch_task_metrics( +async def _fetch_task_metrics( app_state: AppState, agent_id: str | None, since: datetime, now: datetime, -) -> tuple[TaskMetricRecord, ...]: - """Fetch task metrics, falling back to empty on failure.""" +) -> tuple[tuple[TaskMetricRecord, ...], bool]: + """Fetch task metrics, falling back to empty on failure. + + The underlying ``PerformanceTracker`` call is synchronous (in-memory), + but the wrapper is async for consistency with the other fetchers. + + Returns: + ``(records, is_degraded)`` tuple. + """ try: return app_state.performance_tracker.get_task_metrics( agent_id=agent_id, since=since, until=now, - ) + ), False except MemoryError, RecursionError: + logger.error( + API_REQUEST_ERROR, + endpoint="activities", + source=_SRC_PERFORMANCE_TRACKER, + detail="fatal error", + exc_info=True, + ) raise except ServiceUnavailableError: + logger.warning( + API_REQUEST_ERROR, + endpoint="activities", + source=_SRC_PERFORMANCE_TRACKER, + detail="service unavailable", + exc_info=True, + ) raise except Exception: logger.warning( @@ -204,7 +424,7 @@ def _fetch_task_metrics( error="performance_tracker_unavailable", exc_info=True, ) - return () + return (), True async def _fetch_cost_records( @@ -212,19 +432,37 @@ async def _fetch_cost_records( agent_id: str | None, since: datetime, now: datetime, -) -> tuple[CostRecord, ...]: - """Fetch cost records, falling back to empty on failure.""" +) -> tuple[tuple[CostRecord, ...], bool]: + """Fetch cost records, falling back to empty on failure. + + Returns: + ``(records, is_degraded)`` tuple. + """ if not app_state.has_cost_tracker: - return () + return (), False try: return await app_state.cost_tracker.get_records( agent_id=agent_id, start=since, end=now, - ) + ), False except MemoryError, RecursionError: + logger.error( + API_REQUEST_ERROR, + endpoint="activities", + source=_SRC_COST_TRACKER, + detail="fatal error", + exc_info=True, + ) raise except ServiceUnavailableError: + logger.warning( + API_REQUEST_ERROR, + endpoint="activities", + source=_SRC_COST_TRACKER, + detail="service unavailable", + exc_info=True, + ) raise except Exception: logger.warning( @@ -233,7 +471,7 @@ async def _fetch_cost_records( error="cost_tracker_unavailable", exc_info=True, ) - return () + return (), True async def _fetch_tool_invocations( @@ -241,19 +479,37 @@ async def _fetch_tool_invocations( agent_id: str | None, since: datetime, now: datetime, -) -> tuple[ToolInvocationRecord, ...]: - """Fetch tool invocation records, falling back to empty on failure.""" +) -> tuple[tuple[ToolInvocationRecord, ...], bool]: + """Fetch tool invocation records, falling back to empty on failure. + + Returns: + ``(records, is_degraded)`` tuple. + """ if not app_state.has_tool_invocation_tracker: - return () + return (), False try: return await app_state.tool_invocation_tracker.get_records( agent_id=agent_id, start=since, end=now, - ) + ), False except MemoryError, RecursionError: + logger.error( + API_REQUEST_ERROR, + endpoint="activities", + source=_SRC_TOOL_INVOCATION_TRACKER, + detail="fatal error", + exc_info=True, + ) raise except ServiceUnavailableError: + logger.warning( + API_REQUEST_ERROR, + endpoint="activities", + source=_SRC_TOOL_INVOCATION_TRACKER, + detail="service unavailable", + exc_info=True, + ) raise except Exception: logger.warning( @@ -262,19 +518,37 @@ async def _fetch_tool_invocations( error="tool_invocation_tracker_unavailable", exc_info=True, ) - return () + return (), True async def _safe_delegation_query( coro: Awaitable[tuple[DelegationRecord, ...]], error_label: str, -) -> tuple[DelegationRecord, ...]: - """Run a delegation store query with graceful degradation.""" +) -> tuple[tuple[DelegationRecord, ...], bool]: + """Run a delegation store query with graceful degradation. + + Returns: + ``(records, is_degraded)`` tuple. + """ try: - return await coro + return (await coro), False except MemoryError, RecursionError: + logger.error( + API_REQUEST_ERROR, + endpoint="activities", + source=error_label, + detail="fatal error", + exc_info=True, + ) raise except ServiceUnavailableError: + logger.warning( + API_REQUEST_ERROR, + endpoint="activities", + source=error_label, + detail="service unavailable", + exc_info=True, + ) raise except Exception: logger.warning( @@ -283,7 +557,7 @@ async def _safe_delegation_query( error=error_label, exc_info=True, ) - return () + return (), True async def _fetch_delegation_records( @@ -291,31 +565,42 @@ async def _fetch_delegation_records( agent_id: str | None, since: datetime, now: datetime, -) -> tuple[tuple[DelegationRecord, ...], tuple[DelegationRecord, ...]]: +) -> tuple[ + tuple[DelegationRecord, ...], + tuple[DelegationRecord, ...], + bool, +]: """Fetch delegation records (sent + received), falling back to empty. Returns: - ``(sent, received)`` tuples of delegation records. + ``(sent, received, is_degraded)`` tuple. """ if not app_state.has_delegation_record_store: - return (), () + return (), (), False store = app_state.delegation_record_store if agent_id is None: # Org-wide: each record generates both perspectives. - all_records = await _safe_delegation_query( + all_records, degraded = await _safe_delegation_query( store.get_all_records(start=since, end=now), "delegation_record_store_unavailable", ) - return all_records, all_records + return all_records, all_records, degraded - # Agent-specific: fetch each perspective independently so a + # Agent-specific: fetch each perspective concurrently so a # failure in one does not discard the other. - sent = await _safe_delegation_query( - store.get_records_as_delegator(agent_id, start=since, end=now), - "delegation_delegator_query_failed", - ) - received = await _safe_delegation_query( - store.get_records_as_delegatee(agent_id, start=since, end=now), - "delegation_delegatee_query_failed", - ) - return sent, received + async with asyncio.TaskGroup() as tg: + sent_task = tg.create_task( + _safe_delegation_query( + store.get_records_as_delegator(agent_id, start=since, end=now), + "delegation_delegator_query_failed", + ), + ) + recv_task = tg.create_task( + _safe_delegation_query( + store.get_records_as_delegatee(agent_id, start=since, end=now), + "delegation_delegatee_query_failed", + ), + ) + sent, sent_deg = sent_task.result() + received, recv_deg = recv_task.result() + return sent, received, sent_deg or recv_deg diff --git a/src/synthorg/api/dto.py b/src/synthorg/api/dto.py index 2a859e2fee..6b8d05f440 100644 --- a/src/synthorg/api/dto.py +++ b/src/synthorg/api/dto.py @@ -5,9 +5,7 @@ models because they omit server-generated fields). """ -import re from typing import Self -from urllib.parse import urlparse from pydantic import ( BaseModel, @@ -20,7 +18,6 @@ from synthorg.api.errors import ErrorCategory, ErrorCode # noqa: TC001 from synthorg.budget.currency import DEFAULT_CURRENCY -from synthorg.config.schema import ProviderConfig, ProviderModelConfig # noqa: TC001 from synthorg.core.enums import ( ApprovalRiskLevel, Complexity, @@ -30,7 +27,6 @@ ) from synthorg.core.types import NotBlankStr # noqa: TC001 from synthorg.core.validation import is_valid_action_type -from synthorg.providers.enums import AuthType DEFAULT_LIMIT: int = 50 MAX_LIMIT: int = 200 @@ -71,7 +67,7 @@ class ErrorDetail(BaseModel): type: Documentation URI for the error category. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) detail: NotBlankStr error_code: ErrorCode @@ -107,7 +103,7 @@ class ProblemDetail(BaseModel): when not applicable). """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) type: NotBlankStr title: NotBlankStr @@ -139,7 +135,7 @@ class ApiResponse[T](BaseModel): success: Whether the request succeeded (computed from ``error``). """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) data: T | None = None error: str | None = None @@ -169,7 +165,7 @@ class PaginationMeta(BaseModel): limit: Maximum items per page. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) total: int = Field(ge=0, description="Total matching items") offset: int = Field(ge=0, description="Starting offset") @@ -184,15 +180,21 @@ class PaginatedResponse[T](BaseModel): error: Error message (``None`` on success). error_detail: Structured error metadata (``None`` on success). pagination: Pagination metadata. + degraded_sources: Data sources that failed gracefully, resulting + in partial data. Empty when all sources responded normally. success: Whether the request succeeded (computed from ``error``). """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) data: tuple[T, ...] = () error: str | None = None error_detail: ErrorDetail | None = None pagination: PaginationMeta + degraded_sources: tuple[NotBlankStr, ...] = Field( + default=(), + description="Data sources that failed gracefully (partial data)", + ) @model_validator(mode="after") def _validate_error_detail_consistency(self) -> Self: @@ -227,10 +229,10 @@ class CreateTaskRequest(BaseModel): created_by: Agent name of the creator. assigned_to: Optional assignee agent ID. estimated_complexity: Complexity estimate. - budget_limit: Maximum spend in USD (base currency). + budget_limit: Maximum spend in base currency. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) title: NotBlankStr = Field(max_length=256) description: NotBlankStr = Field(max_length=4096) @@ -257,7 +259,7 @@ class UpdateTaskRequest(BaseModel): expected_version: Optimistic concurrency guard. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) title: NotBlankStr | None = Field(default=None, max_length=256) description: NotBlankStr | None = Field(default=None, max_length=4096) @@ -280,7 +282,7 @@ class TransitionTaskRequest(BaseModel): expected_version: Optimistic concurrency guard. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) target_status: TaskStatus = Field(description="Desired target status") assigned_to: NotBlankStr | None = None @@ -298,7 +300,7 @@ class CancelTaskRequest(BaseModel): reason: Reason for cancellation. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) reason: NotBlankStr = Field( max_length=4096, @@ -324,7 +326,7 @@ class CreateApprovalRequest(BaseModel): metadata: Additional key-value pairs. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) action_type: NotBlankStr = Field(max_length=128) title: NotBlankStr = Field(max_length=256) @@ -367,7 +369,7 @@ class ApproveRequest(BaseModel): comment: Optional comment explaining the approval. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) comment: NotBlankStr | None = Field(default=None, max_length=4096) @@ -379,7 +381,7 @@ class RejectRequest(BaseModel): reason: Mandatory reason for rejection. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) reason: NotBlankStr = Field(max_length=4096) @@ -399,7 +401,7 @@ class CoordinateTaskRequest(BaseModel): section config default). """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) agent_names: tuple[NotBlankStr, ...] | None = Field( default=None, @@ -439,7 +441,7 @@ class CoordinationPhaseResponse(BaseModel): error: Error description if the phase failed. """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) phase: NotBlankStr success: bool @@ -471,7 +473,7 @@ class CoordinationResultResponse(BaseModel): is_success: Whether all phases succeeded (computed). """ - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, allow_inf_nan=False) parent_task_id: NotBlankStr = Field(max_length=128) topology: NotBlankStr @@ -496,334 +498,45 @@ def is_success(self) -> bool: return all(p.success for p in self.phases) -# ── Provider management DTOs ──────────────────────────────── - -_PROVIDER_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}[a-z0-9]$") -_RESERVED_PROVIDER_NAMES: frozenset[str] = frozenset( - {"presets", "from-preset", "probe-preset", "discovery-policy"}, +# ── Provider management DTOs (split to dto_providers.py) ──── +# Re-exported for backwards compatibility. +from synthorg.api.dto_providers import ( # noqa: E402 + CreateFromPresetRequest, + CreateProviderRequest, + DiscoverModelsResponse, + ProbePresetRequest, + ProbePresetResponse, + ProviderResponse, + TestConnectionRequest, + TestConnectionResponse, + UpdateProviderRequest, + to_provider_response, ) - -def _validate_provider_name(v: str) -> str: - """Validate a provider name against naming rules. - - Args: - v: Candidate provider name. - - Returns: - The validated name. - - Raises: - ValueError: If the name is invalid or reserved. - """ - if not _PROVIDER_NAME_PATTERN.match(v): - msg = ( - "Provider name must be 2-64 chars, lowercase " - "alphanumeric and hyphens, starting/ending with " - "alphanumeric" - ) - raise ValueError(msg) - if v in _RESERVED_PROVIDER_NAMES: - msg = f"Provider name {v!r} is reserved" - raise ValueError(msg) - return v - - -def _validate_base_url(v: str | None) -> str | None: - """Validate that a base URL uses http or https scheme.""" - if v is None: - return v - parsed = urlparse(v) - if parsed.scheme not in ("http", "https"): - msg = f"base_url must use http or https scheme, got {parsed.scheme!r}" - raise ValueError(msg) - if not parsed.netloc: - msg = "base_url must include a host" - raise ValueError(msg) - return v - - -class CreateProviderRequest(BaseModel): - """Payload for creating a new provider. - - Attributes: - name: Unique provider name (2-64 chars, lowercase alphanumeric + hyphens). - driver: Driver backend name (default ``"litellm"``). - litellm_provider: LiteLLM routing identifier override. - auth_type: Authentication mechanism for this provider. - api_key: API key credential (optional, depends on auth_type). - subscription_token: Bearer token for subscription-based auth. - tos_accepted: Whether the user accepted the subscription ToS warning. - base_url: Provider API base URL. - models: Pre-configured model definitions. - """ - - model_config = ConfigDict(frozen=True) - - name: NotBlankStr = Field(max_length=64) - driver: NotBlankStr = "litellm" - litellm_provider: NotBlankStr | None = None - auth_type: AuthType = AuthType.API_KEY - api_key: NotBlankStr | None = None - subscription_token: NotBlankStr | None = None - tos_accepted: bool = False - base_url: NotBlankStr | None = None - oauth_token_url: NotBlankStr | None = None - oauth_client_id: NotBlankStr | None = None - oauth_client_secret: NotBlankStr | None = None - oauth_scope: NotBlankStr | None = None - custom_header_name: NotBlankStr | None = None - custom_header_value: NotBlankStr | None = None - models: tuple[ProviderModelConfig, ...] = () - - @field_validator("name") - @classmethod - def _validate_name(cls, v: str) -> str: - return _validate_provider_name(v) - - @field_validator("base_url") - @classmethod - def _validate_base_url(cls, v: str | None) -> str | None: - return _validate_base_url(v) - - -class UpdateProviderRequest(BaseModel): - """Payload for updating a provider (partial update). - - All fields are optional -- only provided fields are updated. - ``tos_accepted``: only ``True`` re-stamps the timestamp; - ``False`` and ``None`` are no-ops (acceptance cannot be retracted). - """ - - model_config = ConfigDict(frozen=True) - - driver: NotBlankStr | None = None - litellm_provider: NotBlankStr | None = None - auth_type: AuthType | None = None - api_key: NotBlankStr | None = None - clear_api_key: bool = False - subscription_token: NotBlankStr | None = None - clear_subscription_token: bool = False - tos_accepted: bool | None = None - base_url: NotBlankStr | None = None - oauth_token_url: NotBlankStr | None = None - oauth_client_id: NotBlankStr | None = None - oauth_client_secret: NotBlankStr | None = None - oauth_scope: NotBlankStr | None = None - custom_header_name: NotBlankStr | None = None - custom_header_value: NotBlankStr | None = None - models: tuple[ProviderModelConfig, ...] | None = None - - @field_validator("base_url") - @classmethod - def _validate_base_url(cls, v: str | None) -> str | None: - return _validate_base_url(v) - - @model_validator(mode="after") - def _validate_credential_clear_consistency(self) -> Self: - """Reject simultaneous set and clear for credential fields.""" - if self.api_key is not None and self.clear_api_key: - msg = "api_key and clear_api_key are mutually exclusive" - raise ValueError(msg) - if self.subscription_token is not None and self.clear_subscription_token: - msg = ( - "subscription_token and clear_subscription_token are mutually exclusive" - ) - raise ValueError(msg) - return self - - -class TestConnectionRequest(BaseModel): - """Payload for testing a provider connection. - - Attributes: - model: Model to test (defaults to first model in config). - """ - - model_config = ConfigDict(frozen=True) - - model: NotBlankStr | None = None - - -class TestConnectionResponse(BaseModel): - """Result of a provider connection test. - - Attributes: - success: Whether the connection test succeeded. - latency_ms: Round-trip latency in milliseconds. - error: Error message on failure. - model_tested: Model ID that was tested. - """ - - model_config = ConfigDict(frozen=True) - - success: bool - latency_ms: float | None = None - error: NotBlankStr | None = None - model_tested: NotBlankStr | None = None - - @model_validator(mode="after") - def _validate_success_error_consistency(self) -> Self: - """Ensure success and error fields are consistent.""" - if self.success and self.error is not None: - msg = "successful test must not have an error" - raise ValueError(msg) - if not self.success and self.error is None: - msg = "failed test must have an error message" - raise ValueError(msg) - return self - - -class ProviderResponse(BaseModel): - """Safe provider config for API responses -- secrets stripped. - - Non-secret auth fields are included for frontend edit form UX. - Boolean ``has_*`` indicators signal credential presence without - exposing values. - - Attributes: - driver: Driver backend name. - litellm_provider: LiteLLM routing identifier override. - auth_type: Authentication mechanism. - base_url: Provider API base URL. - models: Configured model definitions. - has_api_key: Whether an API key is set. - has_oauth_credentials: Whether OAuth credentials are configured. - has_custom_header: Whether a custom auth header is configured. - has_subscription_token: Whether a subscription token is set. - tos_accepted_at: ISO timestamp of ToS acceptance (or ``None``). - """ - - model_config = ConfigDict(frozen=True) - - driver: NotBlankStr - litellm_provider: NotBlankStr | None = None - auth_type: AuthType - base_url: NotBlankStr | None - models: tuple[ProviderModelConfig, ...] - has_api_key: bool - has_oauth_credentials: bool - has_custom_header: bool - has_subscription_token: bool = False - tos_accepted_at: str | None = None - oauth_token_url: NotBlankStr | None = None - oauth_client_id: NotBlankStr | None = None - oauth_scope: NotBlankStr | None = None - custom_header_name: NotBlankStr | None = None - - -class CreateFromPresetRequest(BaseModel): - """Payload for creating a provider from a preset. - - Attributes: - preset_name: Name of the preset to create from. - name: Unique provider name (2-64 chars, lowercase alphanumeric + hyphens). - auth_type: Override the preset's default auth type (optional). - subscription_token: Bearer token for subscription-based auth. - tos_accepted: Whether the user accepted the subscription ToS warning. - base_url: Override the preset's default base URL (optional). - """ - - model_config = ConfigDict(frozen=True) - - preset_name: NotBlankStr - name: NotBlankStr = Field(max_length=64) - auth_type: AuthType | None = None - api_key: NotBlankStr | None = None - subscription_token: NotBlankStr | None = None - tos_accepted: bool = False - base_url: NotBlankStr | None = None - models: tuple[ProviderModelConfig, ...] | None = None - - @field_validator("name") - @classmethod - def _validate_name(cls, v: str) -> str: - return _validate_provider_name(v) - - @field_validator("base_url") - @classmethod - def _validate_base_url(cls, v: str | None) -> str | None: - return _validate_base_url(v) - - -class DiscoverModelsResponse(BaseModel): - """Result of provider model auto-discovery. - - Attributes: - discovered_models: Models found on the provider endpoint. - provider_name: Name of the provider that was queried. - """ - - model_config = ConfigDict(frozen=True) - - discovered_models: tuple[ProviderModelConfig, ...] - provider_name: NotBlankStr - - -class ProbePresetRequest(BaseModel): - """Request to probe a preset's candidate URLs for reachability. - - Attributes: - preset_name: Preset identifier to probe. - """ - - model_config = ConfigDict(frozen=True) - - preset_name: NotBlankStr = Field(max_length=64) - - -class ProbePresetResponse(BaseModel): - """Result of probing a preset's candidate URLs. - - Attributes: - url: The first reachable base URL, or ``None`` if none responded. - model_count: Number of models discovered at the URL. - candidates_tried: Number of candidate URLs attempted. - """ - - model_config = ConfigDict(frozen=True) - - url: NotBlankStr | None = None - model_count: int = Field(default=0, ge=0) - candidates_tried: int = Field(default=0, ge=0) - - -def to_provider_response(config: ProviderConfig) -> ProviderResponse: - """Convert a ProviderConfig to a safe ProviderResponse. - - Strips all secrets and provides boolean credential indicators. - - Args: - config: Provider configuration (may contain secrets). - - Returns: - Safe response DTO with secrets stripped. - """ - tos_str = ( - config.tos_accepted_at.isoformat() - if config.tos_accepted_at is not None - else None - ) - return ProviderResponse( - driver=config.driver, - litellm_provider=config.litellm_provider, - auth_type=config.auth_type, - base_url=config.base_url, - models=config.models, - has_api_key=config.api_key is not None, - has_oauth_credentials=( - config.oauth_client_id is not None - and config.oauth_client_secret is not None - and config.oauth_token_url is not None - ), - has_custom_header=( - config.custom_header_name is not None - and config.custom_header_value is not None - ), - has_subscription_token=config.subscription_token is not None, - tos_accepted_at=tos_str, - oauth_token_url=config.oauth_token_url, - oauth_client_id=config.oauth_client_id, - oauth_scope=config.oauth_scope, - custom_header_name=config.custom_header_name, - ) +__all__ = [ + "ApiResponse", + "ApproveRequest", + "CancelTaskRequest", + "CoordinateTaskRequest", + "CoordinationPhaseResponse", + "CoordinationResultResponse", + "CreateApprovalRequest", + "CreateFromPresetRequest", + "CreateProviderRequest", + "CreateTaskRequest", + "DiscoverModelsResponse", + "ErrorDetail", + "PaginatedResponse", + "PaginationMeta", + "ProbePresetRequest", + "ProbePresetResponse", + "ProblemDetail", + "ProviderResponse", + "RejectRequest", + "TestConnectionRequest", + "TestConnectionResponse", + "TransitionTaskRequest", + "UpdateProviderRequest", + "UpdateTaskRequest", + "to_provider_response", +] diff --git a/src/synthorg/api/dto_providers.py b/src/synthorg/api/dto_providers.py index 07fda1691e..40aca615ed 100644 --- a/src/synthorg/api/dto_providers.py +++ b/src/synthorg/api/dto_providers.py @@ -1,13 +1,21 @@ -"""Provider-specific response DTOs. +"""Provider-specific request/response DTOs. Split from ``dto.py`` to keep that file under the 800-line limit. """ -from pydantic import BaseModel, ConfigDict, Field +import re +from typing import TYPE_CHECKING, Self +from urllib.parse import urlparse + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from synthorg.config.schema import ProviderModelConfig # noqa: TC001 from synthorg.core.types import NotBlankStr # noqa: TC001 from synthorg.providers.capabilities import ModelCapabilities # noqa: TC001 +from synthorg.providers.enums import AuthType + +if TYPE_CHECKING: + from synthorg.config.schema import ProviderConfig class ProviderModelResponse(BaseModel): @@ -67,6 +75,342 @@ class ProviderModelResponse(BaseModel): ) +# ── Provider management DTOs ──────────────────────────────── + +_PROVIDER_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}[a-z0-9]$") +_RESERVED_PROVIDER_NAMES: frozenset[str] = frozenset( + {"presets", "from-preset", "probe-preset", "discovery-policy"}, +) + + +def _validate_provider_name(v: str) -> str: + """Validate a provider name against naming rules. + + Args: + v: Candidate provider name. + + Returns: + The validated name. + + Raises: + ValueError: If the name is invalid or reserved. + """ + if not _PROVIDER_NAME_PATTERN.match(v): + msg = ( + "Provider name must be 2-64 chars, lowercase " + "alphanumeric and hyphens, starting/ending with " + "alphanumeric" + ) + raise ValueError(msg) + if v in _RESERVED_PROVIDER_NAMES: + msg = f"Provider name {v!r} is reserved" + raise ValueError(msg) + return v + + +def _validate_base_url(v: str | None) -> str | None: + """Validate that a base URL uses http or https scheme.""" + if v is None: + return v + parsed = urlparse(v) + if parsed.scheme not in ("http", "https"): + msg = f"base_url must use http or https scheme, got {parsed.scheme!r}" + raise ValueError(msg) + if not parsed.netloc: + msg = "base_url must include a host" + raise ValueError(msg) + return v + + +class CreateProviderRequest(BaseModel): + """Payload for creating a new provider. + + Attributes: + name: Unique provider name (2-64 chars, lowercase + hyphens). + driver: Driver backend name (default ``"litellm"``). + litellm_provider: LiteLLM routing identifier override. + auth_type: Authentication mechanism for this provider. + api_key: API key credential (optional, depends on auth_type). + subscription_token: Bearer token for subscription-based auth. + tos_accepted: Whether the user accepted the subscription ToS. + base_url: Provider API base URL. + models: Pre-configured model definitions. + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + name: NotBlankStr = Field(max_length=64) + driver: NotBlankStr = "litellm" + litellm_provider: NotBlankStr | None = None + auth_type: AuthType = AuthType.API_KEY + api_key: NotBlankStr | None = None + subscription_token: NotBlankStr | None = None + tos_accepted: bool = False + base_url: NotBlankStr | None = None + oauth_token_url: NotBlankStr | None = None + oauth_client_id: NotBlankStr | None = None + oauth_client_secret: NotBlankStr | None = None + oauth_scope: NotBlankStr | None = None + custom_header_name: NotBlankStr | None = None + custom_header_value: NotBlankStr | None = None + models: tuple[ProviderModelConfig, ...] = () + + @field_validator("name") + @classmethod + def _validate_name(cls, v: str) -> str: + return _validate_provider_name(v) + + @field_validator("base_url") + @classmethod + def _validate_base_url(cls, v: str | None) -> str | None: + return _validate_base_url(v) + + +class UpdateProviderRequest(BaseModel): + """Payload for updating a provider (partial update). + + All fields are optional -- only provided fields are updated. + ``tos_accepted``: only ``True`` re-stamps the timestamp; + ``False`` and ``None`` are no-ops (cannot be retracted). + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + driver: NotBlankStr | None = None + litellm_provider: NotBlankStr | None = None + auth_type: AuthType | None = None + api_key: NotBlankStr | None = None + clear_api_key: bool = False + subscription_token: NotBlankStr | None = None + clear_subscription_token: bool = False + tos_accepted: bool | None = None + base_url: NotBlankStr | None = None + oauth_token_url: NotBlankStr | None = None + oauth_client_id: NotBlankStr | None = None + oauth_client_secret: NotBlankStr | None = None + oauth_scope: NotBlankStr | None = None + custom_header_name: NotBlankStr | None = None + custom_header_value: NotBlankStr | None = None + models: tuple[ProviderModelConfig, ...] | None = None + + @field_validator("base_url") + @classmethod + def _validate_base_url(cls, v: str | None) -> str | None: + return _validate_base_url(v) + + @model_validator(mode="after") + def _validate_credential_clear_consistency(self) -> Self: + """Reject simultaneous set and clear for credential fields.""" + if self.api_key is not None and self.clear_api_key: + msg = "api_key and clear_api_key are mutually exclusive" + raise ValueError(msg) + if self.subscription_token is not None and self.clear_subscription_token: + msg = ( + "subscription_token and clear_subscription_token are mutually exclusive" + ) + raise ValueError(msg) + return self + + +class TestConnectionRequest(BaseModel): + """Payload for testing a provider connection. + + Attributes: + model: Model to test (defaults to first model in config). + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + model: NotBlankStr | None = None + + +class TestConnectionResponse(BaseModel): + """Result of a provider connection test. + + Attributes: + success: Whether the connection test succeeded. + latency_ms: Round-trip latency in milliseconds. + error: Error message on failure. + model_tested: Model ID that was tested. + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + success: bool + latency_ms: float | None = None + error: NotBlankStr | None = None + model_tested: NotBlankStr | None = None + + @model_validator(mode="after") + def _validate_success_error_consistency(self) -> Self: + """Ensure success and error fields are consistent.""" + if self.success and self.error is not None: + msg = "successful test must not have an error" + raise ValueError(msg) + if not self.success and self.error is None: + msg = "failed test must have an error message" + raise ValueError(msg) + return self + + +class ProviderResponse(BaseModel): + """Safe provider config for API responses -- secrets stripped. + + Non-secret auth fields are included for frontend edit form UX. + Boolean ``has_*`` indicators signal credential presence without + exposing values. + + Attributes: + driver: Driver backend name. + litellm_provider: LiteLLM routing identifier override. + auth_type: Authentication mechanism. + base_url: Provider API base URL. + models: Configured model definitions. + has_api_key: Whether an API key is set. + has_oauth_credentials: Whether OAuth credentials are configured. + has_custom_header: Whether a custom auth header is configured. + has_subscription_token: Whether a subscription token is set. + tos_accepted_at: ISO timestamp of ToS acceptance (or ``None``). + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + driver: NotBlankStr + litellm_provider: NotBlankStr | None = None + auth_type: AuthType + base_url: NotBlankStr | None + models: tuple[ProviderModelConfig, ...] + has_api_key: bool + has_oauth_credentials: bool + has_custom_header: bool + has_subscription_token: bool = False + tos_accepted_at: str | None = None + oauth_token_url: NotBlankStr | None = None + oauth_client_id: NotBlankStr | None = None + oauth_scope: NotBlankStr | None = None + custom_header_name: NotBlankStr | None = None + + +class CreateFromPresetRequest(BaseModel): + """Payload for creating a provider from a preset. + + Attributes: + preset_name: Name of the preset to create from. + name: Unique provider name (2-64 chars, lowercase + hyphens). + auth_type: Override the preset's default auth type (optional). + subscription_token: Bearer token for subscription-based auth. + tos_accepted: Whether the user accepted the subscription ToS. + base_url: Override the preset's default base URL (optional). + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + preset_name: NotBlankStr + name: NotBlankStr = Field(max_length=64) + auth_type: AuthType | None = None + api_key: NotBlankStr | None = None + subscription_token: NotBlankStr | None = None + tos_accepted: bool = False + base_url: NotBlankStr | None = None + models: tuple[ProviderModelConfig, ...] | None = None + + @field_validator("name") + @classmethod + def _validate_name(cls, v: str) -> str: + return _validate_provider_name(v) + + @field_validator("base_url") + @classmethod + def _validate_base_url(cls, v: str | None) -> str | None: + return _validate_base_url(v) + + +class DiscoverModelsResponse(BaseModel): + """Result of provider model auto-discovery. + + Attributes: + discovered_models: Models found on the provider endpoint. + provider_name: Name of the provider that was queried. + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + discovered_models: tuple[ProviderModelConfig, ...] + provider_name: NotBlankStr + + +class ProbePresetRequest(BaseModel): + """Request to probe a preset's candidate URLs for reachability. + + Attributes: + preset_name: Preset identifier to probe. + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + preset_name: NotBlankStr = Field(max_length=64) + + +class ProbePresetResponse(BaseModel): + """Result of probing a preset's candidate URLs. + + Attributes: + url: The first reachable base URL, or ``None`` if none responded. + model_count: Number of models discovered at the URL. + candidates_tried: Number of candidate URLs attempted. + """ + + model_config = ConfigDict(frozen=True, allow_inf_nan=False) + + url: NotBlankStr | None = None + model_count: int = Field(default=0, ge=0) + candidates_tried: int = Field(default=0, ge=0) + + +def to_provider_response(config: ProviderConfig) -> ProviderResponse: + """Convert a ProviderConfig to a safe ProviderResponse. + + Strips all secrets and provides boolean credential indicators. + + Args: + config: Provider configuration (may contain secrets). + + Returns: + Safe response DTO with secrets stripped. + """ + tos_str = ( + config.tos_accepted_at.isoformat() + if config.tos_accepted_at is not None + else None + ) + return ProviderResponse( + driver=config.driver, + litellm_provider=config.litellm_provider, + auth_type=config.auth_type, + base_url=config.base_url, + models=config.models, + has_api_key=config.api_key is not None, + has_oauth_credentials=( + config.oauth_client_id is not None + and config.oauth_client_secret is not None + and config.oauth_token_url is not None + ), + has_custom_header=( + config.custom_header_name is not None + and config.custom_header_value is not None + ), + has_subscription_token=config.subscription_token is not None, + tos_accepted_at=tos_str, + oauth_token_url=config.oauth_token_url, + oauth_client_id=config.oauth_client_id, + oauth_scope=config.oauth_scope, + custom_header_name=config.custom_header_name, + ) + + +# ── Enriched model response ───────────────────────────────── + + def to_provider_model_response( config: ProviderModelConfig, capabilities: ModelCapabilities | None = None, diff --git a/src/synthorg/api/guards.py b/src/synthorg/api/guards.py index 923934a331..0ff32831d7 100644 --- a/src/synthorg/api/guards.py +++ b/src/synthorg/api/guards.py @@ -68,6 +68,15 @@ def _get_role(connection: ASGIConnection) -> HumanRole | None: # type: ignore[t return None +def has_write_role(role: HumanRole) -> bool: + """Return True if the role grants write access. + + Use this for inline role checks instead of importing ``_WRITE_ROLES`` + directly. The write set includes CEO, Manager, and Pair Programmer. + """ + return role in _WRITE_ROLES + + def require_write_access( connection: ASGIConnection, # type: ignore[type-arg] _: object, diff --git a/src/synthorg/hr/activity.py b/src/synthorg/hr/activity.py index 6016cbb5a3..9937f354d8 100644 --- a/src/synthorg/hr/activity.py +++ b/src/synthorg/hr/activity.py @@ -5,13 +5,18 @@ filters career-relevant events. """ +import re from typing import TYPE_CHECKING from pydantic import AwareDatetime, BaseModel, ConfigDict, Field from synthorg.budget.currency import DEFAULT_CURRENCY, format_cost_detail from synthorg.core.types import NotBlankStr # noqa: TC001 -from synthorg.hr.enums import LifecycleEventType +from synthorg.hr.enums import ActivityEventType, LifecycleEventType +from synthorg.observability import get_logger +from synthorg.observability.events.hr import HR_ACTIVITY_REDACTION_MISMATCH + +logger = get_logger(__name__) if TYPE_CHECKING: from synthorg.budget.cost_record import CostRecord @@ -33,7 +38,7 @@ class ActivityEvent(BaseModel): model_config = ConfigDict(frozen=True, allow_inf_nan=False) - event_type: NotBlankStr = Field(description="Event category") + event_type: ActivityEventType = Field(description="Event category") timestamp: AwareDatetime = Field(description="When the event occurred") description: str = Field( default="", @@ -59,7 +64,7 @@ class CareerEvent(BaseModel): model_config = ConfigDict(frozen=True, allow_inf_nan=False) - event_type: NotBlankStr = Field(description="Lifecycle event type") + event_type: LifecycleEventType = Field(description="Lifecycle event type") timestamp: AwareDatetime = Field(description="When the event occurred") description: str = Field( default="", @@ -89,10 +94,11 @@ class CareerEvent(BaseModel): def _lifecycle_to_activity(event: AgentLifecycleEvent) -> ActivityEvent: """Convert a lifecycle event to a timeline activity event.""" + activity_type = ActivityEventType(event.event_type.value) return ActivityEvent( - event_type=event.event_type.value, + event_type=activity_type, timestamp=event.timestamp, - description=event.details or f"Agent {event.event_type.value}", + description=event.details or f"Agent {activity_type.value}", related_ids={"agent_id": str(event.agent_id)}, ) @@ -110,7 +116,7 @@ def _task_metric_to_activity( f"{format_cost_detail(record.cost_usd, currency)})" ) return ActivityEvent( - event_type="task_completed", + event_type=ActivityEventType.TASK_COMPLETED, timestamp=record.completed_at, description=desc, related_ids={ @@ -127,9 +133,11 @@ def _task_metric_to_started_activity( Caller must ensure ``record.started_at`` is not None. """ - assert record.started_at is not None # noqa: S101 + if record.started_at is None: + msg = "started_at must not be None" + raise ValueError(msg) return ActivityEvent( - event_type="task_started", + event_type=ActivityEventType.TASK_STARTED, timestamp=record.started_at, description=f"Task {record.task_id} started", related_ids={ @@ -151,7 +159,7 @@ def _cost_record_to_activity( f"{format_cost_detail(record.cost_usd, currency)})" ) return ActivityEvent( - event_type="cost_incurred", + event_type=ActivityEventType.COST_INCURRED, timestamp=record.timestamp, description=desc, related_ids={ @@ -175,7 +183,7 @@ def _tool_invocation_to_activity( if record.task_id is not None: related_ids["task_id"] = str(record.task_id) return ActivityEvent( - event_type="tool_used", + event_type=ActivityEventType.TOOL_USED, timestamp=record.timestamp, description=desc, related_ids=related_ids, @@ -187,7 +195,7 @@ def _delegation_to_sent_activity( ) -> ActivityEvent: """Convert a delegation record to a delegation_sent activity event.""" return ActivityEvent( - event_type="delegation_sent", + event_type=ActivityEventType.DELEGATION_SENT, timestamp=record.timestamp, description=( f"Delegated task {record.original_task_id} to {record.delegatee_id}" @@ -207,7 +215,7 @@ def _delegation_to_received_activity( ) -> ActivityEvent: """Convert a delegation record to a delegation_received activity event.""" return ActivityEvent( - event_type="delegation_received", + event_type=ActivityEventType.DELEGATION_RECEIVED, timestamp=record.timestamp, description=( f"Received delegation of task {record.original_task_id} " @@ -223,6 +231,52 @@ def _delegation_to_received_activity( ) +# ── Cost event redaction ──────────────────────────────────────── + +# Coupled to the format string in _cost_record_to_activity -- update +# both together if the description format changes. +_COST_DESC_PATTERN = re.compile( + r"^API call to [^(]+ \((\d+\+\d+ tokens), [^)]+\)$", +) + + +def redact_cost_events( + timeline: tuple[ActivityEvent, ...], +) -> tuple[ActivityEvent, ...]: + """Redact model names and costs from cost_incurred event descriptions. + + Produces a new timeline with sensitive details stripped from + ``cost_incurred`` event descriptions. Non-cost events pass through + unchanged. + + Args: + timeline: Activity events (may contain cost_incurred events). + + Returns: + Timeline with redacted cost event descriptions. + """ + result: list[ActivityEvent] = [] + for event in timeline: + if event.event_type == ActivityEventType.COST_INCURRED: + match = _COST_DESC_PATTERN.match(event.description) + if match: + redacted = f"API call ({match.group(1)})" + else: + logger.warning( + HR_ACTIVITY_REDACTION_MISMATCH, + event_type=event.event_type.value, + description_length=len(event.description), + ) + redacted = "API call (details redacted)" + redacted_event = event.model_copy( + update={"description": redacted}, + ) + result.append(redacted_event) + continue + result.append(event) + return tuple(result) + + # ── Timeline builders ──────────────────────────────────────────── @@ -291,7 +345,7 @@ def filter_career_events( """ career: list[CareerEvent] = [ CareerEvent( - event_type=e.event_type.value, + event_type=e.event_type, timestamp=e.timestamp, description=e.details or f"Agent {e.event_type.value}", initiated_by=e.initiated_by, diff --git a/src/synthorg/hr/enums.py b/src/synthorg/hr/enums.py index b07143ea78..86cf056207 100644 --- a/src/synthorg/hr/enums.py +++ b/src/synthorg/hr/enums.py @@ -41,6 +41,38 @@ class LifecycleEventType(StrEnum): DEMOTED = "demoted" +class ActivityEventType(StrEnum): + """Event types produced by the activity feed timeline. + + Superset of ``LifecycleEventType`` plus operational event types + generated from task metrics, cost records, tool invocations, + and delegation records. + """ + + HIRED = "hired" + ONBOARDED = "onboarded" + FIRED = "fired" + OFFBOARDED = "offboarded" + STATUS_CHANGED = "status_changed" + PROMOTED = "promoted" + DEMOTED = "demoted" + TASK_STARTED = "task_started" + TASK_COMPLETED = "task_completed" + COST_INCURRED = "cost_incurred" + TOOL_USED = "tool_used" + DELEGATION_SENT = "delegation_sent" + DELEGATION_RECEIVED = "delegation_received" + + +# Import-time check: ActivityEventType must be a superset of LifecycleEventType. +_lifecycle_values = {e.value for e in LifecycleEventType} +_activity_values = {e.value for e in ActivityEventType} +assert _lifecycle_values <= _activity_values, ( # noqa: S101 + "ActivityEventType must be superset of LifecycleEventType; " + f"missing: {_lifecycle_values - _activity_values}" +) + + class PromotionDirection(StrEnum): """Direction of a seniority level change.""" diff --git a/src/synthorg/observability/events/hr.py b/src/synthorg/observability/events/hr.py index 1abb940396..ffea877357 100644 --- a/src/synthorg/observability/events/hr.py +++ b/src/synthorg/observability/events/hr.py @@ -43,3 +43,7 @@ HR_FIRING_ARCHIVAL_FAILED: Final[str] = "hr.firing.archival_failed" HR_FIRING_NOTIFICATION_FAILED: Final[str] = "hr.firing.notification_failed" HR_ARCHIVAL_ENTRY_FAILED: Final[str] = "hr.archival.entry_failed" + +# ── Activity timeline ────────────────────────────────────────── + +HR_ACTIVITY_REDACTION_MISMATCH: Final[str] = "hr.activity.redaction_pattern_mismatch" diff --git a/tests/unit/api/controllers/test_activities.py b/tests/unit/api/controllers/test_activities.py index aed17c776b..d47c1fbf77 100644 --- a/tests/unit/api/controllers/test_activities.py +++ b/tests/unit/api/controllers/test_activities.py @@ -2,8 +2,10 @@ from datetime import UTC, datetime, timedelta from typing import Any +from unittest.mock import AsyncMock import pytest +from litestar import Litestar from litestar.testing import TestClient from synthorg.budget.cost_record import CostRecord @@ -14,13 +16,15 @@ ) from synthorg.config.schema import RootConfig from synthorg.core.enums import Complexity, TaskType -from synthorg.hr.enums import LifecycleEventType +from synthorg.hr.enums import ActivityEventType, LifecycleEventType from synthorg.hr.models import AgentLifecycleEvent from synthorg.hr.performance.models import TaskMetricRecord from synthorg.hr.performance.tracker import PerformanceTracker from synthorg.tools.invocation_record import ToolInvocationRecord from synthorg.tools.invocation_tracker import ToolInvocationTracker -from tests.unit.api.conftest import FakePersistenceBackend +from tests.unit.api.conftest import FakePersistenceBackend, make_auth_headers + +pytestmark = pytest.mark.unit _NOW = datetime(2026, 3, 24, 12, 0, 0, tzinfo=UTC) _AGENT_ID = "00000000-0000-0000-0000-000000000aaa" @@ -82,7 +86,50 @@ def _make_task_metric( ) -@pytest.mark.unit +async def _make_app_with_broken_tracker( + fake_persistence: FakePersistenceBackend, +) -> Litestar: + """Build an app whose performance tracker always raises.""" + from synthorg.api.app import create_app + from synthorg.api.auth.service import AuthService + from synthorg.budget.tracker import CostTracker + from synthorg.settings.registry import get_registry + from synthorg.settings.service import SettingsService + from tests.unit.api.conftest import ( + FakeMessageBus, + _make_test_auth_service, + _seed_test_users, + ) + + tracker = PerformanceTracker() + + def _raise(**_kwargs: object) -> None: + msg = "simulated failure" + raise RuntimeError(msg) + + tracker.get_task_metrics = _raise # type: ignore[assignment] + + config = RootConfig(company_name="test") + auth_service: AuthService = _make_test_auth_service() + bus = FakeMessageBus() + await bus.start() + _seed_test_users(fake_persistence, auth_service) + settings_service = SettingsService( + repository=fake_persistence.settings, + registry=get_registry(), + config=config, + ) + return create_app( + config=config, + persistence=fake_persistence, + message_bus=bus, + cost_tracker=CostTracker(), + auth_service=auth_service, + settings_service=settings_service, + performance_tracker=tracker, + ) + + class TestActivityFeed: def test_empty_feed(self, test_client: TestClient[Any]) -> None: resp = test_client.get("/api/v1/activities") @@ -260,63 +307,16 @@ async def test_graceful_degradation_no_performance_tracker( ) -> None: """Endpoint still returns lifecycle events when perf tracker fails.""" await fake_persistence.lifecycle_events.save( - _make_lifecycle_event( - timestamp=_NOW - timedelta(hours=1), - ), - ) - - # Build a tracker whose get_task_metrics always raises - from synthorg.hr.performance.tracker import PerformanceTracker - - tracker = PerformanceTracker() - - def _raise(**_kwargs: object) -> None: - msg = "simulated failure" - raise RuntimeError(msg) - - tracker.get_task_metrics = _raise # type: ignore[assignment] - - from synthorg.api.app import create_app - from synthorg.api.auth.service import AuthService - from synthorg.budget.tracker import CostTracker - from synthorg.settings.registry import get_registry - from synthorg.settings.service import SettingsService - from tests.unit.api.conftest import ( - FakeMessageBus, - _make_test_auth_service, - _seed_test_users, - make_auth_headers, - ) - - config = RootConfig(company_name="test") - auth_service: AuthService = _make_test_auth_service() - bus = FakeMessageBus() - await bus.start() - _seed_test_users(fake_persistence, auth_service) - settings_service = SettingsService( - repository=fake_persistence.settings, - registry=get_registry(), - config=config, - ) - app = create_app( - config=config, - persistence=fake_persistence, - message_bus=bus, - cost_tracker=CostTracker(), - auth_service=auth_service, - settings_service=settings_service, - performance_tracker=tracker, + _make_lifecycle_event(timestamp=_NOW - timedelta(hours=1)), ) - + app = await _make_app_with_broken_tracker(fake_persistence) with TestClient(app) as client: resp = client.get( "/api/v1/activities", headers=make_auth_headers("ceo"), ) assert resp.status_code == 200 - body = resp.json() - # Should still return lifecycle events - assert body["pagination"]["total"] >= 1 + assert resp.json()["pagination"]["total"] == 1 async def test_feed_with_cost_records( self, @@ -458,7 +458,7 @@ async def _raise(**_kw: object) -> None: resp = test_client.get("/api/v1/activities") assert resp.status_code == 200 body = resp.json() - assert body["pagination"]["total"] >= 1 + assert body["pagination"]["total"] == 1 async def test_graceful_degradation_broken_delegation_store( self, @@ -483,4 +483,246 @@ async def _raise(*_a: object, **_kw: object) -> None: resp = test_client.get("/api/v1/activities") assert resp.status_code == 200 body = resp.json() - assert body["pagination"]["total"] >= 1 + assert body["pagination"]["total"] == 1 + + def test_filter_by_invalid_type_returns_400( + self, + test_client: TestClient[Any], + ) -> None: + """Invalid event_type values are rejected with 400.""" + resp = test_client.get( + "/api/v1/activities", + params={"type": "bogus_event"}, + ) + assert resp.status_code == 400 + + @pytest.mark.parametrize("event_type", list(ActivityEventType)) + def test_filter_by_valid_enum_types( + self, + test_client: TestClient[Any], + event_type: ActivityEventType, + ) -> None: + """All 13 known event type values are accepted.""" + resp = test_client.get( + "/api/v1/activities", + params={"type": event_type.value}, + ) + assert resp.status_code == 200 + + async def test_service_unavailable_error_propagates( + self, + test_client: TestClient[Any], + fake_persistence: FakePersistenceBackend, + performance_tracker: PerformanceTracker, + ) -> None: + """ServiceUnavailableError from a fetcher results in 503.""" + from synthorg.api.errors import ServiceUnavailableError + + await fake_persistence.lifecycle_events.save( + _make_lifecycle_event(timestamp=_NOW - timedelta(hours=1)), + ) + + def _raise_svc(**_kw: object) -> None: + msg = "service down" + raise ServiceUnavailableError(msg) + + performance_tracker.get_task_metrics = _raise_svc # type: ignore[assignment] + + resp = test_client.get("/api/v1/activities") + assert resp.status_code == 503 + + +def _make_cost_record( + *, + agent_id: str = _AGENT_ID, + timestamp: datetime | None = None, +) -> CostRecord: + return CostRecord( + agent_id=agent_id, + task_id="task-001", + provider="test-provider", + model="test-medium-001", + input_tokens=500, + output_tokens=100, + cost_usd=0.005, + timestamp=timestamp or _NOW - timedelta(hours=1), + ) + + +class TestCostEventRedaction: + """Cost event descriptions are redacted for read-only roles.""" + + @pytest.mark.parametrize("role", ["ceo", "manager", "pair_programmer"]) + async def test_write_role_sees_full_cost_description( + self, + test_client: TestClient[Any], + cost_tracker: CostTracker, + role: str, + ) -> None: + """Write roles see the unredacted cost description.""" + await cost_tracker.record(_make_cost_record()) + resp = test_client.get( + "/api/v1/activities", + headers=make_auth_headers(role), + ) + assert resp.status_code == 200 + desc = resp.json()["data"][0]["description"] + assert "test-medium-001" in desc + assert "tokens" in desc + + @pytest.mark.parametrize("role", ["observer", "board_member"]) + async def test_read_role_sees_redacted_cost_description( + self, + test_client: TestClient[Any], + cost_tracker: CostTracker, + role: str, + ) -> None: + """Read-only roles see redacted cost descriptions.""" + await cost_tracker.record(_make_cost_record()) + resp = test_client.get( + "/api/v1/activities", + headers=make_auth_headers(role), + ) + assert resp.status_code == 200 + desc = resp.json()["data"][0]["description"] + assert "test-medium-001" not in desc + assert "500+100 tokens" in desc + assert desc == "API call (500+100 tokens)" + + async def test_missing_auth_user_sees_redacted( + self, + test_client: TestClient[Any], + cost_tracker: CostTracker, + ) -> None: + """Read-only role triggers redaction (fail-closed).""" + await cost_tracker.record(_make_cost_record()) + # Use a read-only role to verify fail-closed redaction behavior. + # The default test_client uses CEO (write role); observer proves + # that non-write users get redacted descriptions. + resp = test_client.get( + "/api/v1/activities", + headers=make_auth_headers("observer"), + ) + assert resp.status_code == 200 + body = resp.json() + cost_events = [e for e in body["data"] if e["event_type"] == "cost_incurred"] + for event in cost_events: + assert "test-medium-001" not in event["description"] + + +class TestDegradedSources: + """Degraded data sources are reported in the response.""" + + def test_no_degradation_returns_empty_list( + self, + test_client: TestClient[Any], + ) -> None: + resp = test_client.get("/api/v1/activities") + assert resp.status_code == 200 + assert resp.json()["degraded_sources"] == [] + + async def test_degraded_performance_tracker( + self, + fake_persistence: FakePersistenceBackend, + ) -> None: + """Performance tracker failure is reported in degraded_sources.""" + await fake_persistence.lifecycle_events.save( + _make_lifecycle_event(timestamp=_NOW - timedelta(hours=1)), + ) + app = await _make_app_with_broken_tracker(fake_persistence) + with TestClient(app) as client: + resp = client.get( + "/api/v1/activities", + headers=make_auth_headers("ceo"), + ) + assert resp.status_code == 200 + assert "performance_tracker" in resp.json()["degraded_sources"] + + async def test_degraded_tool_tracker( + self, + test_client: TestClient[Any], + fake_persistence: FakePersistenceBackend, + tool_invocation_tracker: ToolInvocationTracker, + ) -> None: + """Tool tracker failure is reported in degraded_sources.""" + await fake_persistence.lifecycle_events.save( + _make_lifecycle_event(timestamp=_NOW - timedelta(hours=1)), + ) + + async def _raise(**_kw: object) -> None: + msg = "simulated failure" + raise RuntimeError(msg) + + tool_invocation_tracker.get_records = _raise # type: ignore[assignment] + + resp = test_client.get("/api/v1/activities") + assert resp.status_code == 200 + body = resp.json() + assert "tool_invocation_tracker" in body["degraded_sources"] + + async def test_degraded_delegation_store( + self, + test_client: TestClient[Any], + fake_persistence: FakePersistenceBackend, + delegation_record_store: DelegationRecordStore, + ) -> None: + """Delegation store failure is reported in degraded_sources.""" + await fake_persistence.lifecycle_events.save( + _make_lifecycle_event(timestamp=_NOW - timedelta(hours=1)), + ) + + async def _raise(*_a: object, **_kw: object) -> None: + msg = "simulated failure" + raise RuntimeError(msg) + + delegation_record_store.get_all_records = _raise # type: ignore[assignment] + delegation_record_store.get_records_as_delegator = _raise # type: ignore[assignment] + delegation_record_store.get_records_as_delegatee = _raise # type: ignore[assignment] + + resp = test_client.get("/api/v1/activities") + assert resp.status_code == 200 + body = resp.json() + assert "delegation_record_store" in body["degraded_sources"] + + async def test_degraded_cost_tracker( + self, + test_client: TestClient[Any], + fake_persistence: FakePersistenceBackend, + cost_tracker: CostTracker, + ) -> None: + """Cost tracker failure is reported in degraded_sources.""" + await fake_persistence.lifecycle_events.save( + _make_lifecycle_event(timestamp=_NOW - timedelta(hours=1)), + ) + + async def _raise(**_kw: object) -> None: + msg = "simulated failure" + raise RuntimeError(msg) + + cost_tracker.get_records = _raise # type: ignore[assignment] + + resp = test_client.get("/api/v1/activities") + assert resp.status_code == 200 + assert "cost_tracker" in resp.json()["degraded_sources"] + + async def test_degraded_budget_config( + self, + test_client: TestClient[Any], + fake_persistence: FakePersistenceBackend, + ) -> None: + """Budget config failure is reported in degraded_sources.""" + await fake_persistence.lifecycle_events.save( + _make_lifecycle_event(timestamp=_NOW - timedelta(hours=1)), + ) + # Break the config resolver + app_state = test_client.app.state.app_state + original = app_state.config_resolver.get_budget_config + app_state.config_resolver.get_budget_config = AsyncMock( + side_effect=RuntimeError("simulated failure"), + ) + try: + resp = test_client.get("/api/v1/activities") + assert resp.status_code == 200 + assert "budget_config" in resp.json()["degraded_sources"] + finally: + app_state.config_resolver.get_budget_config = original diff --git a/tests/unit/api/test_dto.py b/tests/unit/api/test_dto.py index 2a9e9fb9d2..f98ca97cd3 100644 --- a/tests/unit/api/test_dto.py +++ b/tests/unit/api/test_dto.py @@ -170,6 +170,25 @@ def test_error_detail_on_paginated_success_rejected(self) -> None: ) +class TestPaginatedResponseDegradedSources: + def test_degraded_sources_defaults_to_empty(self) -> None: + resp: PaginatedResponse[str] = PaginatedResponse( + pagination=PaginationMeta(total=0, offset=0, limit=10), + ) + assert resp.degraded_sources == () + + def test_degraded_sources_round_trips_through_json(self) -> None: + resp: PaginatedResponse[str] = PaginatedResponse( + pagination=PaginationMeta(total=0, offset=0, limit=10), + degraded_sources=("cost_tracker", "tool_invocation_tracker"), + ) + data = resp.model_dump(mode="json") + assert data["degraded_sources"] == [ + "cost_tracker", + "tool_invocation_tracker", + ] + + class TestPaginatedResponseErrorDetail: def test_error_detail_defaults_to_none(self) -> None: resp: PaginatedResponse[str] = PaginatedResponse( diff --git a/tests/unit/hr/test_activity.py b/tests/unit/hr/test_activity.py index 4346fadcf8..20bb5e4635 100644 --- a/tests/unit/hr/test_activity.py +++ b/tests/unit/hr/test_activity.py @@ -7,8 +7,14 @@ from synthorg.budget.cost_record import CostRecord from synthorg.communication.delegation.models import DelegationRecord from synthorg.core.enums import Complexity, TaskType -from synthorg.hr.activity import filter_career_events, merge_activity_timeline -from synthorg.hr.enums import LifecycleEventType +from synthorg.hr.activity import ( + ActivityEvent, + _cost_record_to_activity, + filter_career_events, + merge_activity_timeline, + redact_cost_events, +) +from synthorg.hr.enums import ActivityEventType, LifecycleEventType from synthorg.hr.models import AgentLifecycleEvent from synthorg.hr.performance.models import TaskMetricRecord from synthorg.tools.invocation_record import ToolInvocationRecord @@ -703,3 +709,123 @@ def test_metadata_and_initiated_by_preserved(self) -> None: assert career[0].initiated_by == "ceo" assert career[0].description == "Hired as developer" assert career[0].metadata == {"reason": "expansion", "team": "backend"} + + +@pytest.mark.unit +class TestActivityEventTypeSuperset: + """ActivityEventType must include every LifecycleEventType value.""" + + def test_lifecycle_values_are_subset(self) -> None: + lifecycle_values = {e.value for e in LifecycleEventType} + activity_values = {e.value for e in ActivityEventType} + missing = lifecycle_values - activity_values + assert not missing, ( + f"LifecycleEventType values missing from ActivityEventType: {missing}" + ) + + +@pytest.mark.unit +class TestRedactCostEvents: + def test_redacts_model_and_cost_from_description(self) -> None: + event = ActivityEvent( + event_type=ActivityEventType.COST_INCURRED, + timestamp=_NOW, + description="API call to test-medium-001 (500+100 tokens, EUR0.01)", + ) + result = redact_cost_events((event,)) + assert len(result) == 1 + assert result[0].description == "API call (500+100 tokens)" + + def test_leaves_non_cost_events_unchanged(self) -> None: + hired = ActivityEvent( + event_type=ActivityEventType.HIRED, + timestamp=_NOW, + description="Agent hired", + ) + tool = ActivityEvent( + event_type=ActivityEventType.TOOL_USED, + timestamp=_NOW, + description="Tool read_file executed successfully", + ) + result = redact_cost_events((hired, tool)) + assert result[0].description == "Agent hired" + assert result[1].description == "Tool read_file executed successfully" + + def test_empty_timeline(self) -> None: + assert redact_cost_events(()) == () + + def test_preserves_other_fields(self) -> None: + event = ActivityEvent( + event_type=ActivityEventType.COST_INCURRED, + timestamp=_NOW, + description="API call to test-small-001 (200+50 tokens, $0.005)", + related_ids={"agent_id": "agent-1", "task_id": "task-1"}, + ) + result = redact_cost_events((event,)) + assert result[0].event_type == "cost_incurred" + assert result[0].timestamp == _NOW + assert result[0].related_ids == {"agent_id": "agent-1", "task_id": "task-1"} + + def test_non_matching_description_uses_fallback(self) -> None: + """If the description format doesn't match the regex, use fallback.""" + event = ActivityEvent( + event_type=ActivityEventType.COST_INCURRED, + timestamp=_NOW, + description="Custom cost event description", + ) + result = redact_cost_events((event,)) + assert result[0].description == "API call (details redacted)" + + def test_round_trip_with_real_cost_record(self) -> None: + """Regex matches the actual format produced by _cost_record_to_activity.""" + record = CostRecord( + agent_id="agent-1", + task_id="task-1", + provider="test-provider", + model="test-medium-001", + input_tokens=500, + output_tokens=100, + cost_usd=0.005, + timestamp=_NOW, + ) + event = _cost_record_to_activity(record, currency="EUR") + result = redact_cost_events((event,)) + assert "test-medium-001" not in result[0].description + assert "500+100 tokens" in result[0].description + + def test_mixed_timeline_preserves_order(self) -> None: + """Redaction preserves event ordering in a mixed timeline.""" + events = ( + ActivityEvent( + event_type=ActivityEventType.HIRED, + timestamp=_NOW, + description="Agent hired", + ), + ActivityEvent( + event_type=ActivityEventType.COST_INCURRED, + timestamp=_NOW, + description="API call to test-model (100+50 tokens, EUR0.01)", + ), + ActivityEvent( + event_type=ActivityEventType.TOOL_USED, + timestamp=_NOW, + description="Tool read_file executed successfully", + ), + ActivityEvent( + event_type=ActivityEventType.COST_INCURRED, + timestamp=_NOW, + description="API call to test-model-2 (200+100 tokens, EUR0.02)", + ), + ActivityEvent( + event_type=ActivityEventType.DELEGATION_SENT, + timestamp=_NOW, + description="Delegated task t1 to a2", + ), + ) + result = redact_cost_events(events) + assert len(result) == 5 + assert result[0].description == "Agent hired" + assert result[1].description == "API call (100+50 tokens)" + assert result[2].description == "Tool read_file executed successfully" + assert result[3].description == "API call (200+100 tokens)" + assert result[4].description == "Delegated task t1 to a2" diff --git a/web/src/api/endpoints/activities.ts b/web/src/api/endpoints/activities.ts index 5d28d76488..0dc08393ab 100644 --- a/web/src/api/endpoints/activities.ts +++ b/web/src/api/endpoints/activities.ts @@ -1,9 +1,31 @@ import { apiClient, unwrapPaginated, type PaginatedResult } from '../client' -import type { ActivityItem, PaginatedResponse, PaginationParams } from '../types' +import type { ActivityEvent, ActivityEventType, ActivityItem, PaginatedResponse, PaginationParams } from '../types' + +export interface ActivityFilterParams extends PaginationParams { + type?: ActivityEventType + agent_id?: string + last_n_hours?: 24 | 48 | 168 +} + +/** Map a REST ActivityEvent to the display-oriented ActivityItem shape. */ +export function mapActivityEventToItem(event: ActivityEvent): ActivityItem { + const agentId = event.related_ids.agent_id ?? 'System' + const taskId = event.related_ids.task_id ?? null + return { + id: taskId ?? `${event.timestamp}-${event.event_type}-${agentId}`, + timestamp: event.timestamp, + agent_name: agentId, + action_type: event.event_type, + description: event.description, + task_id: taskId, + department: null, + } +} export async function listActivities( - params?: PaginationParams, + params?: ActivityFilterParams, ): Promise> { - const response = await apiClient.get>('/activities', { params }) - return unwrapPaginated(response) + const response = await apiClient.get>('/activities', { params }) + const result = unwrapPaginated(response) + return { ...result, data: result.data.map(mapActivityEventToItem) } } diff --git a/web/src/api/types.ts b/web/src/api/types.ts index d4b639795e..66cc7b8c69 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -147,8 +147,8 @@ export interface PaginationMeta { /** Discriminated paginated response envelope. */ export type PaginatedResponse = - | { data: T[]; error: null; error_detail: null; success: true; pagination: PaginationMeta } - | { data: null; error: string; error_detail: ErrorDetail; success: false; pagination: null } + | { data: T[]; error: null; error_detail: null; success: true; pagination: PaginationMeta; degraded_sources?: readonly string[] } + | { data: null; error: string; error_detail: ErrorDetail; success: false; pagination: null; degraded_sources?: readonly string[] } // ── Auth ───────────────────────────────────────────────────── @@ -539,11 +539,24 @@ export interface ForecastResponse { // ── Activities ────────────────────────────────────────────── +/** Activity event as returned by the backend REST API. */ +export interface ActivityEvent { + event_type: ActivityEventType + timestamp: string + description: string + related_ids: Record +} + +/** + * Legacy display-oriented activity item derived from {@link ActivityEvent}. + * Used by the dashboard ActivityFeed component. + */ export interface ActivityItem { id: string timestamp: string agent_name: string - action_type: WsEventType + /** REST path produces ActivityEventType; WS path produces WsEventType. */ + action_type: ActivityEventType | WsEventType description: string task_id: string | null department: DepartmentName | null diff --git a/web/src/pages/dashboard/ActivityFeedItem.tsx b/web/src/pages/dashboard/ActivityFeedItem.tsx index cbb168528c..528bdd3aaf 100644 --- a/web/src/pages/dashboard/ActivityFeedItem.tsx +++ b/web/src/pages/dashboard/ActivityFeedItem.tsx @@ -2,14 +2,30 @@ import { Link } from 'react-router' import { Avatar } from '@/components/ui/avatar' import { cn } from '@/lib/utils' import { formatRelativeTime } from '@/utils/format' -import type { ActivityItem, WsEventType } from '@/api/types' +import type { ActivityEventType, ActivityItem, WsEventType } from '@/api/types' interface ActivityFeedItemProps { activity: ActivityItem className?: string } -const ACTION_DOT_COLORS: Partial> = { +/** Dot colors for both REST ActivityEventType and WS WsEventType keys. */ +const ACTION_DOT_COLORS: Partial> = { + // REST activity event types + hired: 'bg-success', + fired: 'bg-danger', + onboarded: 'bg-success', + offboarded: 'bg-warning', + status_changed: 'bg-warning', + promoted: 'bg-success', + demoted: 'bg-warning', + task_started: 'bg-accent', + task_completed: 'bg-success', + cost_incurred: 'bg-warning', + tool_used: 'bg-accent', + delegation_sent: 'bg-accent', + delegation_received: 'bg-accent', + // WS event types 'task.created': 'bg-accent', 'task.updated': 'bg-accent', 'task.status_changed': 'bg-success', @@ -36,7 +52,7 @@ const ACTION_DOT_COLORS: Partial> = { 'system.shutdown': 'bg-warning', } -function getActionDotColor(actionType: WsEventType): string { +function getActionDotColor(actionType: ActivityEventType | WsEventType): string { return ACTION_DOT_COLORS[actionType] ?? 'bg-muted-foreground' }