import json
import os
import re
import ast
from pathlib import Path
import gradio as gr
import httpx
LANGUAGES = ["Python", "JavaScript", "TypeScript", "Rust", "Go", "C++"]
def load_local_env() -> None:
env_path = Path(".env")
if not env_path.exists():
return
for line in env_path.read_text(encoding="utf-8").splitlines():
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip())
load_local_env()
MODAL_VERIFIER_URL = os.environ.get("MODAL_VERIFIER_URL")
MODAL_SANDBOX_URL = os.environ.get("MODAL_SANDBOX_URL")
os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")
def load_static(filename: str) -> str:
return Path("static", filename).read_text(encoding="utf-8")
def endpoint_url(url: str | None, path: str) -> str | None:
if not url:
return None
clean = url.rstrip("/")
if clean.endswith(path):
return clean
return f"{clean}{path}"
custom_html = f"""
Local DraftWebGPU 1.5B
fast browser stream
Cloud CheckModal A10G 14B
llama.cpp verifier
WebGPU not detected. Use Chrome 113+ on desktop for local inference.
Model not loaded
Speculative draftIdle
Waiting for model load...
Idle0 tok/sVerifier idle
"""
async def verify_with_modal(prompt: str, draft_code: str, language: str) -> str:
draft_code = strip_markdown_code_fence(draft_code)
verifier_url = endpoint_url(MODAL_VERIFIER_URL, "/verify")
if not verifier_url:
return json.dumps(
{
"verdict": "PASS",
"reason": "MODAL_VERIFIER_URL is not configured; local demo fallback used.",
}
)
try:
async with httpx.AsyncClient(timeout=180.0) as client:
response = await client.post(
verifier_url,
json={"prompt": prompt, "draft_code": draft_code, "language": language.lower()},
)
response.raise_for_status()
return response.text
except Exception as exc:
return json.dumps({"verdict": "ERROR", "reason": str(exc)})
async def execute_in_sandbox(code: str) -> dict:
code = strip_markdown_code_fence(code)
sandbox_url = endpoint_url(MODAL_SANDBOX_URL, "/execute")
if not sandbox_url:
return {"stdout": "", "stderr": "Sandbox not configured", "returncode": -1}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(sandbox_url, json={"code": code})
response.raise_for_status()
return response.json()
def code_from_verdict(draft_code: str, verdict_json: str) -> str:
draft_code = strip_markdown_code_fence(draft_code)
if not verdict_json:
return draft_code
try:
verdict = json.loads(verdict_json)
except json.JSONDecodeError:
return draft_code
return strip_markdown_code_fence(verdict.get("corrected_code") or draft_code)
def strip_markdown_code_fence(code: str) -> str:
text = (code or "").strip()
if not text:
return ""
opening_fence = re.match(r"^```(?:[a-zA-Z0-9_+#.-]+)?\s*\n?", text)
if opening_fence:
text = text[opening_fence.end() :]
closing_index = text.find("```")
if closing_index >= 0:
text = text[:closing_index]
else:
first_fence = text.find("```")
if first_fence >= 0:
text = text[:first_fence]
return trim_markdown_explanation(text)
def trim_markdown_explanation(text: str) -> str:
explanation = re.compile(
r"^\s*(?:[-*]\s+|\d+\.\s+|#{1,6}\s+|Explanation\s*:|Steps\s*:|Notes?\s*:|The code\b|This code\b)",
re.IGNORECASE,
)
kept = []
for line in text.splitlines():
if explanation.match(line):
break
kept.append(line)
return "\n".join(kept).strip()
async def run_sandbox(language: str, draft_code: str, verdict_json: str) -> str:
if language.lower() != "python":
return "Sandbox execution is currently wired for Python only."
code = prepare_python_for_sandbox(code_from_verdict(draft_code, verdict_json))
if not code.strip():
return "No generated code is available yet."
result = await execute_in_sandbox(code)
stdout = result.get("stdout", "")
stderr = result.get("stderr", "")
returncode = result.get("returncode", "")
return "\n".join(
[
f"returncode: {returncode}",
"",
"stdout:",
stdout or "",
"",
"stderr:",
stderr or "",
]
)
def prepare_python_for_sandbox(code: str) -> str:
code = strip_markdown_code_fence(code)
try:
tree = ast.parse(code)
except SyntaxError:
return code
executable_nodes = (
ast.Assign,
ast.AugAssign,
ast.AnnAssign,
ast.Assert,
ast.Delete,
ast.Expr,
ast.For,
ast.AsyncFor,
ast.While,
ast.If,
ast.Match,
ast.Raise,
ast.Return,
ast.Try,
ast.With,
ast.AsyncWith,
)
has_top_level_execution = any(isinstance(node, executable_nodes) for node in tree.body)
if has_top_level_execution:
return code
for node in tree.body:
if isinstance(node, ast.FunctionDef) and function_has_no_required_args(node):
return f'{code}\n\nif __name__ == "__main__":\n {node.name}()\n'
return code
def function_has_no_required_args(node: ast.FunctionDef) -> bool:
args = node.args
positional = [*args.posonlyargs, *args.args]
required_positional = len(positional) - len(args.defaults)
required_kwonly = sum(
1 for arg, default in zip(args.kwonlyargs, args.kw_defaults) if default is None
)
return required_positional == 0 and required_kwonly == 0
with gr.Blocks(
title="Split-Brain Co-Pilot",
css=load_static("style.css"),
theme=gr.themes.Base(primary_hue="blue", neutral_hue="slate"),
) as demo:
gr.HTML(
"""
Build Small Hackathon ยท 15.5B parameters total
Split-Brain Co-Pilot
One small model drafts in your browser. Another small model checks it on Modal. The UI shows the handoff, verdict, rollback, and executable proof.