feat: add AI usage and rate limit monitoring

backend/app/main.py

@@ -1116,6 +1116,11 @@ def telemetry():
     return store.telemetry()
 
 
+@app.get("/api/ai-usage")
+def ai_usage(minutes: int = 60):
+    return store.ai_usage(minutes)
+
+
 @app.get("/api/orbit-logs")
 def orbit_logs():
     return store.orbit_logs()
@@ -1637,6 +1642,11 @@ def list_telemetry_v1():
     return store.telemetry()
 
 
+@app.get("/api/v1/ai-usage", tags=["Observability"], operation_id="getAiUsage")
+def ai_usage_v1(minutes: int = 60):
+    return store.ai_usage(minutes)
+
+
 @app.get("/api/v1/logs", tags=["Observability"], operation_id="listLogs")
 def list_logs_v1():
     return store.orbit_logs()

backend/app/mcp_server.py

@@ -22,6 +22,112 @@ def _json(value: Any) -> Any:
     return value.model_dump(mode="json") if hasattr(value, "model_dump") else value
 
 
+def _text(value: object, limit: int = 500) -> str:
+    """Keep MCP summaries bounded without losing a useful error description."""
+    return str(value or "")[:limit]
+
+
+def _pipeline_summary(value: Any) -> dict[str, Any]:
+    run = _json(value)
+    steps = run.get("step_results", []) if isinstance(run, dict) else []
+    supervisor = run.get("supervisor_results", []) if isinstance(run, dict) else []
+    files = [
+        file
+        for step in steps
+        if isinstance(step, dict)
+        for file in step.get("data_files", [])
+        if isinstance(file, dict)
+    ]
+    return {
+        key: run.get(key)
+        for key in (
+            "id",
+            "build_id",
+            "build_name",
+            "workflow_id",
+            "workflow_name",
+            "status",
+            "current_phase",
+            "current_step",
+            "execution_mode",
+            "execution_type",
+            "created_at",
+            "updated_at",
+            "finished_at",
+            "loop_limit",
+            "approval_score",
+            "supervisor_status",
+            "supervisor_error",
+        )
+    } | {
+        "step_count": len(steps),
+        "supervisor_result_count": len(supervisor),
+        "evidence_file_count": len(files),
+    }
+
+
+def _pipeline_detail(value: Any, detail: str, max_items: int) -> dict[str, Any]:
+    run = _json(value)
+    result = _pipeline_summary(run)
+    if detail == "summary":
+        return result
+    limit = max(1, min(max_items, 50))
+    if detail == "steps":
+        result["steps"] = [
+            {
+                key: _text(step.get(key))
+                if key in {"id", "phase", "name", "status", "error"}
+                else step.get(key)
+                for key in (
+                    "id",
+                    "phase",
+                    "name",
+                    "loop_index",
+                    "status",
+                    "started_at",
+                    "ended_at",
+                    "exit_code",
+                    "error",
+                )
+            }
+            for step in run.get("step_results", [])[:limit]
+            if isinstance(step, dict)
+        ]
+    elif detail == "supervision":
+        records = run.get("supervisor_results", []) or []
+        result["supervision"] = [
+            {
+                "iteration": record.get("iteration"),
+                "stage": record.get("stage"),
+                "status": record.get("status"),
+                "recorded_at": record.get("recorded_at"),
+                "error": _text(record.get("error")),
+                "score": (record.get("response") or {}).get("evaluation", {}).get("score")
+                if isinstance(record.get("response"), dict)
+                else None,
+                "approval": (record.get("response") or {}).get("evaluation", {}).get("approval")
+                if isinstance(record.get("response"), dict)
+                else None,
+            }
+            for record in records[:limit]
+            if isinstance(record, dict)
+        ]
+    elif detail == "evidence":
+        files = [
+            file
+            for step in run.get("step_results", [])
+            if isinstance(step, dict)
+            for file in step.get("data_files", [])
+            if isinstance(file, dict)
+        ]
+        result["evidence"] = [
+            {key: file.get(key) for key in ("label", "filename", "relative_path", "size", "content_type")}
+            for file in files[:limit]
+        ]
+        result["evidence_truncated"] = len(files) > limit
+    return result
+
+
 def create_mcp_server(
     get_store: Callable[[], ConsoleStore], openapi: Callable[[], dict[str, Any]]
 ) -> FastMCP:
@@ -53,10 +159,14 @@ def openapi_schema() -> str:
     )
     def get_status() -> dict[str, Any]:
         store = get_store()
+        dashboard = store.dashboard()
         return {
             "health": {"status": "ok"},
-            "dashboard": _json(store.dashboard()),
-            "active_pipelines": _json(store.active_evaluations()),
+            "dashboard": {
+                "metrics": dashboard["metrics"],
+                "active_build_count": len(dashboard["active_builds"]),
+            },
+            "active_pipelines": [_pipeline_summary(run) for run in store.active_evaluations()[:20]],
             "docker": preflight_docker(),
         }
 
@@ -71,19 +181,25 @@ def get_project(project_id: str) -> dict[str, Any]:
         return _json(get_store().build(project_id))
 
     @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, idempotentHint=True, openWorldHint=False))
-    def list_pipelines(project_id: str | None = None, status: str | None = None) -> list[dict[str, Any]]:
-        """List retained pipeline runs, optionally filtered by project or status."""
+    def list_pipelines(
+        project_id: str | None = None, status: str | None = None, limit: int = 20
+    ) -> list[dict[str, Any]]:
+        """List compact pipeline summaries; use get_pipeline for a bounded detail section."""
         pipelines = get_store().runs()
         if project_id:
             pipelines = [pipeline for pipeline in pipelines if pipeline.build_id == project_id]
         if status:
             pipelines = [pipeline for pipeline in pipelines if pipeline.status == status]
