""" app.py ------ AutoDevAgent — Gradio UI entry point. This is the only file that imports from the UI layer and wires all components together. It assembles the Gradio Blocks layout, registers event handlers, and launches the app for HuggingFace Spaces. Layout: ┌─────────────────────────────────────────────────────┐ │ AutoDevAgent │ │ ┌──────────────────┐ ┌──────────────────────────┐ │ │ │ Task input │ │ Pipeline Visualiser │ │ │ │ Language select │ │ (fixed nodes, live │ │ │ │ Model switcher │ │ animated arrows) │ │ │ │ Run / Benchmark │ └──────────────────────────┘ │ │ └──────────────────┘ │ │ ┌────────────────────────────────────────────────┐ │ │ │ Output tabs: │ │ │ │ Code | Tests | Explanation | Flowchart │ │ │ └────────────────────────────────────────────────┘ │ │ ┌────────────────────────────────────────────────┐ │ │ │ Self-reflection accordion │ Stats strip │ │ │ └────────────────────────────────────────────────┘ │ │ ┌────────────────────────────────────────────────┐ │ │ │ HITL panel (visible only on max retries) │ │ │ └────────────────────────────────────────────────┘ │ │ ┌────────────────────────────────────────────────┐ │ │ │ Session history (collapsible) │ │ │ └────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────┘ Usage: python app.py # local dev gradio app.py # HuggingFace Spaces (auto-detected) """ from dotenv import load_dotenv load_dotenv() # load .env before any other imports read env vars import logging import time import gradio as gr from config import settings from pipeline.graph import build_graph from pipeline.state import Language, PipelineStatus, PipelineState from agents.detect_agent import DetectAgent from ui.components import ( build_hitl_panel, build_clarification_panel, build_reflection_panel, build_stats_strip, build_model_switcher, build_language_selector, update_hitl_panel, update_reflection_panel, update_stats_strip, update_language_badge, ) from ui.pipeline_visualiser import ( get_pipeline_html, get_pipeline_html_with_update, get_status_update_js, STATUS_TO_ACTIVE_NODE, ) from ui.session_history import SessionHistory from observability.wandb_tracker import WandbTracker from evaluation.benchmark import BenchmarkRunner from evaluation.metrics import compute_metrics, format_results_for_display logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ------------------------------------------------------------------ # # Example prompts per language # # ------------------------------------------------------------------ # PYTHON_EXAMPLES = [ "Write a recursive function to compute the nth Fibonacci number. Then add memoization to make it efficient for n=100", "Write a function that takes a list of integers and returns the longest strictly increasing subsequence", "Create a classBankAccountwith deposit, withdraw, and get_balance methods. Raise an exception if withdrawal exceeds balance", "Write a function that checks if two strings are anagrams (case-insensitive, ignore spaces and punctuation)", ] SQL_EXAMPLES = [ "Write a query to find the second highest salary from an employees table. Assume no schema is given - infer it", "Calculate the running total of sales per product over time. Use a sales table with columns: product_id, sale_date, amount", "Find customers who have placed orders in the last 30 days but not in the previous 30 days (newly active customers)", "Write a query to de-duplicate a tableuser_eventskeeping only the latest event per user based on event_time", ] # ------------------------------------------------------------------ # # Mermaid renderer # # ------------------------------------------------------------------ # def _parse_test_docstrings(test_code: str) -> dict[str, str]: """ Extract {method_name: docstring} from a unittest file. Handles both single and triple-quoted docstrings. """ import re result = {} # Match: def test_foo(self): followed by an optional docstring pattern = re.compile( r'def\s+(test_\w+)\s*\(self[^)]*\)\s*:\s*\n' r'\s*(?:"""(.*?)"""|\'\'\'(.*?)\'\'\')', re.DOTALL, ) for m in pattern.finditer(test_code or ""): name = m.group(1) doc = (m.group(2) or m.group(3) or "").strip() doc = " ".join(doc.split()) # collapse whitespace result[name] = doc return result def _build_test_summary(test_result, test_code: str = "") -> str: """Return a markdown summary of test results for display in the Tests tab.""" if test_result is None: return "" total = test_result.total_tests passed = test_result.passed_tests failed = test_result.failed_tests icon = "✅" if failed == 0 else "⚠️" docstrings = _parse_test_docstrings(test_code) # Failed test names extracted from failure messages import re failed_names = set() for f in test_result.failures: m = re.search(r'(test_\w+)', f) if m: failed_names.add(m.group(1)) _summary_color = "#22C55E" if failed == 0 else "#F59E0B" lines = [f'{icon} **{passed}/{total} tests passed**\n'] # Per-test breakdown for name, doc in docstrings.items(): status = "❌" if name in failed_names else "✅" desc = f" — *{doc}*" if doc else "" lines.append(f"{status} `{name}`{desc}") if failed > 0: lines.append(f"\n**{failed} failure detail(s):**") for f in test_result.failures: lines.append(f"```\n{f.strip()}\n```") return "\n\n".join(lines) def _generate_report(task: str, lang: str, final_state, elapsed: float) -> str: """ Build a full markdown report for download. Includes task, code, test results with per-test descriptions, explanation, and run stats. """ import re tr = final_state.test_result test_code = final_state.generated_tests or "" test_lines = [] if tr: test_lines += [ f"- **Total:** {tr.total_tests}", f"- **Passed:** {tr.passed_tests}", f"- **Failed:** {tr.failed_tests}", ] # Per-test breakdown with docstrings docstrings = _parse_test_docstrings(test_code) failed_names = set() for f in tr.failures: m = re.search(r'(test_\w+)', f) if m: failed_names.add(m.group(1)) if docstrings: test_lines.append("\n**Per-test breakdown:**") for name, doc in docstrings.items(): status = "❌" if name in failed_names else "✅" desc = f": {doc}" if doc else "" test_lines.append(f"- {status} `{name}`{desc}") if tr.failures: test_lines.append("\n**Failure details:**") for f in tr.failures: test_lines.append(f"```\n{f.strip()}\n```") else: test_lines = ["*No test results available.*"] code_lang = lang if lang in ("python", "sql") else "python" code_block = f"```{code_lang}\n{final_state.final_code()}\n```" tests_block = f"```python\n{final_state.generated_tests or ''}\n```" tokens = final_state.token_usage.total_tokens iters = final_state.debug_iterations status = final_state.status.value sections = [ "# AutoDevAgent — Run Report", f"## Task\n{task}", f"## Language\n`{lang}`", f"## Generated Code\n{code_block}", "## Test Results\n" + "\n".join(test_lines), f"## Test Code\n{tests_block}", f"## Explanation\n{final_state.explanation or '*Not generated.*'}", ( "## Run Stats\n" f"- **Status:** {status}\n" f"- **Debug iterations:** {iters}\n" f"- **Total tokens:** {tokens:,}\n" f"- **Elapsed time:** {elapsed:.1f}s" ), ] return "\n\n---\n\n".join(sections) def _render_mermaid(raw: str) -> str: """ Parse-then-rebuild Mermaid renderer (strict simple mode). Parses whatever the LLM produced into nodes + edges, then rebuilds 100% clean Mermaid using only the safest subset: - graph TD header - rectangle nodes only A["label"] - plain arrows only A --> B (no edge labels, no diamonds) - no back-edges (DFS cycle removal) - max 10 nodes (truncated if LLM went over) This means any LLM syntax error, diamond node, edge label, pipe character, or loop can NEVER reach the Mermaid renderer. """ import re if not raw or not raw.strip(): return "_No flowchart generated for this task._" code = raw.strip() # ── Strip markdown fences ─────────────────────────────────────── # if code.startswith("```"): lines = code.splitlines() end = len(lines) - 1 if lines[-1].strip() == "```" else len(lines) code = "\n".join(lines[1:end]).strip() # ── Handle single-line LLM output ────────────────────────────── # if "\n" not in code: code = re.sub(r'(?<=[}\]\)]) (?=[A-Za-z])', '\n ', code) lines = code.splitlines() # ── Label cleaner ─────────────────────────────────────────────── # # Remove ALL characters that can break Mermaid inside a label. _UNSAFE_RE = re.compile(r'["\';:<>&|?=%^\\*(),./\[\]{}]') def _clean_label(t: str) -> str: t = t.strip().strip('"\'') t = _UNSAFE_RE.sub('', t).strip() t = re.sub(r'\s{2,}', ' ', t) return t[:40] or "step" # cap label length; never empty # ── Reserved keywords that must never be node IDs ─────────────── # _RESERVED = frozenset({ 'flowchart', 'graph', 'subgraph', 'end', 'style', 'classDef', 'class', 'click', 'call', 'href', 'linkStyle', 'direction', 'TB', 'TD', 'BT', 'RL', 'LR', }) # ── Patterns ──────────────────────────────────────────────────── # # Matches any node definition: A["label"], A[label], A("label"), A{label} NODE_DEF_RE = re.compile( r'\b([A-Za-z][A-Za-z0-9_]*)\s*' r'([\[{\(])' r'((?:"[^"]*"|\'[^\']*\'|[^\]}\)])*)' r'[\]}\)]' ) # Matches edges: A --> B or A -->|label| B or A -- label --> B EDGE_RE = re.compile( r'\b([A-Za-z][A-Za-z0-9_]*)\b\s*' r'(?:--+>|==+>|-\.+->)\s*' r'(?:\|[^|\n]*\|)?\s*' # discard any edge label r'\b([A-Za-z][A-Za-z0-9_]*)\b' ) EDGE_LABEL_MID_RE = re.compile( # A -- label --> B r'\b([A-Za-z][A-Za-z0-9_]*)\b\s*--\s*[^->\n]+?\s*-->\s*' r'\b([A-Za-z][A-Za-z0-9_]*)\b' ) # ── Parse ─────────────────────────────────────────────────────── # nodes: dict[str, str] = {} # id → label edges: list[tuple[str, str]] = [] # (src, dst) — no labels kept _in_subgraph = False for line in lines: stripped = line.strip() if not stripped or stripped.startswith('%'): continue lower = stripped.lower() if lower.startswith(('flowchart', 'graph')): continue if lower.startswith('subgraph'): _in_subgraph = True continue if lower == 'end' and _in_subgraph: _in_subgraph = False continue if _in_subgraph: continue # Collect node definitions for nm in NODE_DEF_RE.finditer(stripped): nid, content = nm.group(1), nm.group(3) if nid in _RESERVED: continue if nid not in nodes: nodes[nid] = _clean_label(content) # Strip node definitions before edge matching line_for_edges = NODE_DEF_RE.sub(lambda m: m.group(1), stripped) # Try A -- label --> B first, then plain A --> B m = EDGE_LABEL_MID_RE.search(line_for_edges) if m: edges.append((m.group(1), m.group(2))) else: m = EDGE_RE.search(line_for_edges) if m: edges.append((m.group(1), m.group(2))) # ── Remap reserved-keyword node IDs ───────────────────────────── # _id_remap: dict[str, str] = {} _suffix = 0 def _safe_id(nid: str) -> str: nonlocal _suffix if nid not in _RESERVED: return nid if nid not in _id_remap: _suffix += 1 _id_remap[nid] = f"N{_suffix}" return _id_remap[nid] nodes = {_safe_id(k): v for k, v in nodes.items()} edges = [(_safe_id(s), _safe_id(d)) for s, d in edges] # Ensure every node referenced in an edge has a definition for src, dst in edges: for nid in (src, dst): if nid not in nodes: nodes[nid] = nid if not nodes and not edges: logger.warning("Mermaid parser: nothing extracted — plain-text fallback") return "_Flowchart could not be rendered — the diagram syntax was invalid._" # ── Deduplicate edges ─────────────────────────────────────────── # seen: set[tuple[str, str]] = set() unique_edges: list[tuple[str, str]] = [] for s, d in edges: if (s, d) not in seen: seen.add((s, d)) unique_edges.append((s, d)) edges = unique_edges # ── Remove back-edges (DFS cycle detection) ───────────────────── # adj: dict[str, list[str]] = {} for s, d in edges: adj.setdefault(s, []).append(d) WHITE, GRAY, BLACK = 0, 1, 2 color: dict[str, int] = {n: WHITE for n in nodes} back_pairs: set[tuple[str, str]] = set() for start in list(nodes): if color.get(start, WHITE) != WHITE: continue stack = [(start, iter(adj.get(start, [])))] color[start] = GRAY while stack: node, children = stack[-1] try: child = next(children) if color.get(child, WHITE) == GRAY: back_pairs.add((node, child)) elif color.get(child, WHITE) == WHITE: color[child] = GRAY stack.append((child, iter(adj.get(child, [])))) except StopIteration: color[node] = BLACK stack.pop() edges = [(s, d) for s, d in edges if (s, d) not in back_pairs] # ── Drop orphaned subgraphs — keep only nodes reachable from root ─ # # The LLM sometimes generates disconnected subgraphs (e.g. nodes D and E # with D→E but no edge from the main flow into D). Find the root (first # node with no incoming edges) then BFS to collect only reachable nodes. _incoming: set[str] = {d for _, d in edges} _root_candidates = [n for n in nodes if n not in _incoming] if _root_candidates: _root = _root_candidates[0] _adj: dict[str, list[str]] = {} for s, d in edges: _adj.setdefault(s, []).append(d) _reachable: set[str] = set() _queue = [_root] while _queue: _cur = _queue.pop(0) if _cur in _reachable: continue _reachable.add(_cur) for _nb in _adj.get(_cur, []): if _nb not in _reachable: _queue.append(_nb) _dropped = set(nodes) - _reachable if _dropped: logger.info("Mermaid: dropping orphaned nodes %s", _dropped) nodes = {k: v for k, v in nodes.items() if k in _reachable} edges = [(s, d) for s, d in edges if s in _reachable and d in _reachable] # ── Cap at 10 nodes ───────────────────────────────────────────── # MAX_NODES = 10 if len(nodes) > MAX_NODES: keep = set(list(nodes)[:MAX_NODES]) nodes = {k: v for k, v in nodes.items() if k in keep} edges = [(s, d) for s, d in edges if s in keep and d in keep] # ── Sanity check ─────────────────────────────────────────────── # if len(nodes) < 2 or not edges: logger.warning("Mermaid rebuild: too little structure (%d nodes, %d edges)", len(nodes), len(edges)) return "_Flowchart could not be rendered — not enough structure was extracted._" # ── Rebuild: rectangles only, plain arrows, no edge labels ────── # out_lines = ["graph TD"] for nid, label in nodes.items(): out_lines.append(f' {nid}["{label}"]') for src, dst in edges: out_lines.append(f' {src} --> {dst}') result = "\n".join(out_lines) logger.info("Mermaid rebuilt (%d nodes, %d edges):\n%s", len(nodes), len(edges), result) return f"```mermaid\n{result}\n```" # ------------------------------------------------------------------ # # Core run handler # # ------------------------------------------------------------------ # def run_agent( task: str, language: str, model: str, session_history_state, skip_clarification: bool = False, ): """ Main handler — runs the full pipeline and yields UI updates. Generator function so Gradio can cancel it via the Stop button. Uses LangGraph streaming to yield per-node pipeline visualiser updates, giving live node-by-node animation. Yields: 1. Immediately — clears outputs and shows "Running..." in status bar. 2. After each LangGraph node — updates pipeline_vis and status_bar. 3. After pipeline completes — populates all output components. """ # Gradio serializes gr.State to JSON between requests — reconstruct # SessionHistory if it was deserialised as a plain list or dict. if not isinstance(session_history_state, SessionHistory): session_history_state = SessionHistory() _empty = ("", "", "", "", gr.update(visible=False), "", "", "") _no_change = ( gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), ) # 2 trailing no-ops for clarification panel (hidden by default) _clarify_hidden = (gr.update(visible=False), gr.update()) # 4 trailing no-ops for HITL components (hidden by default) _hitl_hidden = (gr.update(visible=False), gr.update(), gr.update(), gr.update()) if not task or not task.strip(): yield _empty + ("Please enter a task description.", session_history_state, get_pipeline_html(), gr.update(), gr.update(visible=False, value=""), gr.update(), gr.update()) + _clarify_hidden + _hitl_hidden return # "auto" means ModelRouter decides; anything else is a pinned override if model and model != "auto": settings.groq_model_primary = model start_time = time.time() # ── Reset error cache for this run ────────────────────────────── # from agents.debug_agent import _error_cache as _debug_cache _debug_cache.reset() # ── Step 1: Non-coding task gate (runs silently, no card animation) # # Reject non-programming tasks immediately before any other checks. from agents.detect_agent import DetectAgent _detector = DetectAgent() is_coding, gate_reason = _detector.is_coding_task(task) if not is_coding: session_history_state.record_skip(task, gate_reason) _skip_msg = ( "⚠️ **I'm a code assistant for Python and SQL.**\n\n" "I can't help with questions like this. Please give me a programming task — for example:\n" "- *Write a Python function to sort a list*\n" "- *Write a SQL query to find the top 5 customers by revenue*" ) yield _empty + ( _skip_msg, session_history_state, get_pipeline_html(), gr.update(), gr.update(visible=False, value=""), gr.update(), gr.update(), ) + _clarify_hidden + _hitl_hidden return # ── Step 2: Scope check ─────────────────────────────────────────── # Confirmed coding task — but is it a single snippet or a full project? # "Build a YouTube app" passes the coding gate but is weeks of work. _in_scope, _scope_suggestion = _detector.is_in_scope(task) if not _in_scope: session_history_state.record_skip(task, "out of scope — full project, not a single snippet") _suggestion_line = ( f"\n\nFor example: *{_scope_suggestion}*" if _scope_suggestion else "" ) _scope_msg = ( "📦 **That's a large project, not a single code snippet.**\n\n" "This assistant generates focused Python/SQL code — a single function, " "query, or short script (≤ ~100 lines).\n\n" "Could you break it down into a specific, concrete task?" f"{_suggestion_line}" ) yield _empty + ( _scope_msg, session_history_state, get_pipeline_html(), gr.update(), gr.update(visible=False, value=""), gr.update(), gr.update(), ) + _clarify_hidden + _hitl_hidden return # ── Step 3: Clarification check ────────────────────────────────── # In-scope coding task — check if the requirement is specific enough. # Skipped when the user has already answered a clarification question. if not skip_clarification: yield _empty + ( "⏳ Checking task clarity...", session_history_state, get_pipeline_html_with_update("clarify", 0, False, {}), gr.update(), gr.update(visible=False, value=""), gr.update(), gr.update(), ) + _clarify_hidden + _hitl_hidden from agents.clarification_agent import ClarificationAgent _clarify_result = ClarificationAgent().check(task) if not _clarify_result["clear"]: _question = _clarify_result["question"] logger.info("ClarificationAgent: needs clarification — %s", _question) _clarify_visible = ( gr.update(visible=True), # clarification_panel gr.update(value=_question), # clarification_q_md ) yield _empty + ( "🤔 Please clarify your task before I proceed.", session_history_state, get_pipeline_html_with_update("clarify", 0, True, {}), gr.update(), gr.update(visible=False, value=""), gr.update(), gr.update(), ) + _clarify_visible + _hitl_hidden return # ── Step 3: Show Language Detector as active before detection runs # yield _empty + ("⏳ Detecting language...", session_history_state, get_pipeline_html_with_update("detect", 0, False, {}), gr.update(), gr.update(visible=False, value=""), gr.update(), gr.update()) + _clarify_hidden + _hitl_hidden # ── Language resolution ───────────────────────────────────────── # # "auto" → run model detection; "python"/"sql" → trust user's choice _is_auto_lang = (language == "auto") try: detection = _detector.detect(task) if _is_auto_lang: # Auto mode: use detected language (fall back to Python) if detection.language != Language.UNKNOWN and detection.confidence in ("high", "medium"): lang = detection.language logger.info( "run_agent: auto-detected language=%s (%s)", lang.value, detection.confidence, ) else: lang = Language.PYTHON logger.info("run_agent: detection low-confidence, defaulting to Python") else: # Manual selection: honour the radio as-is lang = Language(language.lower()) if language.lower() in ("python", "sql") else Language.PYTHON except Exception as e: logger.warning("run_agent: detection failed (%s), using fallback", e) if _is_auto_lang: lang = Language.PYTHON else: lang = Language(language.lower()) if language.lower() in ("python", "sql") else Language.PYTHON # Build detection badge message _lang_cap = lang.value.capitalize() if _is_auto_lang: _detect_msg = f'✓ Detected {_lang_cap}. Using {_lang_cap} runtime.' else: _detect_msg = f'✓ Selected {_lang_cap}. Using {_lang_cap} runtime.' _detect_badge_update = gr.update(visible=True, value=_detect_msg) # Yield the badge immediately after detection using _no_change so nothing else # is touched. This guarantees the badge renders on the FIRST run — Gradio batches # rapid pre-pipeline yields on first page load and only paints the last state, so # putting the badge inside a yield that also clears outputs (via _empty) risks it # never being painted on its own frame. A dedicated no-change yield gives the # browser a clean frame just for the badge. yield _no_change + ( "⏳ Running pipeline...", session_history_state, gr.update(), # pipeline_vis — no change (keep detect:active) gr.update(), # language_radio — no change _detect_badge_update, # detect_badge — show NOW gr.update(), # test_summary_md gr.update(), # download_btn ) + _clarify_hidden + _hitl_hidden from agents.model_router import route_models # Pre-run router so the plan node skips it and model badges appear from step 1 prebuilt_ma = route_models(task, lang.value, model).to_dict() initial_state = PipelineState(task=task, language=lang, model_assignments=prebuilt_ma) graph = build_graph() logger.info( "run_agent: starting pipeline — task: %s, language: %s", task[:60], lang.value, ) last_chunk: dict = {} try: for chunk in graph.stream(initial_state, stream_mode="values", config={"recursion_limit": settings.graph_recursion_limit}): last_chunk = chunk # Skip the initial state snapshot (status == idle) raw_status = chunk.get("status", PipelineStatus.IDLE) if raw_status == PipelineStatus.IDLE: continue status_val = raw_status.value if hasattr(raw_status, "value") else str(raw_status) iterations = chunk.get("debug_iterations", 0) or 0 is_error = raw_status in ( PipelineStatus.CLASSIFYING, PipelineStatus.DEBUGGING, PipelineStatus.AWAITING_HUMAN, PipelineStatus.TEST_FAILING, PipelineStatus.TEST_DEBUGGING, ) # Map status → LangGraph node name for the visualiser _status_to_node = { "clarifying": "clarify", "planning": "plan", "generating": "generate", "regenerating": "regen", # fresh regen after debug exhausted "executing": "execute", "classifying": "classify_error", "debugging": "debug_agent", "testing": "test", "test_failing": "test", "test_debugging": "test_debug_agent", "explaining": "explain", "success": "success", "awaiting_human": "human_loop", } node_name = _status_to_node.get(status_val, "") ma = chunk.get("model_assignments") or {} pipeline_html = ( get_pipeline_html_with_update(node_name, iterations, is_error, ma) if node_name else get_pipeline_html() ) yield _no_change + ( f"⏳ {status_val.replace('_', ' ').capitalize()}...", session_history_state, pipeline_html, gr.update(), gr.update(), # detect_badge — no change during streaming gr.update(), gr.update(), ) + _clarify_hidden + _hitl_hidden except Exception as e: logger.error("run_agent: pipeline error: %s", e) yield _empty + (f"❌ Pipeline error: {e}", session_history_state, get_pipeline_html(), gr.update(), gr.update(), gr.update(), gr.update()) + _clarify_hidden + _hitl_hidden return if not last_chunk: yield _empty + ("❌ Pipeline returned no results.", session_history_state, get_pipeline_html(), gr.update(), gr.update(), gr.update(), gr.update()) + _clarify_hidden + _hitl_hidden return elapsed = round(time.time() - start_time, 2) # ── Build final PipelineState from last streaming chunk ───────── # try: final_state = PipelineState(**last_chunk) except Exception as e: logger.error("run_agent: could not reconstruct final state: %s", e) yield _empty + (f"❌ State error: {e}", session_history_state, get_pipeline_html(), gr.update(), gr.update(), gr.update(), gr.update()) + _clarify_hidden + _hitl_hidden return session_history_state.record(final_state, exec_time=elapsed) # ── Extract outputs ───────────────────────────────────────────── # final_code = final_state.final_code() tests = final_state.generated_tests explanation = final_state.explanation flowchart = _render_mermaid(final_state.flowchart_mermaid) status = final_state.status.value iterations = final_state.debug_iterations tokens = final_state.token_usage.total_tokens history_md = session_history_state.to_markdown() # ── Reflection summary ────────────────────────────────────────── # reflection_summary = "" if final_state.reflections_history: last_r = final_state.reflections_history[-1] reflection_summary = ( f"**Last reflection (iteration {iterations}):**\n\n" f"- Saw: {last_r.what_i_saw}\n" f"- Thought: {last_r.what_i_think}\n" f"- Did: {last_r.what_i_will_do}" ) _status_colored = ( f'{status}' if status == "success" else f'partial success' if status == "partial_success" else f'{status}' if status in ("failed", "error") else status ) _sep = '  |  ' stats_summary = ( f'Status: ' f'{_status_colored}{_sep}' f'Iterations: ' f'{iterations}{_sep}' f'Tokens: ' f'{tokens:,}{_sep}' f'Time: ' f'{elapsed:.1f}s' ) # Final pipeline HTML is_hitl = final_state.status == PipelineStatus.AWAITING_HUMAN is_partial = final_state.status == PipelineStatus.PARTIAL_SUCCESS final_node_name = "human_loop" if is_hitl else ("partial_success" if is_partial else "success") final_pipeline_html = get_pipeline_html_with_update( final_node_name, iterations, is_hitl or is_partial, final_state.model_assignments or {}, ) # ── Test summary and downloadable report ─────────────────────── # test_summary = _build_test_summary(final_state.test_result, final_state.generated_tests or "") report_md = _generate_report(task, lang.value, final_state, elapsed) import tempfile, os report_path = os.path.join(tempfile.gettempdir(), "autodev_report.md") with open(report_path, "w", encoding="utf-8") as f: f.write(report_md) logger.info( "run_agent: done — status=%s tokens=%d time=%.1fs", status, tokens, elapsed, ) # ── HITL panel updates ────────────────────────────────────────── # if is_hitl: last_error = "" if final_state.execution_result and final_state.execution_result.error_msg: last_error = final_state.execution_result.error_msg hitl_updates = ( gr.update(visible=True), gr.update(value=last_error), gr.update(choices=final_state.hitl_options or [], value=None), gr.update(value=final_code or ""), ) status_msg = '⚠ Max retries reached — edit the code below and click Resubmit, or choose a fix strategy.' elif is_partial: hitl_updates = _hitl_hidden status_msg = ( '⚠ Pipeline complete — code runs, but not all tests passed ' 'after all retry cycles. Check the Tests tab for details.' ) else: hitl_updates = _hitl_hidden status_msg = f'✓ Pipeline complete — {status}' yield ( final_code, tests, explanation, flowchart, gr.update(visible=bool(flowchart)), reflection_summary, stats_summary, history_md, status_msg, session_history_state, final_pipeline_html, gr.update(), # language_radio — keep user's selection unchanged _detect_badge_update, # detect_badge — show "✓ Detected/Selected X..." gr.update(visible=bool(test_summary), value=test_summary), gr.update(visible=not is_hitl, value=report_path if not is_hitl else None), ) + _clarify_hidden + hitl_updates def submit_clarification( answer: str, original_task: str, language: str, model: str, session_history_state, ): """ Re-run the pipeline after the user provides clarification. Appends the clarification to the original task and re-runs. Skips the clarification check on re-run — user has already answered once. """ if not answer or not answer.strip(): yield from run_agent(original_task, language, model, session_history_state, skip_clarification=True) return refined_task = f"{original_task.strip()}\n\nAdditional context: {answer.strip()}" yield from run_agent(refined_task, language, model, session_history_state, skip_clarification=True) def resubmit_agent( edited_code: str, chosen_option: str, original_task: str, language: str, model: str, session_history_state, ): """ Re-run the pipeline after the user edits code or selects a fix strategy from the HITL panel. Builds a revised task that includes the user's edited code (or chosen fix hint) so the agent has extra context on its next attempt. """ hint = "" if chosen_option: hint = f"\n\nApply this specific fix strategy: {chosen_option}" if edited_code and edited_code.strip(): hint += f"\n\nHere is my edited version of the code — continue from here:\n```\n{edited_code.strip()}\n```" revised_task = original_task.strip() + hint yield from run_agent(revised_task, language, model, session_history_state) def detect_language(task: str) -> tuple: """ Run auto-detection and return badge text + radio pre-selection. Args: task: The user's task description. Returns: Tuple of (badge_markdown, detected_language_string). """ if not task or len(task.strip()) < 5: return "", "python" try: agent = DetectAgent() result = agent.detect(task) lang = result.language.value if result.language != Language.UNKNOWN else "python" emoji = {"high": "🟢", "medium": "🟡", "low": "🔴"}.get(result.confidence, "⚪") badge = ( f"{emoji} Auto-detected: **{lang}** " f"({result.confidence} confidence) — {result.reason}" ) return badge, lang except Exception as e: logger.warning("detect_language failed: %s", e) return "", "python" def run_benchmark( model: str, progress: gr.Progress = gr.Progress(), ): """ Run the full evaluation harness, yielding status bar updates and the final Markdown report. Yields: Tuples of (benchmark_output, status_bar). """ settings.groq_model_primary = model runner = BenchmarkRunner() total = len(runner.tasks) results = [] yield "", "⏳ Starting benchmark..." for i, task in enumerate(runner.tasks): progress((i / total), desc=f"Running: {task.name}...") yield "", f"⏳ Running task {i + 1}/{total}: {task.name}..." result = runner._run_task(task) status_icon = "✅" if result.success else "❌" results.append(result) yield "", f"{status_icon} Task {i + 1}/{total} done: {task.name}" progress(1.0, desc="Benchmark complete.") summary = compute_metrics(results) tracker = WandbTracker() tracker.log_benchmark_run(results, summary) tracker.finish() pct = round(summary.success_rate * 100) report = _format_benchmark_report(results, summary, model) yield report, f"✓ Benchmark complete — {summary.successful_tasks}/{summary.total_tasks} passed ({pct}%)" def _format_benchmark_report(results: list, summary, model: str) -> str: """ Build the full Markdown benchmark report shown in the UI. Sections: - Header (date, model, max retries) - Task Results table (pipe-aligned) - Summary Metrics - Failed Task Details (one block per failed task) """ import datetime from config import settings as cfg date_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") # ── Header ────────────────────────────────────────────────────── # lines = [ "# AutoDevAgent – Benchmark Report", f"**Date:** {date_str} ", f"**Model:** {model} ", f"**Max retries:** {cfg.max_debug_retries}", "", "## Task Results", "", ] # ── Build table rows ──────────────────────────────────────────── # headers = ["ID", "Task", "Language", "Success", "Iterations", "Tokens", "Time (s)"] rows = [] for i, r in enumerate(results, 1): iters = f"{r.iterations} (max)" if not r.success and r.iterations >= cfg.max_debug_retries else str(r.iterations) rows.append([ str(i), r.task_name, r.language.capitalize(), "✅" if r.success else "❌", iters, f"{r.total_tokens:,}", f"{r.exec_time:.1f}", ]) # Compute column widths (max of header and each cell) col_widths = [len(h) for h in headers] for row in rows: for j, cell in enumerate(row): col_widths[j] = max(col_widths[j], len(cell)) def fmt_row(cells): return "| " + " | ".join(c.ljust(col_widths[j]) for j, c in enumerate(cells)) + " |" def sep_row(): return "|-" + "-|-".join("-" * w for w in col_widths) + "-|" lines.append(fmt_row(headers)) lines.append(sep_row()) for row in rows: lines.append(fmt_row(row)) # ── Summary Metrics ───────────────────────────────────────────── # pct = round(summary.success_rate * 100) total_time = round(sum(r.exec_time for r in results), 1) lines += [ "", "## Summary Metrics", "", f"- **Success rate:** {pct}% ({summary.successful_tasks}/{summary.total_tasks})", f"- **Average debug iterations:** {summary.avg_iterations:.1f}", f"- **Average tokens per task:** {summary.avg_tokens:,.0f}", f"- **Total execution time:** {total_time} seconds", ] # ── Failed Task Details ───────────────────────────────────────── # failed = [r for r in results if not r.success] if failed: lines += ["", "## Failed Task Details", ""] for i, r in enumerate(results, 1): if r.success: continue last_error = r.error_message or "Unknown error" lines += [ f"### Task {i}: {r.task_name}", f"- **Error:** Max retries exceeded ({r.iterations})", f"- **Last error:** {last_error}", "", ] return "\n".join(lines) # ------------------------------------------------------------------ # # Gradio UI # # ------------------------------------------------------------------ # _fc_js = """ function fcZoom(z) { window._fcZ = z; var t = document.querySelector('.mermaid svg') || document.querySelector('.mermaid'); if (t) { t.style.transform = 'scale('+z+')'; t.style.transformOrigin = 'top center'; t.style.display = 'block'; } var l = document.getElementById('fc-zl'); if (l) l.textContent = Math.round(z * 100) + '%'; } window._fcZ = 1; """ def build_ui() -> gr.Blocks: """ Assemble the full Gradio Blocks UI. Returns: Configured gr.Blocks instance ready to launch. """ with gr.Blocks(title="AutoDevAgent") as demo: # ── Session state ─────────────────────────────────────────── # session_state = gr.State(SessionHistory()) # ── Header ───────────────────────────────────────────────── # gr.Markdown( "# AutoDevAgent\n" "Autonomous code generation, execution, debugging, and testing " "— powered by LangChain + LangGraph + Groq (Llama 3.1) · " "[GitHub](https://github.com) · " "[HuggingFace Space](https://huggingface.co/spaces/SivaSai8143/ai-research-digest)" ) with gr.Row(): # ── Left column ───────────────────────────────────────── # with gr.Column(scale=1, elem_id="left_col"): # ── Top-left: task inputs ─────────────────────────── # task_input = gr.Textbox( label = "Task description", placeholder = "Write your task here or select an example below", lines = 4, elem_id = "task_input", ) with gr.Row(elem_id="lang_row"): with gr.Column(scale=0, min_width=80): gr.Markdown("**Language**") language_radio = gr.Radio( choices = [("Python", "python"), ("SQL", "sql"), ("Auto (recommended)", "auto")], value = "auto", label = None, container = False, elem_id = "language_radio", scale = 1, ) detect_badge = gr.Markdown( value = "", visible = False, elem_id = "detect_badge", ) with gr.Row(elem_id="model_row"): with gr.Column(scale=0, min_width=80): gr.Markdown("**Model**") model_radio = gr.Dropdown( choices = [ ("🤖 Auto — smart routing (recommended)", "auto"), ("⚡ Llama 3.1 8B — fastest", "llama-3.1-8b-instant"), ("⚖️ Llama 4 Scout 17B — balanced", "meta-llama/llama-4-scout-17b-16e-instruct"), ("💪 Llama 3.3 70B — most capable", "llama-3.3-70b-versatile"), ], value = "auto", label = None, container = False, elem_id = "model_dropdown", scale = 1, ) _all_examples = ( ["Select an example task ▼"] + ["── Python ──"] + PYTHON_EXAMPLES + ["── SQL ──"] + SQL_EXAMPLES ) with gr.Row(elem_id="examples_row"): with gr.Column(scale=0, min_width=80): gr.Markdown("**Examples**") example_dropdown = gr.Dropdown( choices = _all_examples, value = "Select an example task ▼", label = None, container = False, elem_id = "example_dropdown", scale = 1, ) with gr.Row(): run_btn = gr.Button("▶ Run", variant="primary", size="sm") stop_btn = gr.Button("■ Stop", variant="stop", size="sm") clear_btn = gr.Button("✕ Clear", variant="secondary", size="sm") # ── Bottom-left: output tabs ──────────────────────── # gr.Markdown("---") # ── Status bar ────────────────────────────────────── # status_bar = gr.Markdown(value="Ready.", elem_id="status_bar") with gr.Tabs(elem_id="output_tabs"): with gr.TabItem("Code"): code_output = gr.Code( label = "Generated code", language = "python", interactive = False, max_lines = 30, elem_id = "code_output", ) with gr.TabItem("Tests"): test_summary_md = gr.Markdown(value="", visible=False, elem_id="test_summary_md") test_output = gr.Code( label = "Generated tests", language = "python", interactive = False, max_lines = 30, elem_id = "test_output", ) download_btn = gr.DownloadButton( label = "Download Report", visible = False, elem_id = "download_btn", ) with gr.TabItem("Explanation"): explain_output = gr.Markdown(value="_No explanation generated yet._", label="Plain-English explanation", elem_id="explain_output") with gr.TabItem("Flowchart"): fc_zoom_bar = gr.HTML(visible=False, value="""
100%
""") flowchart_output = gr.Markdown( value = "_No flowchart generated yet._", label = "Logic flowchart", elem_id = "flowchart_output", ) # ── Stats + reflection ────────────────────────────── # stats_md = gr.Markdown(value="", elem_id="stats_md") with gr.Accordion("Self-reflection log", open=False, visible=False) as reflection_panel: reflection_md = gr.Markdown(value="") # ── Clarification panel ───────────────────────────── # clarification_components = build_clarification_panel() # ── HITL panel ────────────────────────────────────── # hitl_components = build_hitl_panel() # ── Session history ───────────────────────────────── # with gr.Accordion("Session history", open=False): history_md = gr.Markdown(value="*No runs yet.*") # ── Benchmark ─────────────────────────────────────── # with gr.Accordion("Evaluation harness", open=False): gr.Markdown( "Run the 5 built-in benchmark tasks and log results to W&B. " "Uses the currently selected model." ) benchmark_btn = gr.Button("▶ Run benchmark", variant="secondary") benchmark_output = gr.Markdown(value="") # ── Right column: full-height pipeline visualiser ─────── # with gr.Column(scale=1, elem_id="pipeline_col"): pipeline_vis = gr.HTML( value = get_pipeline_html(), label = "Pipeline status", elem_id = "pipeline_vis", ) # ── Event handlers ────────────────────────────────────────── # # Example dropdown — fill task input, skip separator items def _pick_example(val): if val and not val.startswith("──") and not val.startswith("Select an example"): return val, gr.update() # fill task, keep dropdown return gr.update(), gr.update(value="Select an example task ▼") # ignore, reset to placeholder example_dropdown.change( fn = _pick_example, inputs = [example_dropdown], outputs = [task_input, example_dropdown], ) # Unpack HITL components for wiring hitl_panel = hitl_components["panel"] hitl_error_disp = hitl_components["error_display"] hitl_options = hitl_components["options_radio"] hitl_editor = hitl_components["code_editor"] hitl_apply_btn = hitl_components["apply_btn"] hitl_resub_btn = hitl_components["resubmit_btn"] clarification_panel = clarification_components["panel"] clarification_q_md = clarification_components["question_md"] clarification_input = clarification_components["answer_input"] clarification_submit = clarification_components["submit_btn"] # Main run button — 21 outputs (15 normal + 2 clarification + 4 HITL) run_outputs = [ code_output, test_output, explain_output, flowchart_output, fc_zoom_bar, reflection_md, stats_md, history_md, status_bar, session_state, pipeline_vis, language_radio, detect_badge, test_summary_md, download_btn, # Clarification panel (shown when task is unclear) clarification_panel, clarification_q_md, # HITL panel components (shown only on AWAITING_HUMAN) hitl_panel, hitl_error_disp, hitl_options, hitl_editor, ] run_event = run_btn.click( fn = run_agent, inputs = [task_input, language_radio, model_radio, session_state], outputs = run_outputs, queue = True, ) # HITL resubmit — uses edited code + chosen option to re-run pipeline hitl_events = [] for hitl_btn in (hitl_resub_btn, hitl_apply_btn): evt = hitl_btn.click( fn = resubmit_agent, inputs = [hitl_editor, hitl_options, task_input, language_radio, model_radio, session_state], outputs = run_outputs, queue = True, ) hitl_events.append(evt) # Clarification submit — re-runs pipeline with refined task clarify_event = clarification_submit.click( fn = submit_clarification, inputs = [clarification_input, task_input, language_radio, model_radio, session_state], outputs = run_outputs, queue = True, ) stop_btn.click( fn = lambda: (get_pipeline_html(), "Stopped."), inputs = [], outputs = [pipeline_vis, status_bar], cancels = [run_event, clarify_event] + hitl_events, ) # Clear button — resets all outputs and state to initial values clear_outputs = [ task_input, code_output, test_output, explain_output, flowchart_output, fc_zoom_bar, reflection_md, stats_md, history_md, status_bar, session_state, pipeline_vis, detect_badge, test_summary_md, download_btn, example_dropdown, clarification_panel, clarification_q_md, hitl_panel, hitl_error_disp, hitl_options, hitl_editor, ] clear_btn.click( fn = lambda: ( "", # task_input "", # code_output "", # test_output "_No explanation generated yet._", # explain_output "_No flowchart generated yet._", # flowchart_output gr.update(visible=False), # fc_zoom_bar "", # reflection_md "", # stats_md "*No runs yet.*", # history_md "Ready.", # status_bar [], # session_state get_pipeline_html(), # pipeline_vis (reset to idle) gr.update(visible=False, value=""), # detect_badge gr.update(visible=False, value=""), # test_summary_md gr.update(visible=False, value=None), # download_btn gr.update(value="Select an example task ▼"), # example_dropdown gr.update(visible=False), # clarification_panel gr.update(value=""), # clarification_q_md gr.update(visible=False), # hitl_panel gr.update(value=""), # hitl_error_disp gr.update(value=None), # hitl_options gr.update(value=""), # hitl_editor ), inputs = [], outputs = clear_outputs, ) # Benchmark benchmark_btn.click( fn = run_benchmark, inputs = [model_radio], outputs = [benchmark_output, status_bar], ) return demo # ------------------------------------------------------------------ # # Entry point # # ------------------------------------------------------------------ # if __name__ == "__main__": demo = build_ui() demo.launch( server_name = "0.0.0.0", server_port = 7860, share = False, js = _fc_js, theme = gr.themes.Soft(), css = """ body, .gradio-container, .main, .app, svelte-scoped, #root { max-width: 100% !important; width: 100% !important; padding-left: 0 !important; padding-right: 0 !important; margin-left: 0 !important; margin-right: 0 !important; } .gradio-container { padding: 12px !important; box-sizing: border-box !important; } footer { display: none !important; } /* ── Mermaid v11 dark theme ── */ /* SVG canvas — constrain width so TD layout renders vertically */ .mermaid { background: #0d1117 !important; border-radius: 10px !important; display: flex !important; justify-content: center !important; overflow-x: auto !important; } .mermaid svg { background: #0d1117 !important; border-radius: 10px !important; width: 300px !important; max-width: 300px !important; height: auto !important; display: block !important; } /* All node shapes: rect (process), polygon (diamond), circle, path */ .mermaid .node rect, .mermaid .node circle, .mermaid .node ellipse, .mermaid .node polygon, .mermaid .node path, .mermaid .node .basic, .mermaid .node .label-container, .mermaid .cluster rect { fill: #1e3a5f !important; stroke: #4a90d9 !important; stroke-width: 1.5px !important; } /* v11 uses foreignObject > div > .nodeLabel for label text */ .mermaid .nodeLabel, .mermaid .nodeLabel p, .mermaid .node .label, .mermaid foreignObject div { color: #e2e8f0 !important; fill: #e2e8f0 !important; font-family: system-ui, sans-serif !important; font-size: 13px !important; } /* SVG text fallback (older sub-graphs, titles) */ .mermaid text, .mermaid text tspan { fill: #e2e8f0 !important; } /* Edge lines */ .mermaid .edgePath path, .mermaid .flowchart-link, .mermaid path.path { stroke: #4a90d9 !important; stroke-width: 2px !important; } /* Arrowheads */ .mermaid .arrowheadPath, .mermaid marker path, .mermaid .arrowMarkerPath { fill: #4a90d9 !important; stroke: #4a90d9 !important; } /* Edge labels */ .mermaid .edgeLabel, .mermaid .edgeLabel rect, .mermaid .edgeLabel .label { background: #1e293b !important; fill: #1e293b !important; color: #94a3b8 !important; } .mermaid .edgeLabel span, .mermaid .edgeLabel p { color: #94a3b8 !important; background: #1e293b !important; } /* ── Output tabs: natural height when empty, capped when full ── */ #output_tabs { max-height: 350px !important; overflow: hidden !important; display: flex !important; flex-direction: column !important; } /* Tab button bar stays fixed */ #output_tabs > .tab-nav { flex-shrink: 0 !important; } /* Each tab pane scrolls when content overflows */ #output_tabs > .tabitem, #output_tabs .tabitem > div { overflow-y: auto !important; max-height: 300px !important; } /* ── Cap code/test editors when content fills in ── */ #code_output .cm-editor, #test_output .cm-editor { max-height: 260px !important; overflow-y: auto !important; } /* ── Explanation & Flowchart markdown scroll ── */ #explain_output, #flowchart_output { max-height: 200px !important; overflow-y: auto !important; } /* ── Inline label + control rows (Language / Model / Examples) ── */ #lang_row, #model_row, #examples_row { align-items: center !important; gap: 8px !important; margin-bottom: 4px !important; } #lang_row > *, #model_row > *, #examples_row > * { margin-bottom: 0 !important; } /* Label markdown: fixed narrow width, no extra padding */ #lang_row > .prose, #model_row > .prose, #examples_row > .prose { min-width: 80px !important; max-width: 80px !important; flex-shrink: 0 !important; padding: 0 !important; margin: 0 !important; } /* ── Collapse test summary when empty (no gap above Generated tests) ── */ #test_summary_md:empty, #test_summary_md .prose:empty, #test_summary_md > .block:empty { display: none !important; } #test_summary_md { min-height: 0 !important; padding: 0 !important; margin: 0 !important; } /* ── Language detection badge ── */ #detect_badge { min-height: 0 !important; padding: 0 !important; margin: 2px 0 4px 0 !important; } #detect_badge .prose { font-size: 12px !important; color: #22C55E !important; padding: 0 !important; margin: 0 !important; } /* ── Pipeline column: natural width, no forced height ── */ #pipeline_col { min-width: 0; } """, )