"""Thin Gradio surface over the authoritative JourneyTest HTTP/service runtime.""" from contextlib import asynccontextmanager import json, os, time, urllib.error, urllib.request import gradio as gr from fastapi import FastAPI, Request from fastapi.responses import Response import httpx, uvicorn API = os.environ.get("JOURNEYTEST_API_BASE", "http://127.0.0.1:7861") def request(path, method="GET", body=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(API + path, data=data, method=method, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=15) as response: return response.status, json.load(response) def normalize_secret(value): if value is None: return None normalized = str(value).strip() return normalized or None def normalize_secrets(values): return {key: normalized for key, value in values.items() if (normalized := normalize_secret(value))} def run_journey(url, goal, max_steps, timeout, screenshots, trace, laya_mode, cognition_profile, hosted_verification, vision_enabled, terminal_verification, openai_key, anthropic_key, gemini_key, hosted_key, vision_key, spark_key, context, domains, confidence): try: metadata = json.loads(context or "{}") except json.JSONDecodeError as error: raise gr.Error(f"Invalid context JSON: {error}") payload = {"url": url, "goal": goal, "maxSteps": int(max_steps), "timeoutMs": int(float(timeout) * 1000), "screenshots": screenshots, "trace": trace, "confidenceThreshold": float(confidence), "context": {**metadata, "layaMode": laya_mode}, "sessionCredentials": normalize_secrets({"OPENAI_API_KEY": openai_key, "ANTHROPIC_API_KEY": anthropic_key, "GEMINI_API_KEY": gemini_key, "LAYA_HOSTED_API_KEY": hosted_key, "LAYA_VISION_API_KEY": vision_key, "SPARK_OPENAI_API_KEY": spark_key}), "cognitionProfile": cognition_profile, "cognitionOptions": {"hostedVerification": hosted_verification, "vision": vision_enabled, "cloudTerminalVerification": terminal_verification}, "allowedDomains": [x.strip() for x in domains.split(",") if x.strip()]} _, created = request("/api/v1/journeys", "POST", payload); journey_id = created["journey_id"] while True: status_code, result = request(f"/api/v1/journeys/{journey_id}/result") if status_code == 200: steps = result.get("steps", []); last = steps[-1] if steps else {}; decision = last.get("decision", {}) summary = {"journey_id": journey_id, "status": result["status"], "current_step": result["step_count"], "current_url": result["final_url"], "candidate_count": len(last.get("candidates", [])), "latest_operation": decision.get("operation"), "selected_target": decision.get("elementIndex"), "confidence": decision.get("confidence")} screenshots_out = result.get("artifacts", {}).get("screenshots", []) yield summary, result.get("events", []), screenshots_out[-1] if screenshots_out else None, result, result return yield {"journey_id": journey_id, "status": result.get("status", "running")}, [], None, {}, result time.sleep(.75) PROFILE_LABELS = { "Local Laya + Cloud": "local-cloud", "Local Laya + Spark + Cloud": "local-spark-cloud", "Local + Hosted Laya + Cloud": "dual-laya-cloud", "Local + Hosted Laya + Spark + Cloud": "dual-laya-spark-cloud", } try: _, RUNTIME_INFO = request("/api/v1/info") except Exception: RUNTIME_INFO = {"cognition": {}} COGNITION = RUNTIME_INFO.get("cognition") or {} DEFAULT_PROFILE = COGNITION.get("cognitionProfile", "local-cloud") IS_SPACE = (COGNITION.get("source") or {}).get("deployment") == "hugging-face" def apply_settings(profile_label, hosted_verification, vision_enabled, terminal_verification, *secrets): profile = PROFILE_LABELS[profile_label] configured = [name for name, value in zip(["OpenAI", "Anthropic", "Gemini", "Hosted Laya", "Laya Vision", "Spark"], secrets) if normalize_secret(value)] suffix = f" Session-only credentials supplied for: {', '.join(configured)}." if configured else "" return profile, hosted_verification, vision_enabled, terminal_verification, f"Applied `{profile}` for this browser session.{suffix} Credentials are not echoed or written." def save_settings(profile_label, hosted_enabled, hosted_url, vision_enabled, vision_url, spark_enabled, spark_url, spark_model, cloud_enabled, provider, openai_base_url, model, attempts, *secrets): if IS_SPACE: return "Running on Hugging Face Space. To persist credentials: Space → Settings → Repository secrets." keys = ["OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GEMINI_API_KEY", "LAYA_HOSTED_API_KEY", "LAYA_VISION_API_KEY", "SPARK_OPENAI_API_KEY"] body = {"COGNITION_PROFILE": PROFILE_LABELS[profile_label], "LAYA_HOSTED_ENABLED": str(hosted_enabled).lower(), "LAYA_HOSTED_BASE_URL": hosted_url or "", "LAYA_VISION_ENABLED": str(vision_enabled).lower(), "LAYA_VISION_BASE_URL": vision_url or "", "SPARK_REASONING_ENABLED": str(spark_enabled).lower(), "SPARK_OPENAI_BASE_URL": spark_url or "", "SPARK_MODEL": spark_model, "CLOUD_REASONING_ENABLED": str(cloud_enabled).lower(), "REASONING_PROVIDER": provider, "OPENAI_BASE_URL": openai_base_url or "", "OPENAI_MODEL": model, "CLOUD_MAX_ATTEMPTS": str(int(attempts))} body.update(normalize_secrets(dict(zip(keys, secrets)))) _, result = request("/api/v1/settings/local", "POST", body); return result["message"] def test_connections(): _, result = request("/api/v1/settings/test", "POST", {}); return result def build_ui(): with gr.Blocks(title="JourneyTest") as demo: gr.Markdown("# JourneyTest Developer UI\nAll runs use the shared JourneyService.") with gr.Row(): url = gr.Textbox(label="Start URL") goal = gr.Textbox(label="Goal", lines=3) profile_state = gr.State(DEFAULT_PROFILE) hosted_verify_state = gr.State(False) vision_state = gr.State(False) terminal_verify_state = gr.State(False) with gr.Row(): max_steps = gr.Number(15, label="Max steps") timeout = gr.Number(120, label="Timeout seconds") screenshots = gr.Checkbox(True, label="Capture screenshots") trace = gr.Checkbox(False, label="Capture trace") mode = gr.Dropdown(["auto", "remote", "embedded"], value="auto", label="Laya mode") with gr.Accordion("Advanced", open=False): context = gr.Code("{}", language="json", label="Context JSON") domains = gr.Textbox(label="Allowed domains (comma-separated)") confidence = gr.Slider(0, 1, .15, label="Confidence threshold") run = gr.Button("Run journey", variant="primary") with gr.Accordion("⚙ Settings", open=False): gr.Markdown("## Cognition Profile") default_label = next((label for label, value in PROFILE_LABELS.items() if value == DEFAULT_PROFILE), "Local Laya + Cloud") profile_select = gr.Dropdown(list(PROFILE_LABELS), value=default_label, label="Profile") with gr.Row(): hosted_verify = gr.Checkbox(False, label="Hosted Laya verification") vision_toggle = gr.Checkbox(False, label="Laya Vision") terminal_verify = gr.Checkbox(False, label="Cloud terminal verification") gr.Markdown("## System 1") with gr.Row(): hosted_enabled = gr.Checkbox(False, label="Hosted Laya enabled") hosted_url = gr.Textbox(label="Hosted Laya base URL") gr.Markdown("## Vision") vision_url = gr.Textbox(label="Laya Vision base URL") gr.Markdown("## Local System 2") with gr.Row(): spark_enabled = gr.Checkbox(DEFAULT_PROFILE.endswith("spark-cloud"), label="Spark enabled") spark_url = gr.Textbox(label="Spark OpenAI base URL") spark_model = gr.Textbox("spark-x2.5-1.7b", label="Spark model") gr.Markdown("## Cloud System 2") with gr.Row(): cloud_enabled = gr.Checkbox(True, label="Cloud enabled") provider = gr.Dropdown(["openai", "anthropic", "gemini"], value="openai", label="Provider protocol") openai_base_url = gr.Textbox(label="OpenAI-compatible Base URL") reasoning_model = gr.Textbox(label="Model ID") gr.Markdown("Credentials are session-only unless **Save local configuration** is explicitly selected. Existing values are never loaded into these fields.") with gr.Row(): openai_key = gr.Textbox(label="OpenAI — " + credential_status("openai"), type="password") anthropic_key = gr.Textbox(label="Anthropic — " + credential_status("anthropic"), type="password") gemini_key = gr.Textbox(label="Gemini — " + credential_status("gemini"), type="password") with gr.Row(): hosted_key = gr.Textbox(label="Hosted Laya — " + credential_status("hostedLaya"), type="password") vision_key = gr.Textbox(label="Laya Vision — " + credential_status("layaVision"), type="password") spark_key = gr.Textbox(label="Spark auth — " + credential_status("spark"), type="password") gr.Markdown("## Advanced / retries") attempts = gr.Number(3, minimum=1, maximum=5, label="Cloud max attempts") deployment = "Running on Hugging Face Space. Session changes are ephemeral. To persist a credential: Space → Settings → Repository secrets." if IS_SPACE else "Running on localhost. Save local configuration writes `.env.local` with mode 0600." gr.Markdown("## Deployment\n" + deployment) settings_status = gr.Markdown(json.dumps(COGNITION, indent=2)) with gr.Row(): apply_button = gr.Button("Apply for this session", variant="primary") test_button = gr.Button("Test connections") save_button = gr.Button("Save local configuration", visible=not IS_SPACE) with gr.Row(): status = gr.JSON(label="Live status") screenshot = gr.Image(label="Latest screenshot", type="filepath") events = gr.JSON(label="Event feed") result = gr.JSON(label="Result") raw = gr.JSON(label="Raw JourneyResult JSON") with gr.Tab("Developer / API"): gr.Markdown(f"API base URL: `{API}`\n\nHealth: `{API}/api/v1/health` · Ready: `{API}/api/v1/ready`\n\n```bash\ncurl -X POST {API}/api/v1/journeys -H 'content-type: application/json' -d '{{\"url\":\"https://example.com\",\"goal\":\"Open pricing\"}}'\n```") # Gradio 5 requires registration in a Blocks context. Re-enter only after # every referenced component has been constructed. with demo: secret_inputs = [openai_key, anthropic_key, gemini_key, hosted_key, vision_key, spark_key] run.click(run_journey, [url, goal, max_steps, timeout, screenshots, trace, mode, profile_state, hosted_verify_state, vision_state, terminal_verify_state, *secret_inputs, context, domains, confidence], [status, events, screenshot, result, raw]) apply_button.click(apply_settings, [profile_select, hosted_verify, vision_toggle, terminal_verify, *secret_inputs], [profile_state, hosted_verify_state, vision_state, terminal_verify_state, settings_status]) test_button.click(test_connections, [], settings_status) save_button.click(save_settings, [profile_select, hosted_enabled, hosted_url, vision_toggle, vision_url, spark_enabled, spark_url, spark_model, cloud_enabled, provider, openai_base_url, reasoning_model, attempts, *secret_inputs], settings_status) return demo def credential_status(name): value = (COGNITION.get("credentials") or {}).get(name) return "configured ✓" if value is True or value == "configured" else "not configured" def build_app(): demo = build_ui() @asynccontextmanager async def lifespan(app): # One pool is safe to reuse; transport errors do not poison httpx's # client. Explicit connect/read/write/pool limits keep failures # distinguishable without extending journey execution timeouts. app.state.upstream = httpx.AsyncClient(timeout=httpx.Timeout(connect=5, read=30, write=10, pool=5)) try: yield finally: await app.state.upstream.aclose() app = FastAPI(title="JourneyTest API Proxy", docs_url="/docs", lifespan=lifespan) try: _, public_openapi = request("/api/v1/openapi.json") app.openapi = lambda: public_openapi except Exception: pass @app.get("/health") async def health_endpoint(incoming: Request): return await proxy_request(incoming, f"{API}/health", timeout=10) @app.get("/api-docs") async def api_docs_endpoint(incoming: Request): return await proxy_request(incoming, f"{API}/api-docs", timeout=10) @app.api_route("/api/{path:path}", methods=["GET", "POST"]) async def api_proxy(path: str, incoming: Request): return await proxy_request(incoming, f"{API}/api/{path}") return gr.mount_gradio_app(app, demo, path="/"), demo async def proxy_request(incoming, url, timeout=None): client = incoming.app.state.upstream try: upstream = await client.request(incoming.method, url, content=await incoming.body(), headers={"content-type": incoming.headers.get("content-type", "application/json")}, timeout=timeout) except httpx.TimeoutException: return Response(json.dumps({"error": {"code": "UPSTREAM_TIMEOUT", "message": "JourneyTest API did not respond before the proxy deadline.", "retryable": True}}), status_code=504, media_type="application/json") except httpx.RequestError: return Response(json.dumps({"error": {"code": "UPSTREAM_UNAVAILABLE", "message": "JourneyTest API is temporarily unavailable.", "retryable": True}}), status_code=502, media_type="application/json") headers = {key: value for key, value in upstream.headers.items() if key.lower() in {"content-type", "content-disposition", "cache-control"}} return Response(upstream.content, status_code=upstream.status_code, headers=headers) if __name__ == "__main__": application, _demo = build_app() uvicorn.run(application, host=os.environ.get("JOURNEYTEST_HOST", "0.0.0.0"), port=int(os.environ.get("JOURNEYTEST_PORT", "7860")))