diff --git a/agent/dashboard/routes.py b/agent/dashboard/routes.py index ba4f9f19..c9717c6e 100644 --- a/agent/dashboard/routes.py +++ b/agent/dashboard/routes.py @@ -141,6 +141,12 @@ from .thread_api import ( send_dashboard_message, stream_dashboard_thread, ) +from .user_credentials import ( + CurrentsCredentialsUpdate, + connect_currents, + disconnect_currents, + get_currents_status, +) from .user_mappings import ( delete_mapping, get_mapping, @@ -399,6 +405,31 @@ async def get_my_mapping( return mapping or {} +@router.get("/my-credentials/currents") +async def get_my_currents_status( + session: dict[str, Any] = _SESSION_DEP, +) -> dict[str, Any]: + status = await get_currents_status(session["sub"]) + return status.get("currents", {"connected": False}) + + +@router.put("/my-credentials/currents") +async def connect_my_currents( + update: CurrentsCredentialsUpdate, + session: dict[str, Any] = _SESSION_DEP, +) -> dict[str, Any]: + status = await connect_currents(session["sub"], update) + return status.get("currents", {"connected": False}) + + +@router.delete("/my-credentials/currents") +async def disconnect_my_currents( + session: dict[str, Any] = _SESSION_DEP, +) -> dict[str, Any]: + status = await disconnect_currents(session["sub"]) + return status.get("currents", {"connected": False}) + + @router.get("/slack/login") async def slack_login( _session: dict[str, Any] = _SESSION_DEP, diff --git a/agent/dashboard/user_credentials.py b/agent/dashboard/user_credentials.py new file mode 100644 index 00000000..e4940839 --- /dev/null +++ b/agent/dashboard/user_credentials.py @@ -0,0 +1,109 @@ +"""Per-user third-party service credentials (Currents.dev). + +Credentials are encrypted at rest with :mod:`agent.encryption` and stored in a +dedicated LangGraph Store namespace, keyed by the user's GitHub login. The +sandbox never holds these keys — they feed server-side read-only tools. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from typing import Any + +from langgraph_sdk import get_client +from pydantic import BaseModel, field_validator + +from ..encryption import decrypt_token, encrypt_token + +logger = logging.getLogger(__name__) + +USER_CREDENTIALS_NAMESPACE: list[str] = ["user_credentials"] +CURRENTS_KEY = "currents" + +CURRENTS_API_BASE = "https://api.currents.dev/v1" + + +def _client(): + return get_client() + + +def _last4(value: str) -> str: + return value[-4:] if len(value) >= 4 else value + + +async def _get_provider(login: str, key: str) -> dict[str, Any] | None: + try: + item = await _client().store.get_item([*USER_CREDENTIALS_NAMESPACE, login], key) + except Exception as e: # noqa: BLE001 + logger.debug("user credentials lookup failed for %s/%s: %s", login, key, e) + return None + if item is None: + return None + value = item.get("value") if isinstance(item, dict) else getattr(item, "value", None) + return value if isinstance(value, dict) else None + + +async def _put_provider(login: str, key: str, value: dict[str, Any]) -> None: + await _client().store.put_item([*USER_CREDENTIALS_NAMESPACE, login], key, value) + + +async def _delete_provider(login: str, key: str) -> None: + try: + await _client().store.delete_item([*USER_CREDENTIALS_NAMESPACE, login], key) + except Exception as e: # noqa: BLE001 + logger.debug("user credentials delete failed for %s/%s: %s", login, key, e) + + +class CurrentsCredentialsUpdate(BaseModel): + """Connect Currents.dev with an organization API key.""" + + api_key: str + + @field_validator("api_key") + @classmethod + def _require_non_empty(cls, v: object) -> str: + if not isinstance(v, str) or not v.strip(): + raise ValueError("api_key must be a non-empty string") + return v.strip() + + +async def get_currents_status(login: str) -> dict[str, Any]: + """Return a redacted, dashboard-safe view of the user's Currents key.""" + currents = await _get_provider(login, CURRENTS_KEY) + return { + "currents": { + "connected": True, + "api_key_last4": currents.get("api_key_last4", ""), + "updated_at": currents.get("updated_at"), + } + if currents + else {"connected": False}, + } + + +async def connect_currents(login: str, update: CurrentsCredentialsUpdate) -> dict[str, Any]: + await _put_provider( + login, + CURRENTS_KEY, + { + "encrypted_api_key": encrypt_token(update.api_key), + "api_key_last4": _last4(update.api_key), + "updated_at": datetime.now(UTC).isoformat(), + }, + ) + return await get_currents_status(login) + + +async def disconnect_currents(login: str) -> dict[str, Any]: + await _delete_provider(login, CURRENTS_KEY) + return await get_currents_status(login) + + +async def get_currents_api_key(login: str) -> str | None: + """Return the decrypted Currents API key, or ``None`` when not connected.""" + currents = await _get_provider(login, CURRENTS_KEY) + if not isinstance(currents, dict): + return None + api_key = decrypt_token(currents.get("encrypted_api_key", "")) + return api_key or None diff --git a/agent/integrations/currents_tools.py b/agent/integrations/currents_tools.py new file mode 100644 index 00000000..5aed9189 --- /dev/null +++ b/agent/integrations/currents_tools.py @@ -0,0 +1,178 @@ +"""Server-side, read-only Currents.dev tools for e2e test investigation. + +Credentials are per-user (encrypted at rest in the user-credentials Store +namespace). The tools run in the LangGraph server process and call the +Currents REST API directly — the sandbox never holds a Currents key. + +The surface is intentionally read-only: fetch runs, instances, test results, +and list projects so the agent can dig into e2e test failures including +screenshots and DOM snapshots. +""" + +from __future__ import annotations + +import logging +from typing import Any + +import httpx +from langchain_core.tools import BaseTool, StructuredTool + +from ..dashboard.user_credentials import CURRENTS_API_BASE, get_currents_api_key + +logger = logging.getLogger(__name__) + +_TIMEOUT = httpx.Timeout(30.0, connect=10.0) + + +def _headers(api_key: str) -> dict[str, str]: + return { + "Authorization": f"Bearer {api_key}", + "Accept": "application/json", + "Accept-Encoding": "gzip", + } + + +async def _get(path: str, api_key: str, **params: Any) -> dict[str, Any]: + url = f"{CURRENTS_API_BASE}{path}" + async with httpx.AsyncClient(timeout=_TIMEOUT) as client: + resp = await client.get(url, headers=_headers(api_key), params=params) + resp.raise_for_status() + return resp.json() + + +def _make_tools(api_key: str) -> list[BaseTool]: + async def currents_list_projects( + limit: int = 10, + starting_after: str | None = None, + ) -> dict[str, Any]: + """List Currents.dev projects for your organization. + + Args: + limit: Maximum number of items to return (default 10, max 50). + starting_after: Cursor for pagination. + + Returns: + Dictionary with project list, or an error message. + """ + try: + params: dict[str, Any] = {"limit": max(1, min(limit, 50))} + if starting_after: + params["starting_after"] = starting_after + return await _get("/projects", api_key, **params) + except Exception as e: # noqa: BLE001 + logger.warning("currents_list_projects failed", exc_info=True) + return {"success": False, "error": f"{type(e).__name__}: {e}"} + + async def currents_get_run(run_id: str) -> dict[str, Any]: + """Get a single Currents.dev test run by ID with full details. + + Use this to inspect a specific e2e test run including specs, + screenshots, video URLs, and test stats. + + Args: + run_id: The Currents run ID (e.g. "run_abc123"). + + Returns: + Dictionary with the run details, or an error message. + """ + try: + return await _get(f"/runs/{run_id}", api_key) + except Exception as e: # noqa: BLE001 + logger.warning("currents_get_run failed", exc_info=True) + return {"success": False, "error": f"{type(e).__name__}: {e}"} + + async def currents_find_run( + project_id: str, + ci_build_id: str | None = None, + branch: str | None = None, + ) -> dict[str, Any]: + """Find the most recent completed Currents.dev run matching criteria. + + Args: + project_id: The Currents project ID (e.g. "proj_abc123"). + ci_build_id: Optional CI build ID to find an exact run. + branch: Optional branch name or prefix (append * for prefix match). + + Returns: + Dictionary with the run details, or an error message. + """ + try: + params: dict[str, Any] = {"projectId": project_id} + if ci_build_id: + params["ciBuildId"] = ci_build_id + if branch: + params["branch"] = branch + return await _get("/runs/find", api_key, **params) + except Exception as e: # noqa: BLE001 + logger.warning("currents_find_run failed", exc_info=True) + return {"success": False, "error": f"{type(e).__name__}: {e}"} + + async def currents_list_project_runs( + project_id: str, + limit: int = 10, + status: str | None = None, + branch: str | None = None, + starting_after: str | None = None, + ending_before: str | None = None, + ) -> dict[str, Any]: + """List runs for a Currents.dev project with optional filters. + + Args: + project_id: The Currents project ID. + limit: Maximum number of runs to return (default 10, max 50). + status: Optional status filter: PASSED, FAILED, RUNNING, FAILING. + branch: Optional branch filter (append * for prefix match). + starting_after: Cursor for pagination (next page). + ending_before: Cursor for pagination (previous page). + + Returns: + Dictionary with a list of runs, or an error message. + """ + try: + params: dict[str, Any] = {"limit": max(1, min(limit, 50))} + if status: + params["status"] = status + if branch: + params["branches[]"] = branch + if starting_after: + params["starting_after"] = starting_after + if ending_before: + params["ending_before"] = ending_before + return await _get(f"/projects/{project_id}/runs", api_key, **params) + except Exception as e: # noqa: BLE001 + logger.warning("currents_list_project_runs failed", exc_info=True) + return {"success": False, "error": f"{type(e).__name__}: {e}"} + + async def currents_get_instance(instance_id: str) -> dict[str, Any]: + """Get a single Currents.dev spec file execution instance by ID. + + An instance represents one spec file's execution within a run, + including detailed test results, errors, and attempt history. + + Args: + instance_id: The Currents instance ID (e.g. "inst_abc123"). + + Returns: + Dictionary with the instance details, or an error message. + """ + try: + return await _get(f"/instances/{instance_id}", api_key) + except Exception as e: # noqa: BLE001 + logger.warning("currents_get_instance failed", exc_info=True) + return {"success": False, "error": f"{type(e).__name__}: {e}"} + + return [ + StructuredTool.from_function(coroutine=currents_list_projects), + StructuredTool.from_function(coroutine=currents_get_run), + StructuredTool.from_function(coroutine=currents_find_run), + StructuredTool.from_function(coroutine=currents_list_project_runs), + StructuredTool.from_function(coroutine=currents_get_instance), + ] + + +async def load_currents_tools(login: str) -> list[BaseTool]: + """Return read-only Currents tools when the user has connected Currents.""" + api_key = await get_currents_api_key(login) + if not api_key: + return [] + return _make_tools(api_key) diff --git a/agent/server.py b/agent/server.py index eb3beb66..95b13798 100644 --- a/agent/server.py +++ b/agent/server.py @@ -43,6 +43,7 @@ from .dashboard.agent_usage import record_agent_thread_usage from .dashboard.options import DEFAULT_MODEL_ID, SUPPORTED_MODEL_IDS, model_supports_effort from .dashboard.team_settings import get_team_default_model_pair, get_team_default_repo from .dashboard.user_mappings import email_for_login +from .integrations.currents_tools import load_currents_tools from .integrations.datadog_mcp import load_datadog_tools from .integrations.langsmith import _configure_github_proxy from .integrations.langsmith_tools import load_langsmith_tools @@ -679,6 +680,14 @@ async def get_agent(config: RunnableConfig) -> Pregel: await _observability_authorized(config, profile_login) ) + currents_tools: list[Any] = [] + if profile_login: + try: + currents_tools = await load_currents_tools(profile_login) + except Exception: + logger.warning("Failed to load Currents tools", exc_info=True) + currents_tools = [] + logger.info("Returning agent with sandbox for thread %s", thread_id) main_model = make_model(model_id, **model_kwargs) subagent_model = make_model(subagent_model_id, **subagent_model_kwargs) @@ -710,6 +719,7 @@ async def get_agent(config: RunnableConfig) -> Pregel: slack_read_thread_messages, slack_thread_reply, *observability_tools, + *currents_tools, ], subagents=[_general_purpose_subagent(subagent_model)], backend=backend_factory, diff --git a/tests/test_currents_tools.py b/tests/test_currents_tools.py new file mode 100644 index 00000000..02f6e415 --- /dev/null +++ b/tests/test_currents_tools.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import pytest + +from agent.integrations import currents_tools + + +@pytest.mark.asyncio +async def test_load_currents_tools_empty_when_not_connected() -> None: + with patch.object(currents_tools, "get_currents_api_key", AsyncMock(return_value=None)): + assert await currents_tools.load_currents_tools("alice") == [] + + +@pytest.mark.asyncio +async def test_load_currents_tools_names() -> None: + with patch.object(currents_tools, "get_currents_api_key", AsyncMock(return_value="k")): + tools = await currents_tools.load_currents_tools("alice") + assert {t.name for t in tools} == { + "currents_list_projects", + "currents_get_run", + "currents_find_run", + "currents_list_project_runs", + "currents_get_instance", + } + + +@pytest.mark.asyncio +async def test_currents_get_run_success() -> None: + payload = {"status": "OK", "data": {"runId": "run_123", "status": "failed"}} + with patch.object(currents_tools, "_get", AsyncMock(return_value=payload)): + tools = currents_tools._make_tools("test-key") + get_run = next(t for t in tools if t.name == "currents_get_run") + result = await get_run.ainvoke({"run_id": "run_123"}) + assert result == payload + + +@pytest.mark.asyncio +async def test_currents_get_run_error() -> None: + with patch.object(currents_tools, "_get", AsyncMock(side_effect=RuntimeError("boom"))): + tools = currents_tools._make_tools("bad-key") + get_run = next(t for t in tools if t.name == "currents_get_run") + result = await get_run.ainvoke({"run_id": "run_123"}) + assert result["success"] is False + assert "boom" in result["error"] + + +@pytest.mark.asyncio +async def test_currents_list_projects_caps_limit() -> None: + captured: dict[str, object] = {} + + async def fake_get(path: str, api_key: str, **params): + captured["path"] = path + captured["limit"] = params.get("limit") + return {"status": "OK", "data": []} + + with patch.object(currents_tools, "_get", side_effect=fake_get): + tools = currents_tools._make_tools("k") + list_projects = next(t for t in tools if t.name == "currents_list_projects") + result = await list_projects.ainvoke({"limit": 9999}) + assert result["status"] == "OK" + assert captured["limit"] == 50 + assert captured["path"] == "/projects" + + +@pytest.mark.asyncio +async def test_currents_find_run_passes_params() -> None: + captured: dict[str, object] = {} + + async def fake_get(path: str, api_key: str, **params): + captured["path"] = path + captured.update(params) + return {"status": "OK", "data": {"runId": "r1"}} + + with patch.object(currents_tools, "_get", side_effect=fake_get): + tools = currents_tools._make_tools("k") + find_run = next(t for t in tools if t.name == "currents_find_run") + result = await find_run.ainvoke( + {"project_id": "proj_1", "ci_build_id": "build-42", "branch": "main"} + ) + assert result["status"] == "OK" + assert captured["path"] == "/runs/find" + assert captured["projectId"] == "proj_1" + assert captured["ciBuildId"] == "build-42" + assert captured["branch"] == "main" + + +@pytest.mark.asyncio +async def test_currents_list_project_runs_caps_limit() -> None: + captured: dict[str, object] = {} + + async def fake_get(path: str, api_key: str, **params): + captured["path"] = path + captured.update(params) + return {"status": "OK", "data": []} + + with patch.object(currents_tools, "_get", side_effect=fake_get): + tools = currents_tools._make_tools("k") + list_runs = next(t for t in tools if t.name == "currents_list_project_runs") + await list_runs.ainvoke( + { + "project_id": "proj_1", + "limit": 9999, + "status": "FAILED", + "branch": "main", + "starting_after": "cursor-abc", + } + ) + assert captured["path"] == "/projects/proj_1/runs" + assert captured["limit"] == 50 + assert captured["status"] == "FAILED" + assert captured["branches[]"] == "main" + assert captured["starting_after"] == "cursor-abc" + + +@pytest.mark.asyncio +async def test_currents_get_instance() -> None: + payload = {"status": "OK", "data": {"instanceId": "inst_1"}} + with patch.object(currents_tools, "_get", AsyncMock(return_value=payload)): + tools = currents_tools._make_tools("k") + get_instance = next(t for t in tools if t.name == "currents_get_instance") + result = await get_instance.ainvoke({"instance_id": "inst_1"}) + assert result == payload diff --git a/tests/test_user_credentials.py b/tests/test_user_credentials.py new file mode 100644 index 00000000..55b9cf32 --- /dev/null +++ b/tests/test_user_credentials.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from typing import Any + +import pytest +from cryptography.fernet import Fernet +from pydantic import ValidationError + +from agent.dashboard import user_credentials as uc +from agent.dashboard.user_credentials import CurrentsCredentialsUpdate + + +class _FakeStore: + def __init__(self) -> None: + self.items: dict[tuple[tuple[str, ...], str], dict[str, Any]] = {} + + async def get_item(self, namespace: list[str], key: str): + value = self.items.get((tuple(namespace), key)) + return {"value": value} if value is not None else None + + async def put_item(self, namespace: list[str], key: str, value: dict[str, Any]) -> None: + self.items[(tuple(namespace), key)] = value + + async def delete_item(self, namespace: list[str], key: str) -> None: + self.items.pop((tuple(namespace), key), None) + + +class _FakeClient: + def __init__(self, store: _FakeStore) -> None: + self.store = store + + +@pytest.fixture() +def fake_store(monkeypatch: pytest.MonkeyPatch) -> _FakeStore: + store = _FakeStore() + monkeypatch.setattr(uc, "_client", lambda: _FakeClient(store)) + monkeypatch.setenv("TOKEN_ENCRYPTION_KEY", Fernet.generate_key().decode()) + return store + + +class TestValidators: + def test_empty_key_rejected(self) -> None: + with pytest.raises(ValidationError): + CurrentsCredentialsUpdate(api_key="") + + def test_whitespace_key_rejected(self) -> None: + with pytest.raises(ValidationError): + CurrentsCredentialsUpdate(api_key=" ") + + def test_key_trimmed(self) -> None: + u = CurrentsCredentialsUpdate(api_key=" secret ") + assert u.api_key == "secret" + + +@pytest.mark.asyncio +async def test_currents_roundtrip_and_redaction(fake_store: _FakeStore) -> None: + status = await uc.connect_currents( + "alice", CurrentsCredentialsUpdate(api_key="secret-currents-key-1234") + ) + assert status["currents"]["connected"] is True + assert status["currents"]["api_key_last4"] == "1234" + + record = fake_store.items[(("user_credentials", "alice"), "currents")] + assert record["encrypted_api_key"] != "secret-currents-key-1234" + + api_key = await uc.get_currents_api_key("alice") + assert api_key == "secret-currents-key-1234" + + after = await uc.disconnect_currents("alice") + assert after["currents"]["connected"] is False + assert await uc.get_currents_api_key("alice") is None + + +@pytest.mark.asyncio +async def test_currents_isolation_between_users(fake_store: _FakeStore) -> None: + await uc.connect_currents("alice", CurrentsCredentialsUpdate(api_key="alice-key-abcd")) + await uc.connect_currents("bob", CurrentsCredentialsUpdate(api_key="bob-key-wxyz")) + + assert await uc.get_currents_api_key("alice") == "alice-key-abcd" + assert await uc.get_currents_api_key("bob") == "bob-key-wxyz" + + await uc.disconnect_currents("alice") + assert await uc.get_currents_api_key("alice") is None + assert await uc.get_currents_api_key("bob") == "bob-key-wxyz" + + +@pytest.mark.asyncio +async def test_currents_status_when_not_connected(fake_store: _FakeStore) -> None: + status = await uc.get_currents_status("nobody") + assert status["currents"]["connected"] is False + + +@pytest.mark.asyncio +async def test_get_currents_api_key_none_when_not_connected(fake_store: _FakeStore) -> None: + assert await uc.get_currents_api_key("nobody") is None diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index 05aa3d59..72d87f41 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -161,6 +161,16 @@ export interface LangSmithConnectBody { endpoint?: string | null } +export interface CurrentsCredentialStatus { + connected: boolean + api_key_last4?: string + updated_at?: string | null +} + +export interface CurrentsConnectBody { + api_key: string +} + export interface UserMapping { github_login: string work_email: string @@ -560,6 +570,17 @@ export const api = { request("/team-credentials/langsmith", { method: "DELETE", }), + getMyCurrentsStatus: () => + request("/my-credentials/currents"), + connectCurrents: (body: CurrentsConnectBody) => + request("/my-credentials/currents", { + method: "PUT", + body: JSON.stringify(body), + }), + disconnectCurrents: () => + request("/my-credentials/currents", { + method: "DELETE", + }), listEnabledReviewRepos: () => request<{ repos: Array }>("/enabled-review-repos"), setEnabledReviewRepo: (full_name: string, enabled: boolean) => diff --git a/ui/src/routes/my-settings.tsx b/ui/src/routes/my-settings.tsx index 1367bded..836eec65 100644 --- a/ui/src/routes/my-settings.tsx +++ b/ui/src/routes/my-settings.tsx @@ -1,11 +1,12 @@ import { Navigate, createFileRoute, useNavigate } from "@tanstack/react-router" -import { useQuery, useQueryClient } from "@tanstack/react-query" +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" import { useState } from "react" import { IoLogoSlack } from "react-icons/io5" -import type { SessionUser } from "@/lib/api" +import type { CurrentsConnectBody, SessionUser } from "@/lib/api" import { AppShell, SettingsRow, SettingsSection } from "@/components/AppShell" import { Button } from "@/components/ui/button" +import { Input } from "@/components/ui/input" import { Select, SelectContent, @@ -187,6 +188,83 @@ function NotificationsSection() { ) } +function CurrentsCredentialsSection() { + const qc = useQueryClient() + const creds = useQuery({ + queryKey: ["myCurrents"], + queryFn: api.getMyCurrentsStatus, + }) + const [error, setError] = useState(null) + const [apiKey, setApiKey] = useState("") + + const onSuccess = () => { + qc.invalidateQueries({ queryKey: ["myCurrents"] }) + setError(null) + setApiKey("") + } + const onError = (e: Error) => setError(e.message) + + const connect = useMutation({ + mutationFn: (body: CurrentsConnectBody) => api.connectCurrents(body), + onSuccess, + onError, + }) + const disconnect = useMutation({ + mutationFn: () => api.disconnectCurrents(), + onSuccess, + onError, + }) + + const connected = creds.data?.connected + + return ( + + disconnect.mutate()} + disabled={disconnect.isPending} + > + Disconnect + + ) : ( +
+ setApiKey(e.target.value)} + disabled={creds.isLoading} + /> + +
+ ) + } + /> + {error &&

{error}

} + + ) +} + function MySettingsPage() { const session = useSession() const qc = useQueryClient() @@ -217,7 +295,8 @@ function MySettingsPage() { } const firstModel = options.data?.models[0] - const fallbackModel = options.data?.default_agent_model ?? firstModel?.id ?? "" + const fallbackModel = + options.data?.default_agent_model ?? firstModel?.id ?? "" const fallbackEffort = options.data?.default_agent_reasoning_effort ?? firstModel?.default_effort ?? @@ -287,6 +366,8 @@ function MySettingsPage() { + +