"""HTTP app for the full-flow E2E (served as langgraph dev's http.app).
Mounts, on top of the REAL ``agent.api.app`` app:
- fake GitHub REST API (/fake-gh/...) the real open_pull_request hits this
- fake Slack API (/fake-slack/...) the real slack code hits this
- mock UIs (/mock/slack, /mock/github) what the user/Playwright sees
- control + compose (/control/*, /mock/slack/send) the test driver
Nothing here touches agent logic — it only stands in for the SaaS boundaries
and renders their state back as a user-facing UI.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import os
import sys
import time
from html import escape
from pathlib import Path
from typing import Any
from urllib.parse import quote
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import e2e_env # noqa: E402
import patches # noqa: E402
patches.apply()
import fakes # noqa: E402
import httpx # noqa: E402
from e2e_env import ( # noqa: E402
BASE_URL,
BOT_USER_ID,
DEMO_CHANNEL,
HUMAN_USER,
REPO_ROOT,
TEST_USERS,
)
from fastapi import HTTPException, Request # noqa: E402
from fastapi.responses import ( # noqa: E402
FileResponse,
HTMLResponse,
JSONResponse,
RedirectResponse,
Response,
)
# Slack-user directory the fake ``users.info`` resolves: the default sender used
# by the automated tests plus the named manual-test users.
_SLACK_USERS: dict[str, dict[str, str]] = {
HUMAN_USER: {"name": "devuser", "real_name": "Dev User", "email": "dev@example.com"},
**{
u["slack_id"]: {"name": u["login"], "real_name": u["name"], "email": u["email"]}
for u in TEST_USERS
},
}
from agent.api.app import app # noqa: E402
from agent.dashboard import routes as dashboard_routes # noqa: E402
from agent.dashboard.oauth import COOKIE_NAME, issue_session # noqa: E402
from agent.utils import github_checks, github_org_membership # noqa: E402
from agent.utils.thread_ids import generate_thread_id_from_slack_thread # noqa: E402
from agent.webhooks import common as webhook_common # noqa: E402
GITHUB_WEBHOOK_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]
SLACK_SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"]
STATIC_DIR = Path(__file__).parent / "static"
CURRENT_THREAD: dict[str, str | None] = {"channel": DEMO_CHANNEL, "thread_ts": None}
fakes.seed_bare_remote()
_real_dispatch_agent_run = webhook_common.dispatch_agent_run
_real_email_for_login = webhook_common.email_for_login
async def _fake_installation_token(*_args: object, **_kwargs: object) -> str:
return "dummy-installation-token"
async def _fake_installation_token_with_expiry(
*_args: object, **_kwargs: object
) -> tuple[str, None]:
return "dummy-installation-token", None
async def _fake_fetch_pr_metadata(pr_ref: Any, *, token: str) -> dict[str, Any] | None: # noqa: ARG001
pr = fakes.find_pull(pr_ref.number)
return _gh_pr_json(pr) if pr is not None else None
async def _fake_reviewer_token(*_args: object, **_kwargs: object) -> tuple[str, None]:
return "dummy-installation-token", None
async def _fake_reaction(*_args: object, **_kwargs: object) -> bool:
return True
async def _fake_started_comment(*_args: object, **_kwargs: object) -> None:
return None
async def _fake_active_org_member(username: str, org: str) -> bool:
return bool(username and org)
async def _record_email_mapping_lookup(login: str) -> str | None:
fakes.EMAIL_MAPPING_LOOKUPS.append(login)
return await _real_email_for_login(login)
async def _record_review_dispatch(
thread_id: str,
prompt: str,
configurable: dict[str, Any],
*,
source: str,
assistant_id: str = "agent",
**kwargs: object,
) -> dict[str, str]:
if assistant_id != "reviewer":
return await _real_dispatch_agent_run(
thread_id,
prompt,
configurable,
source=source,
assistant_id=assistant_id,
**kwargs,
)
run_id = f"review-run-{len(fakes.REVIEW_DISPATCHES) + 1}"
fakes.REVIEW_DISPATCHES.append(
{
"thread_id": thread_id,
"prompt": prompt,
"configurable": configurable,
"source": source,
"assistant_id": assistant_id,
"run_id": run_id,
}
)
return {"run_id": run_id}
async def _fake_installations_and_repos(
_login: str,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
return (
[{"id": 1, "account": {"login": e2e_env.OWNER, "type": "Organization"}}],
[{"full_name": f"{e2e_env.OWNER}/{e2e_env.REPO}", "private": False}],
)
webhook_common.get_github_app_installation_token = _fake_installation_token
webhook_common.get_github_app_installation_token_with_expiry = _fake_installation_token_with_expiry
webhook_common.fetch_github_pr_metadata = _fake_fetch_pr_metadata
webhook_common._reviewer_token_for_repo = _fake_reviewer_token
webhook_common.react_to_github_comment = _fake_reaction
webhook_common.post_review_started_comment = _fake_started_comment
webhook_common.dispatch_agent_run = _record_review_dispatch
webhook_common.email_for_login = _record_email_mapping_lookup
github_org_membership.is_user_active_org_member = _fake_active_org_member
webhook_common.is_user_active_org_member = _fake_active_org_member
dashboard_routes._fetch_user_installations_and_repos = _fake_installations_and_repos
github_checks._GITHUB_API_BASE = e2e_env.FAKE_GITHUB_API
# --- control + Slack compose (the test driver) -----------------------------
@app.post("/control/reset")
async def control_reset() -> JSONResponse:
fakes.reset()
CURRENT_THREAD["thread_ts"] = None
return JSONResponse({"ok": True})
@app.post("/control/review-pr")
async def control_review_pr() -> JSONResponse:
fakes.reset()
pr = fakes.create_review_pull(e2e_env.OWNER, e2e_env.REPO)
return JSONResponse(_gh_pr_json(pr))
@app.get("/control/review-dispatches")
async def control_review_dispatches() -> JSONResponse:
return JSONResponse(fakes.REVIEW_DISPATCHES)
@app.get("/control/email-mapping-lookups")
async def control_email_mapping_lookups() -> JSONResponse:
return JSONResponse(fakes.EMAIL_MAPPING_LOOKUPS)
@app.get("/control/check-runs")
async def control_check_runs() -> JSONResponse:
return JSONResponse(fakes.CHECK_RUNS)
@app.post("/control/review-check/evaluate")
async def control_review_check_evaluate(request: Request) -> JSONResponse:
body = await request.json()
outcome = body.get("outcome")
if not isinstance(outcome, dict):
raise HTTPException(400, "outcome must be an object")
head_sha = str(body.get("head_sha") or f"evaluation-{len(fakes.CHECK_RUNS) + 1}")
check_run_id = await github_checks.create_review_check_run(
owner=e2e_env.OWNER,
repo=e2e_env.REPO,
head_sha=head_sha,
token="dummy-installation-token",
)
if check_run_id is None:
raise HTTPException(500, "failed to create check run")
conclusion, title, summary = github_checks.review_check_conclusion(outcome)
completed = await github_checks.complete_review_check_run(
owner=e2e_env.OWNER,
repo=e2e_env.REPO,
check_run_id=check_run_id,
token="dummy-installation-token",
conclusion=conclusion,
title=title,
summary=summary,
)
if not completed:
raise HTTPException(500, "failed to complete check run")
return JSONResponse({"id": check_run_id, "conclusion": conclusion, "title": title})
@app.get("/control/state")
async def control_state() -> JSONResponse:
return JSONResponse(
{"channel": CURRENT_THREAD["channel"], "thread_ts": CURRENT_THREAD["thread_ts"]}
)
@app.post("/mock/slack/send")
async def slack_send(request: Request) -> JSONResponse:
"""Simulate a user posting in Slack: store the message, then deliver the
signed Events-API webhook to the real /webhooks/slack route."""
form = await request.json()
text = str(form.get("text", ""))
mention_bot = bool(form.get("mention_bot", True))
# Sender defaults to the first test user (Alice) — the canonical owner the
# automated tests log in as; the mock UI passes the chosen test user.
user_id = str(form.get("user") or TEST_USERS[0]["slack_id"])
channel = DEMO_CHANNEL
ts = fakes.new_thread_ts()
CURRENT_THREAD["thread_ts"] = ts
fakes.add_slack_message(channel, ts, user=user_id, text=text, is_bot=False)
payload = {
"type": "event_callback",
"event_id": f"Ev{ts}",
"authorizations": [{"user_id": BOT_USER_ID}],
"event": {
"type": "app_mention" if mention_bot else "message",
"channel": channel,
"user": user_id,
"text": text,
"ts": ts,
"thread_ts": ts,
},
}
raw = json.dumps(payload).encode()
req_ts = str(int(time.time()))
base = f"v0:{req_ts}:{raw.decode()}".encode()
sig = "v0=" + hmac.new(SLACK_SIGNING_SECRET.encode(), base, hashlib.sha256).hexdigest()
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://harness") as client:
resp = await client.post(
"/webhooks/slack",
content=raw,
headers={
"X-Slack-Signature": sig,
"X-Slack-Request-Timestamp": req_ts,
"Content-Type": "application/json",
},
)
return JSONResponse(
{
"thread_ts": ts,
"thread_id": generate_thread_id_from_slack_thread(channel, ts),
"webhook_status": resp.status_code,
"webhook": resp.json(),
}
)
@app.post("/control/login")
async def control_login(request: Request) -> JSONResponse:
"""Simulate a signed-in dashboard user by minting the real session cookie."""
form = await request.json()
login = str(form.get("login", "dev-user"))
email = str(form.get("email", "dev@example.com"))
token = issue_session(login=login, email=email, avatar_url=None)
resp = JSONResponse({"ok": True, "login": login, "email": email})
resp.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax", secure=False, path="/")
return resp
@app.get("/control/login")
async def control_login_get(login: str = "", email: str = "", next_url: str = "") -> Response:
"""Browser login. With no ``login``, render a dropdown of the test users;
with ``?login=`` (email resolved from the registry, or pass ``&email=``),
mint the session cookie and redirect into the dashboard. Use a separate
browser/profile per user — each has its own cookie jar."""
# Land on the dashboard origin (DASHBOARD_BASE_URL — the Vite HMR server in
# dev:mock), not this harness, so the cookie + the hot-reloading UI line up.
ui = os.environ.get("DASHBOARD_BASE_URL", "").rstrip("/")
dest = next_url or (f"{ui}/agents" if ui else "/agents")
if not login:
options = "".join(f'' for u in TEST_USERS)
return HTMLResponse(
f"""Mock login
Sign in (mock)
Tip: use a separate browser or profile per
user so their sessions don't overwrite each other.
"""
)
if not email:
match = next((u for u in TEST_USERS if u["login"] == login), None)
email = match["email"] if match else f"{login}@example.com"
token = issue_session(login=login, email=email, avatar_url=None)
resp = RedirectResponse(url=dest, status_code=303)
resp.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax", secure=False, path="/")
return resp
@app.get("/dashboard/api/auth/login")
async def mock_github_login(redirect_to: str = "") -> Response:
"""E2E stand-in for the dashboard OAuth start route.
The real route would redirect to github.com. Keep the dashboard-facing URL
intact, then hand off to the fake GitHub simulator so Playwright exercises a
browser login flow instead of test code pre-minting a session cookie.
"""
ui = os.environ.get("DASHBOARD_BASE_URL", "").rstrip("/")
dest = redirect_to or (f"{ui}/agents" if ui else "/agents")
return RedirectResponse(f"/fake-gh/login/oauth/authorize?redirect_to={quote(dest)}", 302)
@app.get("/fake-gh/login/oauth/authorize")
async def fake_github_authorize(redirect_to: str = "", login: str = "") -> Response:
"""Fake GitHub OAuth consent/login page for dashboard e2e tests."""
ui = os.environ.get("DASHBOARD_BASE_URL", "").rstrip("/")
dest = redirect_to or (f"{ui}/agents" if ui else "/agents")
if not login:
options = "".join(
f'"
for u in TEST_USERS
)
return HTMLResponse(
f"""GitHub · Authorize open-swe
Authorize open-swe
Pick a fake GitHub account to continue.
"""
)
match = next((u for u in TEST_USERS if u["login"] == login), None)
email = match["email"] if match else f"{login}@example.com"
token = issue_session(login=login, email=email, avatar_url=None)
resp = RedirectResponse(url=dest, status_code=303)
resp.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax", secure=False, path="/")
return resp
# The real dashboard registered /dashboard/api/auth/login first (via
# include_router), so Starlette would match it before ours. Move ours to the
# front of the table so the mock picker shadows the real OAuth redirect.
for _i, _route in enumerate(app.router.routes):
if getattr(_route, "endpoint", None) is mock_github_login:
app.router.routes.insert(0, app.router.routes.pop(_i))
break
@app.post("/control/logout")
async def control_logout() -> JSONResponse:
resp = JSONResponse({"ok": True})
resp.delete_cookie(COOKIE_NAME, path="/")
return resp
# --- serve the REAL built ui/ SPA, same-origin so the session cookie works ----
# The "Open in Web" link (DASHBOARD_BASE_URL/agents/{id}) lands on the real app;
# it calls /dashboard/api/* (same origin) and streams via the dashboard proxy.
UI_PUBLIC = REPO_ROOT / "ui" / ".output" / "public"
_ASSETS_ROOT = (UI_PUBLIC / "assets").resolve()
def _ui_file(name: str) -> FileResponse:
path = UI_PUBLIC / name
if not path.is_file():
raise HTTPException(404, f"{name} not built — run `bun run build` in ui/")
return FileResponse(path)
@app.get("/assets/{asset_path:path}")
async def ui_asset(asset_path: str) -> FileResponse:
# Explicit route, not app.mount(StaticFiles): LangGraph's custom-app loader
# serves APIRoutes but drops sub-app Mounts, so a mount 404s under it.
target = (_ASSETS_ROOT / asset_path).resolve()
if not str(target).startswith(str(_ASSETS_ROOT)) or not target.is_file():
raise HTTPException(404, "asset not found")
return FileResponse(target)
@app.get("/_shell.html", response_class=HTMLResponse)
async def ui_shell() -> FileResponse:
return _ui_file("_shell.html")
@app.get("/manifest.webmanifest")
async def ui_manifest() -> FileResponse:
return _ui_file("manifest.webmanifest")
@app.get("/favicon.png")
async def ui_favicon() -> FileResponse:
return _ui_file("favicon.png")
@app.get("/apple-touch-icon.png")
async def ui_apple_icon() -> FileResponse:
return _ui_file("apple-touch-icon.png")
@app.get("/logo-mark.png")
async def ui_logo_mark() -> FileResponse:
return _ui_file("logo-mark.png")
# Client routes used by the handoff tests: serve the SPA shell; the client
# router boots at the current URL. Kept explicit (no catch-all) so LangGraph's
# own root routes — which the dashboard proxy calls server-side — are untouched.
@app.get("/agents/{thread_id}", response_class=HTMLResponse)
async def ui_agents_thread(thread_id: str) -> FileResponse: # noqa: ARG001
return _ui_file("_shell.html")
@app.get("/agents/{thread_id}/plan", response_class=HTMLResponse)
async def ui_agents_plan(thread_id: str) -> FileResponse: # noqa: ARG001
return _ui_file("_shell.html")
@app.get("/review", response_class=HTMLResponse)
async def ui_review() -> FileResponse:
return _ui_file("_shell.html")
@app.get("/review/repositories/{owner}", response_class=HTMLResponse)
async def ui_review_repositories(owner: str) -> FileResponse: # noqa: ARG001
return _ui_file("_shell.html")
@app.get("/login", response_class=HTMLResponse)
async def ui_login() -> FileResponse:
return _ui_file("_shell.html")
@app.get("/mock/users")
async def mock_users() -> JSONResponse:
"""The named test users that drive the Slack sender + login dropdowns."""
return JSONResponse(TEST_USERS)
@app.get("/mock/slack/messages")
async def slack_messages() -> JSONResponse:
msgs = fakes.slack_messages(CURRENT_THREAD["channel"])
return JSONResponse(
[
{
"user": m["user"],
"text": m["text"],
"is_bot": m["is_bot"],
"ts": m["ts"],
"thread_ts": m["thread_ts"],
}
for m in msgs
]
)
# --- mock UIs --------------------------------------------------------------
@app.get("/mock/slack", response_class=HTMLResponse)
async def mock_slack_page() -> str:
return (STATIC_DIR / "slack.html").read_text()
@app.get("/mock/github", response_class=HTMLResponse)
async def mock_github_page() -> str:
return (STATIC_DIR / "github.html").read_text()
def _pr_html_url(pr: dict[str, Any]) -> str:
return f"{BASE_URL}/mock/github/{pr['owner']}/{pr['repo']}/pull/{pr['number']}"
@app.get("/mock/github/data")
async def mock_github_data() -> JSONResponse:
return JSONResponse(
[
{
"number": p["number"],
"title": p["title"],
"head": p["head"],
"base": p["base"],
"state": p["state"],
"draft": p["draft"],
"author": p["author"],
"body": p["body"],
"files": p["files"],
"url": _pr_html_url(p),
}
for p in fakes.PULLS
]
)
@app.get("/mock/github/{owner}/{repo}/pull/{number}", response_class=HTMLResponse)
async def mock_github_pr(owner: str, repo: str, number: int) -> HTMLResponse: # noqa: ARG001
pr = fakes.find_pull(number)
if pr is None:
return HTMLResponse(f"