-        return [_json(pipeline) for pipeline in pipelines]
+        return [_pipeline_summary(pipeline) for pipeline in pipelines[: max(1, min(limit, 50))]]
 
     @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, idempotentHint=True, openWorldHint=False))
-    def get_pipeline(pipeline_id: str) -> dict[str, Any]:
-        """Get a pipeline's current state, phases, evidence, and supervisor results."""
-        return _json(get_store().run(pipeline_id))
+    def get_pipeline(
+        pipeline_id: str,
+        detail: Literal["summary", "steps", "supervision", "evidence"] = "summary",
+        max_items: int = 20,
+    ) -> dict[str, Any]:
+        """Get one bounded pipeline section; summary is the default and never returns raw logs or HTML."""
+        return _pipeline_detail(get_store().run(pipeline_id), detail, max_items)
 
     @mcp.tool(
         description="Start a project pipeline. Use test mode for a one-off safe validation run.",

backend/app/providers.py

@@ -8,16 +8,20 @@
 import json
 import os
 import subprocess
+import time
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Callable
 
 import requests
+from opentelemetry import trace
+from opentelemetry.trace import Status, StatusCode
 
 
 @dataclass(frozen=True)
 class ModelSettings:
+    profile_name: str = ""
     provider: str = "azure-openai"
     model: str = ""
     endpoint: str = ""
@@ -27,20 +31,76 @@ class ModelSettings:
 
 
 class AzureOpenAIProvider:
+    @staticmethod
+    def _post(settings: ModelSettings, key: str, payload: dict[str, Any]) -> dict[str, Any]:
+        started = time.perf_counter()
+        with trace.get_tracer("orbit.providers").start_as_current_span("gen_ai.azure.responses") as span:
+            input_value = payload.get("input", "")
+            input_chars = (
+                len(input_value)
+                if isinstance(input_value, str)
+                else len(json.dumps(input_value, ensure_ascii=False))
+            )
+            tools = payload.get("tools")
+            span.set_attributes(
+                {
+                    "gen_ai.system": "azure_openai",
+                    "gen_ai.request.profile": settings.profile_name or settings.model,
+                    "gen_ai.request.model": settings.model,
+                    "gen_ai.request.mode": "tool_result" if payload.get("previous_response_id") else "prompt",
+                    "gen_ai.request.input_chars": input_chars,
+                    "gen_ai.request.tool_count": len(tools) if isinstance(tools, list) else 0,
+                }
+            )
+            try:
+                response = requests.post(
+                    f"{settings.endpoint.rstrip('/')}/responses",
+                    headers={"api-key": key, "Content-Type": "application/json"},
+                    json=payload,
+                    timeout=180,
+                )
+                span.set_attribute("http.response.status_code", response.status_code)
+                for name in (
+                    "x-ratelimit-limit-requests",
+                    "x-ratelimit-remaining-requests",
+                    "x-ratelimit-limit-tokens",
+                    "x-ratelimit-remaining-tokens",
+                    "retry-after",
+                ):
+                    if response.headers.get(name):
+                        span.set_attribute(f"azure.{name.replace('-', '_')}", response.headers[name])
+                response.raise_for_status()
+                body = response.json()
+                usage = body.get("usage", {}) if isinstance(body, dict) else {}
+                for source, target in (
+                    ("input_tokens", "gen_ai.usage.input_tokens"),
+                    ("output_tokens", "gen_ai.usage.output_tokens"),
+                    ("total_tokens", "gen_ai.usage.total_tokens"),
+                ):
+                    if isinstance(usage.get(source), int):
+                        span.set_attribute(target, usage[source])
+                output = body.get("output_text", "") if isinstance(body, dict) else ""
+                if isinstance(output, str):
+                    span.set_attribute("gen_ai.response.output_chars", len(output))
+                return body
+            except requests.RequestException as error:
+                span.set_attribute("error.type", type(error).__name__)
+                span.set_attribute("error.message", str(error)[:500])
+                span.record_exception(error)
+                span.set_status(Status(StatusCode.ERROR))
+                raise
+            finally:
+                span.set_attribute(
+                    "gen_ai.request.duration_ms", round((time.perf_counter() - started) * 1000)
+                )
+
     def complete(self, settings: ModelSettings, prompt: str) -> str:
         key = os.environ.get(settings.secret_env, "")
         if not key or not settings.endpoint or not settings.model:
             raise RuntimeError(
                 "Azure OpenAI provider requires endpoint, model, and configured secret environment variable."
             )
-        response = requests.post(
-            f"{settings.endpoint.rstrip('/')}/responses",
-            headers={"api-key": key, "Content-Type": "application/json"},
-            json={"model": settings.model, "input": prompt},
-            timeout=180,
-        )
-        response.raise_for_status()
-        payload = response.json()
+        payload = self._post(settings, key, {"model": settings.model, "input": prompt})
         output_text = payload.get("output_text")
         if isinstance(output_text, str) and output_text:
             return output_text
@@ -78,14 +138,7 @@ def complete_with_tools(
             payload: dict[str, Any] = {"model": settings.model, "input": input_value, "tools": definitions}
             if response_id:
                 payload["previous_response_id"] = response_id
-            response = requests.post(
-                f"{settings.endpoint.rstrip('/')}/responses",
-                headers={"api-key": key, "Content-Type": "application/json"},
-                json=payload,
-                timeout=180,
-            )
-            response.raise_for_status()
-            body = response.json()
+            body = self._post(settings, key, payload)
             response_id = body.get("id")
             calls = [
                 item