# ./app.py """ This UI layer sets up a clean initialization screen. It uses a layout state logic gate: when the page is loaded, it shows a clean login dialog box. If a valid password is sent, the complete history is queried; if an empty string or incorrect password is submitted, history retrieval is bypassed entirely, and the main workspace is unlocked securely without global state bleeding. """ import gradio as gr from core_logic import chat_function from storage import save_chat, load_history, get_chat_content, get_secret_password from git_agent import manage_github_repo import os """ print("==================================================") print(f"DEBUG: HF_TOKEN exists? {bool(os.getenv('HF_TOKEN'))}") print(f"DEBUG: APP_PASSWORD exists? {bool(os.getenv('APP_PASSWORD'))}") if os.getenv('APP_PASSWORD'): print(f"DEBUG: APP_PASSWORD length is {len(os.getenv('APP_PASSWORD'))}") print("==================================================") """ with gr.Blocks() as demo: chat_id_state = gr.State("") staged_files_state = gr.State([]) # Session state to securely isolate the user's validated password string per browser tab instance user_session_password = gr.State("") # ==================== LAYER 1: AUTHENTICATION GATE ==================== with gr.Column(visible=True) as login_layout: gr.Markdown("## 🔐 CoderG Enterprise Access Gate") gr.Markdown( "⚠️ **Workspace Notice:** Please leave the password field blank and press the **Unlock Workspace Environment** to proceed directly to the workspace." ) password_input = gr.Textbox( label="Security Access Password", placeholder="Enter password or leave blank for unauthenticated mode...", type="password" ) login_btn = gr.Button("Unlock Workspace Environment", variant="primary") # ==================== LAYER 2: MAIN WORKSPACE ==================== with gr.Row(visible=False) as main_workspace: # --- Left Panel: Sidebar History --- with gr.Column(scale=1, variant="secondary"): gr.Markdown("### 🛠️ Silicon Architect") new_btn = gr.Button("➕ New Chat", variant="primary") history_list = gr.Dataset( components=[gr.Textbox(visible=False)], label="Recent Conversations", samples=[], type="values", samples_per_page=20 ) # --- Center Panel: Main Core Multimodal Chat --- with gr.Column(scale=3): #chatbot = gr.Chatbot(show_label=False, height=700) active_chat_indicator = gr.Markdown("### 📝 Active Session: *New Conversation*") chatbot = gr.Chatbot(show_label=False, height=660) # Tweak height slightly to give header room chat_input = gr.Textbox( interactive=True, placeholder="Discuss architecture, paste code blocks, or ask CoderG to produce course documentation...", show_label=False, lines=1, max_lines=10, scale=8, submit_btn=False ) upload_btn = gr.UploadButton( "📎 Attach Documents/Images", file_count="multiple", file_types=[".png", ".jpg", ".jpeg", ".bmp", ".pdf", ".xlsx", ".xls", ".doc", ".docx", ".md", ".py", ".html", ".css", ".js", ".json", ".csv", ".zip", ".tar.gz", ".log", ".txt"], scale=2 ) upload_status = gr.Markdown("") # --- Right Panel: Agentic Control Tower --- with gr.Column(scale=1, variant="secondary"): gr.Markdown("### 🚀 CoderG Authorization Core") gr.Markdown("_Authorize code outputs to be compiled into dedicated remote repositories._") target_repo = gr.Textbox( label="Target Repository Name", placeholder="e.g., advanced-python-course", value="dynamic-course-repo" ) commit_txt = gr.Textbox( label="Commit Message", value="Automated generation via CoderG Agent" ) staged_files = gr.Textbox( label="Staged Files (Comma-separated)", value="COURSE_README.md" ) approve_btn = gr.Button("Approve & Push to GitHub", variant="primary") gr.Markdown("#### 📊 Deployment Telemetry Logs") output_log = gr.Markdown("_Awaiting local environment staging completion..._") # --- UI ROUTING HANDLERS --- def process_login_validation(password_attempt): """Step 1: Authenticates strings and updates layout toggles and browser session state.""" target_password = get_secret_password() clean_attempt = str(password_attempt).strip() if password_attempt is not None else "" # Explicit Guardrail: Raise an error only if they typed an incorrect password. if target_password and clean_attempt and clean_attempt != target_password: raise gr.Error("❌ Invalid security token entered. Access to environment denied.") # Returns layout visibility frames alongside the locked-in session password state string return gr.update(visible=False), gr.update(visible=True), clean_attempt def populate_history_component(session_password): """Step 2: Updates dataset samples based cleanly on isolated session token values.""" # Executes your requested logic: passing session_password down into the storage mechanism loaded_samples = load_history(user_password=session_password) return gr.update(samples=loaded_samples) # --- CORE WORKSPACE LOGIC --- def handle_file_upload(uploaded_files, current_staged_files): if not current_staged_files: current_staged_files = [] for file_obj in uploaded_files: file_path = file_obj.name if hasattr(file_obj, 'name') else file_obj if file_path and file_path not in current_staged_files: current_staged_files.append(file_path) status_msg = f"🟢 **{len(current_staged_files)} file(s) staged successfully and attached to next prompt.**" return current_staged_files, status_msg def bot_response(message, history, chat_id): user_content = message["text"] clean_history_snapshot = list(history) history.append({"role": "user", "content": user_content}) history.append({"role": "assistant", "content": ""}) for partial_resp in chat_function(message, clean_history_snapshot): history[-1]["content"] = partial_resp yield history """ def handle_save(history, chat_id, session_password): new_id = save_chat(chat_id, history) # Keeps sidebar refreshes tied strictly to the current session token authentication context current_list = load_history(user_password=session_password) if [new_id] not in current_list: current_list.insert(0, [new_id]) return new_id, gr.update(samples=current_list) """ def handle_save(history, chat_id, session_password): new_id = save_chat(chat_id, history) # Keeps sidebar refreshes tied strictly to the current session token authentication context current_list = load_history(user_password=session_password) if [new_id] not in current_list: current_list.insert(0, [new_id]) # We return the new_id, the dataset update, AND we update the indicator view # in case a brand new chat just generated its first true cloud file name. return new_id, gr.update(samples=current_list), f"### 📝 Active Session: `{new_id}`" """ def load_past_chat(selected_list): chat_id = selected_list[0] content = get_chat_content(chat_id) return content, chat_id """ def load_past_chat(selected_list): chat_id = selected_list[0] content = get_chat_content(chat_id) # Returns chat log array, updates chat ID tracking state, and updates the header text return content, chat_id, f"### 📝 Active Session: `{chat_id}`" def push_authorized(repo_name, commit_msg, files_list): files = [f.strip() for f in files_list.split(",") if f.strip()] if not repo_name.strip(): yield "❌ **Deployment Aborted:** Repository name cannot be empty." return yield "◌ _Connecting to GitHub REST API Engine..._" result = manage_github_repo(repo_name.strip(), commit_msg, files) yield f"{result}" # ==================== BIND EVENT LISTENER LIFECYCLES ==================== # Decoupled sequence: Click validates credentials and captures token -> THEN queries data array conditionally login_btn.click( fn=process_login_validation, inputs=[password_input], outputs=[login_layout, main_workspace, user_session_password] ).then( fn=populate_history_component, inputs=[user_session_password], outputs=[history_list] ) upload_btn.upload( fn=handle_file_upload, inputs=[upload_btn, staged_files_state], outputs=[staged_files_state, upload_status] ) def process_submission(message_text, current_staged_files, history, chat_id): if not message_text.strip() and not current_staged_files: return history, "", current_staged_files, "" payload = {"text": message_text, "files": current_staged_files} for updated_history in bot_response(payload, history, chat_id): yield updated_history, "", [], "" chat_input.submit( fn=process_submission, inputs=[chat_input, staged_files_state, chatbot, chat_id_state], outputs=[chatbot, chat_input, staged_files_state, upload_status] ).then( fn=handle_save, inputs=[chatbot, chat_id_state, user_session_password], outputs=[chat_id_state, history_list, active_chat_indicator] # Updates header text post-save ) history_list.click( fn=load_past_chat, inputs=[history_list], outputs=[chatbot, chat_id_state, active_chat_indicator] # Feeds selected chat ID into header ) """ new_btn.click( fn=lambda session_pass: ([], "", [], load_history(user_password=session_pass), "", "_Awaiting local environment staging completion..._"), inputs=[user_session_password], outputs=[chatbot, chat_id_state, staged_files_state, history_list, upload_status, output_log] ) """ new_btn.click( fn=lambda session_pass: ( [], "", [], load_history(user_password=session_pass), "", "_Awaiting local environment staging completion..._", "### 📝 Active Session: *New Conversation*" ), inputs=[user_session_password], outputs=[chatbot, chat_id_state, staged_files_state, history_list, upload_status, output_log, active_chat_indicator] ) approve_btn.click( fn=push_authorized, inputs=[target_repo, commit_txt, staged_files], outputs=[output_log] ) demo.launch(theme=gr.themes.Soft(), css="styles.css")