diff --git a/Server/src/main.py b/Server/src/main.py index b057f3d96..90c8b5101 100644 --- a/Server/src/main.py +++ b/Server/src/main.py @@ -14,6 +14,7 @@ CustomToolService, resolve_project_id_for_unity_instance, ) +from services.overseer_ledger import register_overseer_routes from core.config import config from starlette.routing import WebSocketRoute from starlette.responses import JSONResponse @@ -383,6 +384,11 @@ def create_mcp_server(project_scoped_tools: bool) -> FastMCP: global custom_tool_service custom_tool_service = CustomToolService( mcp, project_scoped_tools=project_scoped_tools) + register_overseer_routes( + mcp, + include_internal_routes=not config.http_remote_hosted, + allow_default_tokens=not config.http_remote_hosted, + ) @mcp.custom_route("/health", methods=["GET"]) async def health_http(_: Request) -> JSONResponse: diff --git a/Server/src/services/overseer_ledger.py b/Server/src/services/overseer_ledger.py new file mode 100644 index 000000000..5bea9e0e8 --- /dev/null +++ b/Server/src/services/overseer_ledger.py @@ -0,0 +1,469 @@ +"""Local customer-support overseer simulation and read-only dashboard routes. + +This module deliberately has no provider integrations. It gives the dashboard a +deterministic event stream that can be replaced by real adapters later without +changing the read model or its authorization boundary. +""" + +from __future__ import annotations + +import copy +import os +import threading +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from enum import Enum +from typing import Any, Callable, Mapping + +from starlette.requests import Request +from starlette.responses import JSONResponse + + +class EventType(str, Enum): + """Event categories emitted by the support workflow.""" + + AGENT_ACTIVITY = "agent_activity" + REVENUE = "revenue" + COST = "cost" + APPROVAL_PENDING = "approval_pending" + APPROVAL_RESOLVED = "approval_resolved" + FAILED_ACTION = "failed_action" + AGENT_HEALTH = "agent_health" + + +class Role(str, Enum): + """Roles supported by the local dashboard authentication boundary.""" + + VIEWER = "viewer" + OPERATOR = "operator" + ADMIN = "admin" + + +@dataclass(frozen=True) +class AuthPrincipal: + """The role associated with a successfully authenticated bearer token.""" + + role: Role + + +@dataclass(frozen=True) +class OverseerEvent: + """An immutable ledger entry.""" + + event_id: str + event_type: EventType + occurred_at: str + run_id: str + team: str | None = None + agent_id: str | None = None + data: Mapping[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + payload = asdict(self) + payload["event_type"] = self.event_type.value + payload["data"] = copy.deepcopy(dict(self.data)) + return payload + + +@dataclass(frozen=True) +class SimulationReport: + """Summary returned after a deterministic simulation run.""" + + run_id: str + emitted_event_ids: tuple[str, ...] + event_count: int + + def to_dict(self) -> dict[str, Any]: + return { + "run_id": self.run_id, + "emitted_event_ids": list(self.emitted_event_ids), + "event_count": self.event_count, + } + + +class EventLedger: + """Thread-safe in-memory event ledger and read-model projections.""" + + def __init__(self, clock: Callable[[], datetime] | None = None): + self._clock = clock or (lambda: datetime.now(timezone.utc)) + self._events: list[OverseerEvent] = [] + self._event_sequence = 0 + self._run_sequence = 0 + self._lock = threading.RLock() + + @property + def events(self) -> list[OverseerEvent]: + """Return a copy so callers cannot mutate the ledger.""" + with self._lock: + return copy.deepcopy(self._events) + + def emit( + self, + event_type: EventType, + *, + run_id: str, + team: str | None = None, + agent_id: str | None = None, + data: Mapping[str, Any] | None = None, + ) -> OverseerEvent: + with self._lock: + self._event_sequence += 1 + event = OverseerEvent( + event_id=f"evt-{self._event_sequence:06d}", + event_type=event_type, + occurred_at=self._clock().astimezone(timezone.utc).isoformat(), + run_id=run_id, + team=team, + agent_id=agent_id, + data=copy.deepcopy(dict(data or {})), + ) + self._events.append(event) + return event + + def run_simulation(self, cycles: int = 1) -> SimulationReport: + """Emit a predictable support workflow without contacting any provider.""" + if not isinstance(cycles, int) or isinstance(cycles, bool) or not 1 <= cycles <= 10: + raise ValueError("cycles must be an integer between 1 and 10") + + emitted: list[str] = [] + with self._lock: + for _ in range(cycles): + self._run_sequence += 1 + run_id = f"sim-{self._run_sequence:03d}" + specs = ( + ("agent-support-1", "support", "ticket_triage", 4800, 1200, 0.98), + ("agent-sales-1", "sales", "renewal_followup", 7200, 2100, 0.96), + ) + for agent_id, team, task, revenue, cost, health in specs: + emitted.append(self.emit( + EventType.AGENT_ACTIVITY, + run_id=run_id, + team=team, + agent_id=agent_id, + data={"status": "working", "task": task, "progress": 0.5}, + ).event_id) + emitted.append(self.emit( + EventType.REVENUE, + run_id=run_id, + team=team, + agent_id=agent_id, + data={"amount": revenue, "currency": "USD", "source": "simulation"}, + ).event_id) + emitted.append(self.emit( + EventType.COST, + run_id=run_id, + team=team, + agent_id=agent_id, + data={"amount": cost, "currency": "USD", "source": "simulation"}, + ).event_id) + emitted.append(self.emit( + EventType.AGENT_HEALTH, + run_id=run_id, + team=team, + agent_id=agent_id, + data={"status": "healthy", "score": health}, + ).event_id) + + approval_id = f"approval-{self._run_sequence:03d}" + emitted.append(self.emit( + EventType.APPROVAL_PENDING, + run_id=run_id, + team="support", + agent_id="agent-support-1", + data={ + "approval_id": approval_id, + "action": "issue_refund", + "amount": 325, + "currency": "USD", + "status": "pending", + }, + ).event_id) + emitted.append(self.emit( + EventType.FAILED_ACTION, + run_id=run_id, + team="support", + agent_id="agent-support-1", + data={ + "action_id": f"failed-{self._run_sequence:03d}", + "action": "send_email", + "reason": "provider_not_connected", + "retryable": True, + }, + ).event_id) + + return SimulationReport(run_id=run_id, emitted_event_ids=tuple(emitted), event_count=len(emitted)) + + def _approval_projection(self) -> dict[str, dict[str, Any]]: + approvals: dict[str, dict[str, Any]] = {} + for event in self._events: + if event.event_type == EventType.APPROVAL_PENDING: + approval_id = str(event.data["approval_id"]) + approvals[approval_id] = { + "approval_id": approval_id, + "team": event.team, + "agent_id": event.agent_id, + "run_id": event.run_id, + "action": event.data.get("action"), + "amount": event.data.get("amount"), + "currency": event.data.get("currency"), + "status": "pending", + "requested_at": event.occurred_at, + } + elif event.event_type == EventType.APPROVAL_RESOLVED: + approval_id = str(event.data["approval_id"]) + if approval_id in approvals: + approvals[approval_id].update({ + "status": event.data.get("status", "approved"), + "resolved_at": event.occurred_at, + "resolved_by": event.data.get("resolved_by"), + }) + return approvals + + def pending_approvals(self) -> list[dict[str, Any]]: + with self._lock: + return [ + copy.deepcopy(approval) + for approval in self._approval_projection().values() + if approval["status"] == "pending" + ] + + def approve(self, approval_id: str, resolved_by: str = "admin") -> dict[str, Any]: + """Resolve one approval by appending an event, never editing history.""" + with self._lock: + approval = self._approval_projection().get(approval_id) + if approval is None: + raise KeyError(f"Approval '{approval_id}' was not found") + if approval["status"] != "pending": + raise ValueError(f"Approval '{approval_id}' is already resolved") + event = self.emit( + EventType.APPROVAL_RESOLVED, + run_id=approval["run_id"], + team=approval["team"], + agent_id=approval["agent_id"], + data={ + "approval_id": approval_id, + "status": "approved", + "resolved_by": resolved_by, + }, + ) + result = copy.deepcopy(approval) + result.update({ + "status": "approved", + "resolved_at": event.occurred_at, + "resolved_by": resolved_by, + }) + return result + + def activity(self, limit: int = 50) -> list[dict[str, Any]]: + with self._lock: + events = [event for event in self._events if event.event_type == EventType.AGENT_ACTIVITY] + return [event.to_dict() for event in reversed(events[-limit:])] + + def financials(self) -> list[dict[str, Any]]: + with self._lock: + totals: dict[str, dict[str, Any]] = {} + for event in self._events: + if event.event_type not in (EventType.REVENUE, EventType.COST) or not event.team: + continue + team = totals.setdefault(event.team, { + "team": event.team, "currency": event.data.get("currency", "USD"), + "revenue": 0, "costs": 0, "profit": 0, + }) + amount = event.data.get("amount", 0) + if event.event_type == EventType.REVENUE: + team["revenue"] += amount + else: + team["costs"] += amount + for team in totals.values(): + team["profit"] = team["revenue"] - team["costs"] + return [copy.deepcopy(totals[name]) for name in sorted(totals)] + + def failed_actions(self, limit: int = 50) -> list[dict[str, Any]]: + with self._lock: + events = [event for event in self._events if event.event_type == EventType.FAILED_ACTION] + return [event.to_dict() for event in reversed(events[-limit:])] + + def health(self) -> list[dict[str, Any]]: + with self._lock: + latest: dict[str, OverseerEvent] = {} + for event in self._events: + if event.event_type == EventType.AGENT_HEALTH and event.agent_id: + latest[event.agent_id] = event + return [event.to_dict() for event in (latest[name] for name in sorted(latest))] + + def dashboard(self, limit: int = 50) -> dict[str, Any]: + return { + "activity": self.activity(limit), + "pending_approvals": self.pending_approvals(), + "financials": self.financials(), + "failed_actions": self.failed_actions(limit), + "health": self.health(), + } + + +class AuthError(Exception): + """HTTP-friendly authentication or authorization failure.""" + + def __init__(self, status_code: int, message: str): + super().__init__(message) + self.status_code = status_code + + +class OverseerAuth: + """Bearer-token authentication with a local-only deterministic fallback.""" + + _ROLE_ORDER = {Role.VIEWER: 1, Role.OPERATOR: 2, Role.ADMIN: 3} + + def __init__(self, tokens: Mapping[Role | str, str]): + normalized: dict[str, Role] = {} + for role, token in tokens.items(): + role = Role(role) + if token: + normalized[token] = role + self._tokens = normalized + + @classmethod + def from_environment(cls, allow_defaults: bool = True) -> "OverseerAuth": + defaults = { + Role.VIEWER: "overseer-viewer", + Role.OPERATOR: "overseer-operator", + Role.ADMIN: "overseer-admin", + } + tokens = {} + for role in Role: + env_name = f"UNITY_MCP_OVERSEER_{role.value.upper()}_TOKEN" + token = os.environ.get(env_name) + if token is None and allow_defaults: + token = defaults[role] + tokens[role] = token or "" + return cls(tokens) + + def authorize(self, authorization: str | None, minimum_role: Role = Role.VIEWER) -> AuthPrincipal: + if not authorization or not authorization.startswith("Bearer "): + raise AuthError(401, "Bearer token required") + token = authorization[7:].strip() + role = self._tokens.get(token) + if role is None: + raise AuthError(401, "Invalid bearer token") + if self._ROLE_ORDER[role] < self._ROLE_ORDER[minimum_role]: + raise AuthError(403, f"{minimum_role.value} role required") + return AuthPrincipal(role=role) + + +def _response(payload: dict[str, Any], status_code: int = 200) -> JSONResponse: + return JSONResponse({"success": True, **payload}, status_code=status_code) + + +def _error(error: AuthError | Exception, status_code: int | None = None) -> JSONResponse: + return JSONResponse( + {"success": False, "error": str(error)}, + status_code=status_code or getattr(error, "status_code", 500), + ) + + +def _limit(request: Request) -> int: + try: + value = int(request.query_params.get("limit", "50")) + except ValueError as exc: + raise ValueError("limit must be an integer between 1 and 200") from exc + if not 1 <= value <= 200: + raise ValueError("limit must be an integer between 1 and 200") + return value + + +def register_overseer_routes( + mcp: Any, + ledger: EventLedger | None = None, + auth: OverseerAuth | None = None, + *, + include_internal_routes: bool = True, + allow_default_tokens: bool = True, +) -> EventLedger: + """Register authenticated dashboard routes and optional local action routes.""" + ledger = ledger or EventLedger() + auth = auth or OverseerAuth.from_environment(allow_defaults=allow_default_tokens) + + async def require(request: Request, role: Role = Role.VIEWER) -> AuthPrincipal: + return auth.authorize(request.headers.get("authorization"), role) + + @mcp.custom_route("/api/overseer/activity", methods=["GET"]) + async def overseer_activity(request: Request) -> JSONResponse: + try: + await require(request) + return _response({"activity": ledger.activity(_limit(request))}) + except (AuthError, ValueError) as exc: + return _error(exc, getattr(exc, "status_code", 400)) + + @mcp.custom_route("/api/overseer/approvals", methods=["GET"]) + async def overseer_approvals(request: Request) -> JSONResponse: + try: + await require(request) + return _response({"pending_approvals": ledger.pending_approvals()}) + except AuthError as exc: + return _error(exc) + + @mcp.custom_route("/api/overseer/financials", methods=["GET"]) + async def overseer_financials(request: Request) -> JSONResponse: + try: + await require(request) + return _response({"financials": ledger.financials()}) + except AuthError as exc: + return _error(exc) + + @mcp.custom_route("/api/overseer/failed-actions", methods=["GET"]) + async def overseer_failed_actions(request: Request) -> JSONResponse: + try: + await require(request) + return _response({"failed_actions": ledger.failed_actions(_limit(request))}) + except (AuthError, ValueError) as exc: + return _error(exc, getattr(exc, "status_code", 400)) + + @mcp.custom_route("/api/overseer/health", methods=["GET"]) + async def overseer_health(request: Request) -> JSONResponse: + try: + await require(request) + return _response({"health": ledger.health()}) + except AuthError as exc: + return _error(exc) + + @mcp.custom_route("/api/overseer/dashboard", methods=["GET"]) + async def overseer_dashboard(request: Request) -> JSONResponse: + try: + await require(request) + return _response({"dashboard": ledger.dashboard(_limit(request))}) + except (AuthError, ValueError) as exc: + return _error(exc, getattr(exc, "status_code", 400)) + + if include_internal_routes: + @mcp.custom_route("/api/internal/overseer/simulate", methods=["POST"]) + async def overseer_simulate(request: Request) -> JSONResponse: + try: + await require(request, Role.OPERATOR) + body = await request.json() + report = ledger.run_simulation(body.get("cycles", 1) if isinstance(body, dict) else 1) + return _response({"simulation": report.to_dict()}, status_code=201) + except AuthError as exc: + return _error(exc) + except (ValueError, TypeError) as exc: + return _error(exc, 400) + + @mcp.custom_route( + "/api/internal/overseer/approvals/{approval_id}/approve", + methods=["POST"], + ) + async def overseer_approve(request: Request) -> JSONResponse: + try: + principal = await require(request, Role.ADMIN) + approval_id = request.path_params["approval_id"] + return _response({ + "approval": ledger.approve(approval_id, resolved_by=principal.role.value), + }) + except AuthError as exc: + return _error(exc) + except (KeyError, ValueError) as exc: + return _error(exc, 404 if isinstance(exc, KeyError) else 409) + + # Expose the ledger for embedding applications and focused tests without + # making it part of the HTTP API. + setattr(mcp, "overseer_ledger", ledger) + return ledger diff --git a/Server/tests/test_overseer_ledger.py b/Server/tests/test_overseer_ledger.py new file mode 100644 index 000000000..7d81c38b0 --- /dev/null +++ b/Server/tests/test_overseer_ledger.py @@ -0,0 +1,143 @@ +"""Focused tests for the local overseer simulation and dashboard boundary.""" + +import json +import asyncio +from datetime import datetime, timezone + +import pytest +from starlette.requests import Request + +from services.overseer_ledger import ( + EventLedger, + EventType, + OverseerAuth, + Role, + register_overseer_routes, +) + + +class RecordingMcp: + def __init__(self): + self.routes = {} + + def custom_route(self, path, methods=None): + def decorator(handler): + self.routes[(path, tuple(methods or ()))] = handler + return handler + + return decorator + + +def request(path="/", token="overseer-viewer", method="GET", body=None, path_params=None): + headers = [(b"authorization", f"Bearer {token}".encode())] + raw_body = json.dumps(body).encode() if body is not None else b"" + headers.append((b"content-type", b"application/json")) + + async def receive(): + return {"type": "http.request", "body": raw_body, "more_body": False} + + scope = { + "type": "http", + "method": method, + "path": path, + "query_string": b"", + "headers": headers, + "path_params": path_params or {}, + } + return Request(scope, receive) + + +def test_simulation_emits_all_event_categories_and_aggregates(): + clock = lambda: datetime(2026, 1, 1, tzinfo=timezone.utc) + ledger = EventLedger(clock=clock) + + report = ledger.run_simulation() + + assert report.event_count == 10 + assert {event.event_type for event in ledger.events} == { + EventType.AGENT_ACTIVITY, + EventType.REVENUE, + EventType.COST, + EventType.APPROVAL_PENDING, + EventType.FAILED_ACTION, + EventType.AGENT_HEALTH, + } + assert ledger.financials() == [ + {"team": "sales", "currency": "USD", "revenue": 7200, "costs": 2100, "profit": 5100}, + {"team": "support", "currency": "USD", "revenue": 4800, "costs": 1200, "profit": 3600}, + ] + assert len(ledger.pending_approvals()) == 1 + assert len(ledger.failed_actions()) == 1 + assert len(ledger.health()) == 2 + + +def test_approval_is_append_only_and_requires_pending_approval(): + ledger = EventLedger() + ledger.run_simulation() + approved = ledger.approve("approval-001") + + assert approved["status"] == "approved" + assert ledger.pending_approvals() == [] + assert ledger.events[-1].event_type == EventType.APPROVAL_RESOLVED + with pytest.raises(ValueError): + ledger.approve("approval-001") + + +def test_dashboard_routes_require_bearer_auth_and_return_read_models(): + mcp = RecordingMcp() + ledger = EventLedger() + ledger.run_simulation() + auth = OverseerAuth({ + Role.VIEWER: "viewer-token", + Role.OPERATOR: "operator-token", + Role.ADMIN: "admin-token", + }) + register_overseer_routes(mcp, ledger, auth) + + dashboard = mcp.routes[("/api/overseer/dashboard", ("GET",))] + unauthorized = asyncio.run(dashboard(request(token="wrong-token"))) + assert unauthorized.status_code == 401 + + response = asyncio.run(dashboard(request(token="viewer-token"))) + assert response.status_code == 200 + payload = json.loads(response.body) + assert payload["success"] is True + assert payload["dashboard"]["financials"][0]["profit"] > 0 + + +def test_internal_routes_enforce_operator_and_admin_roles(): + mcp = RecordingMcp() + ledger = EventLedger() + auth = OverseerAuth({ + Role.VIEWER: "viewer-token", + Role.OPERATOR: "operator-token", + Role.ADMIN: "admin-token", + }) + register_overseer_routes(mcp, ledger, auth) + + simulate = mcp.routes[("/api/internal/overseer/simulate", ("POST",))] + forbidden = asyncio.run(simulate(request( + method="POST", token="viewer-token", body={"cycles": 1}, + ))) + assert forbidden.status_code == 403 + created = asyncio.run(simulate(request( + method="POST", token="operator-token", body={"cycles": 1}, + ))) + assert created.status_code == 201 + + approve = mcp.routes[( + "/api/internal/overseer/approvals/{approval_id}/approve", + ("POST",), + )] + operator_response = asyncio.run(approve(request( + method="POST", + token="operator-token", + path_params={"approval_id": "approval-001"}, + ))) + assert operator_response.status_code == 403 + admin_response = asyncio.run(approve(request( + method="POST", + token="admin-token", + path_params={"approval_id": "approval-001"}, + ))) + assert admin_response.status_code == 200