adding a user approval step to jobs api + ruff formatting

agent/core/agent_loop.py

@@ -15,6 +15,16 @@ from agent.core.tools import ToolRouter
 ToolCall = ChatCompletionMessageToolCall
 
 
+def _needs_approval(tool_name: str, tool_args: dict) -> bool:
+    """Check if a tool call requires user approval before execution"""
+    if tool_name != "hf_jobs":
+        return False
+
+    # Check if it's a run or uv operation
+    operation = tool_args.get("operation", "")
+    return operation in ["run", "uv"]
+
+
 class Handlers:
     """Handler functions for each operation type"""
 
@@ -33,9 +43,10 @@ class Handlers:
 
             Laminar.set_trace_session_id(session_id=session.session_id)
 
-        # Add user message to history
-        user_msg = Message(role="user", content=text)
-        session.context_manager.add_message(user_msg)
+        # Add user message to history only if there's actual content
+        if text:
+            user_msg = Message(role="user", content=text)
+            session.context_manager.add_message(user_msg)
 
         # Send event that we're processing
         await session.send_event(
@@ -97,6 +108,28 @@ class Handlers:
                     tool_name = tc.function.name
                     tool_args = json.loads(tc.function.arguments)
 
+                    # Check if this tool requires user approval
+                    if _needs_approval(tool_name, tool_args):
+                        await session.send_event(
+                            Event(
+                                event_type="approval_required",
+                                data={
+                                    "tool": tool_name,
+                                    "arguments": tool_args,
+                                    "tool_call_id": tc.id,
+                                },
+                            )
+                        )
+
+                        # Store pending approval and return early
+                        session.pending_approval = {
+                            "tool_call": tc,
+                            "arguments": tool_args,
+                        }
+
+                        # Return early - wait for EXEC_APPROVAL operation
+                        return None
+
                     await session.send_event(
                         Event(
                             event_type="tool_call",
@@ -191,6 +224,85 @@ class Handlers:
 
         await session.send_event(Event(event_type="undo_complete"))
 
+    @staticmethod
+    async def exec_approval(
+        session: Session, approved: bool, feedback: str | None = None
+    ) -> None:
+        """Handle job execution approval"""
+        if not session.pending_approval:
+            await session.send_event(
+                Event(
+                    event_type="error",
+                    data={"error": "No pending approval to process"},
+                )
+            )
+            return
+
+        tc = session.pending_approval["tool_call"]
+        tool_args = session.pending_approval["arguments"]
+        tool_name = tc.function.name
+
+        if approved:
+            # Execute the pending tool
+            await session.send_event(
+                Event(
+                    event_type="tool_call",
+                    data={"tool": tool_name, "arguments": tool_args},
+                )
+            )
+
+            output, success = await session.tool_router.call_tool(tool_name, tool_args)
+
+            # Add tool result to context
+            tool_msg = Message(
+                role="tool",
+                content=output,
+                tool_call_id=tc.id,
+                name=tool_name,
+            )
+            session.context_manager.add_message(tool_msg)
+
+            await session.send_event(
+                Event(
+                    event_type="tool_output",
+                    data={
+                        "tool": tool_name,
+                        "output": output,
+                        "success": success,
+                    },
+                )
+            )
+        else:
+            # User rejected - add cancellation message to context
+            cancellation_msg = "Job execution cancelled by user"
+            if feedback:
+                cancellation_msg += f". User feedback: {feedback}"
+
+            tool_msg = Message(
+                role="tool",
+                content=cancellation_msg,
+                tool_call_id=tc.id,
+                name=tool_name,
+            )
+            session.context_manager.add_message(tool_msg)
+
+            await session.send_event(
+                Event(
+                    event_type="tool_output",
+                    data={
+                        "tool": tool_name,
+                        "output": cancellation_msg,
+                        "success": False,
+                    },
+                )
+            )
+
+        # Clear pending approval
+        session.pending_approval = None
+
+        # Continue agent loop with empty input to process the tool result
+        await Handlers.run_agent(session, "")
+
     @staticmethod
     async def shutdown(session: Session) -> bool:
         """Handle shutdown (like shutdown in codex.rs:1329)"""
@@ -226,6 +338,12 @@ async def process_submission(session: Session, submission) -> bool:
         await Handlers.undo(session)
         return True
 
+    if op.op_type == OpType.EXEC_APPROVAL:
+        approved = op.data.get("approved", False) if op.data else False
+        feedback = op.data.get("feedback") if op.data else None
+        await Handlers.exec_approval(session, approved, feedback)
+        return True
+
     if op.op_type == OpType.SHUTDOWN:
         return not await Handlers.shutdown(session)
 

agent/core/session.py

@@ -53,6 +53,7 @@ class Session:
         )
         self.is_running = True
         self.current_task: asyncio.Task | None = None
+        self.pending_approval: Optional[dict[str, Any]] = None
 
     async def send_event(self, event: Event) -> None:
         """Send event back to client"""

agent/main.py

@@ -47,10 +47,13 @@ class Submission:
 
 async def event_listener(
     event_queue: asyncio.Queue,
+    submission_queue: asyncio.Queue,
     turn_complete_event: asyncio.Event,
     ready_event: asyncio.Event,
 ) -> None:
     """Background task that listens for events and displays them"""
+    submission_id = [1000]  # Use list to make it mutable in closure
+
     while True:
         try:
             event = await event_queue.get()
@@ -96,6 +99,69 @@ async def event_listener(
                 old_tokens = event.data.get("old_tokens", 0) if event.data else 0
                 new_tokens = event.data.get("new_tokens", 0) if event.data else 0
                 print(f"📦 Compacted context: {old_tokens} → {new_tokens} tokens")
+            elif event.event_type == "approval_required":
+                # Display job details and prompt for approval
+                tool_name = event.data.get("tool", "") if event.data else ""
+                arguments = event.data.get("arguments", {}) if event.data else {}
+
+                print("\n" + "=" * 60)
+                print("⚠️  JOB EXECUTION APPROVAL REQUIRED")
+                print("=" * 60)
+
+                operation = arguments.get("operation", "")
+                args = arguments.get("args", {})
+
+                print(f"Operation: {operation}")
+
+                if operation == "uv":
+                    script = args.get("script", "")
+                    dependencies = args.get("dependencies", [])
+                    print(f"Script to run:\n{script}")
+                    if dependencies:
+                        print(f"Dependencies: {', '.join(dependencies)}")
+                elif operation == "run":
+                    image = args.get("image", "")
+                    command = args.get("command", "")
+                    print(f"Docker image: {image}")
+                    print(f"Command: {command}")
+
+                # Common parameters
+                flavor = args.get("flavor", "cpu-basic")
+                detached = args.get("detached", False)
+                print(f"Hardware: {flavor}")
+                print(f"Detached mode: {detached}")
+
+                secrets = args.get("secrets", [])
+                if secrets:
+                    print(f"Secrets: {', '.join(secrets)}")
+
+                print("=" * 60)
+
+                # Get user decision
+                loop = asyncio.get_event_loop()
+                response = await loop.run_in_executor(
+                    None,
+                    input,
+                    "Approve? (y=yes, n=no, or provide feedback to reject): ",
+                )
+
+                response = response.strip()
+                approved = response.lower() in ["y", "yes"]
+                feedback = (
+                    None if approved or response.lower() in ["n", "no"] else response
+                )
+
+                # Submit approval
+                submission_id[0] += 1
+                approval_submission = Submission(
+                    id=f"approval_{submission_id[0]}",
+                    operation=Operation(
+                        op_type=OpType.EXEC_APPROVAL,
+                        data={"approved": approved, "feedback": feedback},
+                    ),
+                )
+                await submission_queue.put(approval_submission)
+                print("=" * 60 + "\n")
             # Silently ignore other events
 
         except asyncio.CancelledError:
@@ -145,7 +211,7 @@ async def main():
 
     # Start event listener in background
     listener_task = asyncio.create_task(
-        event_listener(event_queue, turn_complete_event, ready_event)
+        event_listener(event_queue, submission_queue, turn_complete_event, ready_event)
     )
 
     # Wait for agent to initialize

agent/tools/jobs_tool.py

@@ -13,9 +13,12 @@ from huggingface_hub import HfApi
 from huggingface_hub.utils import HfHubHTTPError
 
 from agent.tools.types import ToolResult
-from agent.tools.utilities import (format_job_details, format_jobs_table,
-                                   format_scheduled_job_details,
-                                   format_scheduled_jobs_table)
+from agent.tools.utilities import (
+    format_job_details,
+    format_jobs_table,
+    format_scheduled_job_details,
+    format_scheduled_jobs_table,
+)
 
 # Hardware flavors
 CPU_FLAVORS = ["cpu-basic", "cpu-upgrade", "cpu-performance", "cpu-xl"]
@@ -446,7 +449,7 @@ To inspect, call this tool with `{{"operation": "inspect", "args": {{"job_id": "
 
             # Not detached - wait for completion and stream logs
             print(f"Job started: {job.id}")
-            print(f"Streaming logs...\n---\n")
+            print("Streaming logs...\n---\n")
 
             final_status, all_logs = await self._wait_for_job_completion(
                 job_id=job.id,
@@ -519,7 +522,7 @@ To check logs, call this tool with `{{"operation": "logs", "args": {{"job_id": "
 
             # Not detached - wait for completion and stream logs
             print(f"UV Job started: {job.id}")
-            print(f"Streaming logs...\n---\n")
+            print("Streaming logs...\n---\n")
 
             final_status, all_logs = await self._wait_for_job_completion(
                 job_id=job.id,