Stabilize task execution and UI lifecycle

apps/admin_console/core/state.py

@@ -28,7 +28,9 @@ class ServerState:
     def __init__(self):
         self.ipc_subscribers: list[Callable[[str, Any], None]] = []
         self.ipc_server: asyncio.Server | None = None
+        self.ipc_serve_task: asyncio.Task | None = None
         self.ipc_port: int | None = None
+        self.is_shutting_down: bool = False
 
         self.current_process: asyncio.subprocess.Process | None = None
         self.current_goal: str | None = None
@@ -41,6 +43,7 @@ class ServerState:
         # Unified single source of truth for task queue
         self.queue_items: list[dict[str, Any]] = []
         self._wake_event: asyncio.Event | None = None
+        self._shutdown_event: asyncio.Event | None = None
         self.worker_task: asyncio.Task | None = None
 
     @property
@@ -58,6 +61,22 @@ class ServerState:
             self._wake_event = asyncio.Event()
         return self._wake_event
 
+    @property
+    def shutdown_event(self) -> asyncio.Event:
+        """Event set as soon as the HTTP server receives a shutdown signal."""
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+
+        if self._shutdown_event is None or (
+            loop
+            and getattr(self._shutdown_event, "_loop", None) is not None
+            and self._shutdown_event._loop != loop
+        ):
+            self._shutdown_event = asyncio.Event()
+        return self._shutdown_event
+
     @property
     def task_queue(self) -> list[dict[str, Any]]:
         """Backward compatibility alias for queue_items."""
@@ -131,6 +150,19 @@ class ServerState:
     def is_paused(self) -> bool:
         return PAUSE_FILE.exists()
 
+    @property
+    def paused_error(self) -> str | None:
+        """Return the persisted pause reason for clients that missed the SSE event."""
+        if not PAUSE_FILE.exists():
+            return None
+        try:
+            message = PAUSE_FILE.read_text(encoding="utf-8", errors="replace").strip()
+        except OSError:
+            return "AI model request failed. The task is paused."
+        if message.startswith("LLM Error: "):
+            message = message[len("LLM Error: ") :]
+        return message or "AI model request failed. The task is paused."
+
     def add_subscriber(self, callback: Callable[[str, Any], None]):
         if callback not in self.ipc_subscribers:
             self.ipc_subscribers.append(callback)

apps/admin_console/database/repositories/step_repository.py

@@ -206,6 +206,48 @@ class StepRepository:
 
         return result_payload
 
+    @staticmethod
+    def _extract_legacy_pause_error(payload_raw: Any) -> str | None:
+        """Extract an LLM error from the warning trace used by older runners."""
+        payload_obj = payload_raw
+        if isinstance(payload_raw, str):
+            try:
+                payload_obj = json.loads(payload_raw)
+            except Exception:
+                return None
+        if not isinstance(payload_obj, dict):
+            return None
+
+        message = payload_obj.get("message")
+        if not isinstance(message, str) or "LLM Error:" not in message:
+            return None
+        pause_marker = ". Pausing execution"
+        if pause_marker not in message:
+            return None
+
+        error = message.split("LLM Error:", 1)[1].split(pause_marker, 1)[0].strip()
+        return error or None
+
+    def _normalize_display_trace(self, trace_dict: dict[str, Any]) -> dict[str, Any]:
+        """Normalize traces consumed by the task stream's tool/error cards."""
+        if trace_dict.get("type") == "log":
+            error = self._extract_legacy_pause_error(trace_dict.get("payload"))
+            if error:
+                trace_dict.update(
+                    {
+                        "type": "llm_call",
+                        "name": "llm_pause",
+                        "status": "failed",
+                        "payload": {"error": error, "pause": True},
+                    }
+                )
+                return trace_dict
+
+        trace_dict["payload"] = self._clean_tool_payload(
+            trace_dict.get("payload"), trace_dict.get("type")
+        )
+        return trace_dict
+
     def get_session_steps(self, session_id: str, client: str | None = None) -> list[dict[str, Any]]:
         with db_session(self.db_path) as conn:
             cursor = conn.cursor()
@@ -249,17 +291,18 @@ class StepRepository:
                         " t1.payload, t2.name as agent_name FROM traces t1 LEFT"
                         " JOIN traces t2 ON t1.parent_trace_id = t2.trace_id WHERE"
                         " t1.step_id = ? AND (t1.type = 'tool' OR t1.type = 'action' OR (t1.type ="
-                        " 'llm_call' AND t1.status = 'failed')) ORDER BY"
+                        " 'llm_call' AND t1.status = 'failed') OR (t1.type = 'log' AND"
+                        " t1.payload LIKE '%LLM Error:%Pausing execution%' AND NOT EXISTS"
+                        " (SELECT 1 FROM traces t3 WHERE t3.session_id = t1.session_id AND"
+                        " t3.type = 'llm_call' AND t3.status = 'failed' AND"
+                        " ABS(t3.timestamp - t1.timestamp) < 2))) ORDER BY"
                         " t1.timestamp ASC",
                         (step_id,),
                     )
                     trace_rows = cursor.fetchall()
                     generic_tools = []
                     for tr in trace_rows:
-                        trace_dict = dict(tr)
-                        trace_dict["payload"] = self._clean_tool_payload(
-                            trace_dict.get("payload"), trace_dict.get("type")
-                        )
+                        trace_dict = self._normalize_display_trace(dict(tr))
                         generic_tools.append(trace_dict)
                     step_dict["generic_tools"] = generic_tools
 
@@ -320,7 +363,11 @@ class StepRepository:
                 " t2.name as agent_name FROM traces t1 LEFT JOIN traces t2 ON"
                 " t1.parent_trace_id = t2.trace_id WHERE t1.session_id = ? AND"
                 " (t1.step_id IS NULL OR t1.step_id = '') AND (t1.type = 'tool' OR"
-                " (t1.type = 'llm_call' AND t1.status = 'failed')) ORDER BY"
+                " (t1.type = 'llm_call' AND t1.status = 'failed') OR (t1.type = 'log'"
+                " AND t1.payload LIKE '%LLM Error:%Pausing execution%' AND NOT EXISTS"
+                " (SELECT 1 FROM traces t3 WHERE t3.session_id = t1.session_id AND"
+                " t3.type = 'llm_call' AND t3.status = 'failed' AND"
+                " ABS(t3.timestamp - t1.timestamp) < 2))) ORDER BY"
                 " t1.timestamp ASC",
                 (session_id,),
             )
@@ -330,10 +377,7 @@ class StepRepository:
                 sorted_steps = sorted(steps, key=lambda s: s.get("timestamp") or 0)
 
                 for tr in stepless_rows:
-                    trace_dict = dict(tr)
-                    trace_dict["payload"] = self._clean_tool_payload(
-                        trace_dict.get("payload"), trace_dict.get("type")
-                    )
+                    trace_dict = self._normalize_display_trace(dict(tr))
 
                     tr_time = trace_dict.get("timestamp") or 0
                     target_step = None

apps/admin_console/routers/tasks.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import asyncio
+from contextlib import suppress
 import json
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import StreamingResponse
@@ -148,6 +149,7 @@ async def get_status():
         is_paused = state.is_paused
         return {
             "status": "paused" if is_paused else "running",
+            "paused_error": state.paused_error if is_paused else None,
             "goal": running_goal,
             "pid": state.current_process.pid if state.current_process else None,
             "session_id": running_sid,
@@ -163,6 +165,7 @@ async def get_status():
         conn_model_info = model_service.get_active_model_info(conn_profile)
         return {
             "status": "paused" if is_paused else "running",
+            "paused_error": state.paused_error if is_paused else None,
             "goal": conn_info.get("goal"),
             "pid": conn_info.get("pid"),
             "session_id": latest_session_id,
@@ -184,6 +187,7 @@ async def get_status():
 async def stream_events(session_id: str, client: str | None = None):
     async def event_generator():
         queue = asyncio.Queue()
+        event_loop = asyncio.get_running_loop()
 
         def callback(event_type, data):
             try:
@@ -210,22 +214,57 @@ async def stream_events(session_id: str, client: str | None = None):
                             return
 
                 sanitized_data = ipc_service.sanitize_event_data(event_type, data)
-                loop = asyncio.get_running_loop()
-                loop.call_soon_threadsafe(queue.put_nowait, (event_type, sanitized_data))
+                event_loop.call_soon_threadsafe(queue.put_nowait, (event_type, sanitized_data))
             except RuntimeError:
                 pass
 
         state.add_subscriber(callback)
         yield f'event: info\ndata: {{"message": "Subscribed to session {session_id}"}}\n\n'
 
+        shutdown_waiter = asyncio.create_task(state.shutdown_event.wait())
+        queue_waiter = None
         try:
-            while True:
-                try:
-                    event_type, data = await asyncio.wait_for(queue.get(), timeout=5.0)
+            while not state.is_shutting_down:
+                queue_waiter = asyncio.create_task(queue.get())
+                done, _ = await asyncio.wait(
+                    {queue_waiter, shutdown_waiter},
+                    timeout=5.0,
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+                if shutdown_waiter in done:
+                    queue_waiter.cancel()
+                    with suppress(asyncio.CancelledError):
+                        await queue_waiter
+                    queue_waiter = None
+                    break
+                if queue_waiter in done:
+                    event_type, data = queue_waiter.result()
+                    queue_waiter = None
                     yield f"event: {event_type}\ndata: {json.dumps(data, default=str)}\n\n"
-                except TimeoutError:
+                else:
+                    queue_waiter.cancel()
+                    with suppress(asyncio.CancelledError):
+                        await queue_waiter
+                    queue_waiter = None
                     yield "event: keep-alive\ndata: {}\n\n"
         except asyncio.CancelledError:
+            raise
+        finally:
+            waiters = (queue_waiter, shutdown_waiter)
+            for waiter in waiters:
+                if waiter is not None and not waiter.done():
+                    waiter.cancel()
+            for waiter in waiters:
+                if waiter is not None:
+                    with suppress(asyncio.CancelledError):
+                        await waiter
             state.remove_subscriber(callback)
 
-    return StreamingResponse(event_generator(), media_type="text/event-stream")
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache, no-transform",
+            "X-Accel-Buffering": "no",
+        },
+    )

apps/admin_console/server.py

@@ -20,7 +20,9 @@ Modular entrypoint for full trace inspection, step replay, and task execution ma
 import asyncio
 import os
 from pathlib import Path
+import signal
 import sys
+from types import FrameType
 
 # Bootstrap sys.path to allow running from any CWD
 _current_p = Path(__file__).resolve().parent
@@ -43,6 +45,9 @@ for _p in (str(_workspace_root), str(_apps_dir), str(_admin_console_dir), str(_c
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, HTMLResponse
+import uvicorn
+
+from artemis.runtime import DeviceExecutionLock
 
 try:
     from admin_console.core.config import (
@@ -97,6 +102,11 @@ app.add_middleware(
 @app.on_event("startup")
 async def on_startup():
     """Startup lifecycle hooks."""
+    state.is_shutting_down = False
+    state.shutdown_event.clear()
+    cleaned_device_locks = DeviceExecutionLock.cleanup_stale_locks()
+    if cleaned_device_locks:
+        print(f"[ServerStartup] Removed {cleaned_device_locks} stale device lock(s).")
     asyncio.create_task(asyncio.to_thread(task_queue_service.archive_older_replays_on_launch))
     asyncio.create_task(
         asyncio.to_thread(task_queue_service.verify_chunks_exist_on_launch, replay_manager)
@@ -110,6 +120,39 @@ async def on_startup():
     state.worker_task = asyncio.create_task(task_queue_service.queue_worker())
 
 
+@app.on_event("shutdown")
+async def on_shutdown():
+    """Stop task and IPC children before the UI server exits."""
+    state.is_shutting_down = True
+    state.shutdown_event.set()
+    task_queue_service._broadcast_event("server_shutdown", {"status": "stopping"})
+    owned_session_ids = {
+        str(item["session_id"])
+        for item in state.queue_items
+        if isinstance(item, dict) and item.get("status") == "running" and item.get("session_id")
+    }
+
+    worker = state.worker_task
+    if worker is not None and not worker.done():
+        worker.cancel()
+        try:
+            await asyncio.wait_for(worker, timeout=5.0)
+        except (asyncio.CancelledError, TimeoutError):
+            pass
+    state.worker_task = None
+
+    if state.current_process is not None and state.current_process.returncode is None:
+        await task_queue_service._terminate_worker_process(state.current_process)
+    DeviceExecutionLock.cleanup_stale_locks()
+    state.current_process = None
+    state.queue_items.clear()
+    for session_id in owned_session_ids:
+        session_repo.update_session_status(session_id, "cancelled")
+
+    await ipc_service.stop_server()
+    state.ipc_subscribers.clear()
+
+
 # Mount modular routers
 app.include_router(stream.router)
 app.include_router(media.router)
@@ -278,11 +321,80 @@ task_queue = state.task_queue
 queue_goals = state.queue_goals
 
 
-def main():
-    import uvicorn
+class ArtemisUvicornServer(uvicorn.Server):
+    """Notify application streams before Uvicorn waits for them to close."""
 
+    @staticmethod
+    def _task_is_active() -> bool:
+        """Check task state without doing process I/O from a signal handler."""
+        proc = state.current_process
+        if proc is not None and proc.returncode is None:
+            return True
+        return any(
+            isinstance(item, dict) and item.get("status") == "running" for item in state.queue_items
+        )
+
+    @staticmethod
+    def _schedule_signal_report(message: str) -> None:
+        """Defer console output so a signal cannot re-enter Colorama writes."""
+        try:
+            asyncio.get_running_loop().call_soon(print, message)
+        except RuntimeError:
+            try:
+                os.write(2, f"{message}\n".encode(errors="replace"))
+            except OSError:
+                pass
+
+    def handle_exit(self, sig: int, frame: FrameType | None) -> None:
+        try:
+            signal_name = signal.Signals(sig).name
+        except ValueError:
+            signal_name = "UNKNOWN"
+        frame_location = "unknown"
+        if frame is not None:
+            frame_location = f"{frame.f_code.co_filename}:{frame.f_lineno}"
+        signal_report = (
+            f"[ServerSignal] Received {signal_name} ({sig}); "
+            f"server PID={os.getpid()}, parent PID={os.getppid()}, frame={frame_location}"
+        )
+        self._schedule_signal_report(signal_report)
+
+        if sys.platform == "win32" and sig == signal.SIGINT and self._task_is_active():
+            self._schedule_signal_report(
+                "[ServerSignal] Ignored Windows SIGINT while a mobile task is active. "
+                "Stop the task first, or press Ctrl+Break to force server shutdown."
+            )
+            return
+
+        state.is_shutting_down = True
+        state.shutdown_event.set()
+        super().handle_exit(sig, frame)
+
+
+def run_ui_server(host: str, port: int, reload: bool = False) -> None:
+    """Run the UI server with bounded, signal-aware graceful shutdown."""
+    if reload:
+        uvicorn.run(
+            app,
+            host=host,
+            port=port,
+            reload=True,
+            timeout_graceful_shutdown=5,
+        )
+        return
+
+    config = uvicorn.Config(
+        app,
+        host=host,
+        port=port,
+        timeout_graceful_shutdown=5,
+    )
+    ArtemisUvicornServer(config).run()
+
+
+def main():
     port = int(os.environ.get("ANTIGRAVITY_SIDECAR_WEB_PORT", 8000))
-    uvicorn.run(app, host="0.0.0.0", port=port)
+    run_ui_server(host="0.0.0.0", port=port)
 
 
 if __name__ == "__main__":

apps/admin_console/services/ipc_service.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import asyncio
+from contextlib import suppress
 import json
 from typing import Any
 
@@ -134,73 +135,77 @@ class IPCService:
     async def start_server(cls):
         async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
             current_session_id = None
-            while True:
-                line = await reader.readline()
-                if not line:
-                    break
-                try:
-                    payload = json.loads(line.decode("utf-8"))
-                    event_type = payload.get("event_type")
-                    data = payload.get("data")
+            try:
+                while not state.is_shutting_down:
+                    line = await reader.readline()
+                    if not line:
+                        break
+                    try:
+                        payload = json.loads(line.decode("utf-8"))
+                        event_type = payload.get("event_type")
+                        data = payload.get("data")
 
-                    if event_type == "session_started" and data:
-                        incoming_sid = data.get("session_id")
-                        if incoming_sid and str(incoming_sid) in getattr(
-                            state, "cancelled_session_ids", set()
-                        ):
+                        if event_type == "session_started" and data:
+                            incoming_sid = data.get("session_id")
+                            if incoming_sid and str(incoming_sid) in getattr(
+                                state, "cancelled_session_ids", set()
+                            ):
+                                print(
+                                    f"[IPC] Ignoring session_started for cancelled session: {incoming_sid}"
+                                )
+                                continue
+
+                            state.active_session_id = incoming_sid
+                            current_session_id = incoming_sid
+                            pid = data.get("pid")
+                            goal = data.get("initial_goal")
+                            device_info = data.get("device_info")
+                            if isinstance(device_info, dict) and device_info.get("profile"):
+                                state.current_profile = device_info.get("profile")
+                            if goal:
+                                state.current_goal = goal
                             print(
-                                f"[IPC] Ignoring session_started for cancelled session: {incoming_sid}"
+                                f"[IPC] Session started: {state.active_session_id} (profile: {state.current_profile})"
                             )
-                            continue
+                            if current_session_id:
+                                state.active_connections[str(current_session_id)] = {
+                                    "writer": writer,
+                                    "pid": pid,
+                                    "goal": goal,
+                                    "profile": state.current_profile,
+                                }
+                        elif event_type == "session_ended" and data:
+                            ended_sid = data.get("session_id")
+                            if state.active_session_id and str(state.active_session_id) == str(
+                                ended_sid
+                            ):
+                                state.active_session_id = None
+                                state.current_goal = None
+                                state.current_profile = None
 
-                        state.active_session_id = incoming_sid
-                        current_session_id = incoming_sid
-                        pid = data.get("pid")
-                        goal = data.get("initial_goal")
-                        device_info = data.get("device_info")
-                        if isinstance(device_info, dict) and device_info.get("profile"):
-                            state.current_profile = device_info.get("profile")
-                        if goal:
-                            state.current_goal = goal
-                        print(
-                            f"[IPC] Session started: {state.active_session_id} (profile: {state.current_profile})"
-                        )
-                        if current_session_id:
-                            state.active_connections[str(current_session_id)] = {
-                                "writer": writer,
-                                "pid": pid,
-                                "goal": goal,
-                                "profile": state.current_profile,
-                            }
-                    elif event_type == "session_ended" and data:
-                        ended_sid = data.get("session_id")
-                        if state.active_session_id and str(state.active_session_id) == str(
-                            ended_sid
-                        ):
-                            state.active_session_id = None
-                            state.current_goal = None
-                            state.current_profile = None
-
-                    for cb in list(state.ipc_subscribers):
-                        try:
-                            cb(event_type, data)
-                        except Exception:
-                            pass
-                except Exception as e:
-                    print(f"IPC parse error: {e}")
-
-            if current_session_id and str(current_session_id) in state.active_connections:
-                try:
-                    del state.active_connections[str(current_session_id)]
-                except KeyError:
-                    pass
-            writer.close()
-            await writer.wait_closed()
+                        for cb in list(state.ipc_subscribers):
+                            try:
+                                cb(event_type, data)
+                            except Exception:
+                                pass
+                    except (ValueError, UnicodeDecodeError, TypeError) as e:
+                        print(f"IPC parse error: {e}")
+            except asyncio.CancelledError:
+                raise
+            except (ConnectionResetError, BrokenPipeError, OSError):
+                # Normal when a task or the UI server is shutting down on Windows.
+                pass
+            finally:
+                if current_session_id and str(current_session_id) in state.active_connections:
+                    state.active_connections.pop(str(current_session_id), None)
+                writer.close()
+                with suppress(ConnectionResetError, BrokenPipeError, OSError):
+                    await writer.wait_closed()
 
         server = await asyncio.start_server(handle_client, "127.0.0.1", 0, limit=1024 * 1024 * 100)
         state.ipc_port = server.sockets[0].getsockname()[1]
         state.ipc_server = server
-        asyncio.create_task(server.serve_forever())
+        state.ipc_serve_task = asyncio.create_task(server.serve_forever())
         print(f"Internal IPC server started on port {state.ipc_port}")
         try:
             from artemis.config import write_ipc_port
@@ -209,5 +214,34 @@ class IPCService:
         except Exception as e:
             print(f"Failed to write IPC port file: {e}")
 
+    @classmethod
+    async def stop_server(cls):
+        """Close IPC listeners and task connections during application shutdown."""
+        for connection in list(state.active_connections.values()):
+            writer = connection.get("writer")
+            if writer is not None:
+                writer.close()
+                with suppress(ConnectionResetError, BrokenPipeError, OSError):
+                    await writer.wait_closed()
+        state.active_connections.clear()
+
+        if state.ipc_server is not None:
+            state.ipc_server.close()
+            await state.ipc_server.wait_closed()
+            state.ipc_server = None
+
+        if state.ipc_serve_task is not None:
+            state.ipc_serve_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await state.ipc_serve_task
+            state.ipc_serve_task = None
+        state.ipc_port = None
+        try:
+            from artemis.config import clear_ipc_port
+
+            clear_ipc_port()
+        except Exception as e:
+            print(f"Failed to clear IPC port state: {e}")
+
 
 ipc_service = IPCService()

apps/admin_console/services/task_queue_service.py

@@ -16,6 +16,7 @@ import asyncio
 from datetime import datetime
 import os
 import shutil
+import subprocess
 import sys
 import time
 from typing import Any
@@ -76,6 +77,31 @@ class TaskQueueService:
             if not (isinstance(t, dict) and t.get("session_id") == session_id)
         ]
 
+    @staticmethod
+    def _subprocess_creation_kwargs() -> dict[str, int]:
+        """Keep task workers out of the UI server's Windows console group.
+
+        Without a new process group, a console control event intended for the
+        uvicorn server is also delivered to a task that is still importing
+        Artemis, causing STATUS_CONTROL_C_EXIT (0xC000013A).
+        """
+        if sys.platform == "win32":
+            return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP}
+        return {}
+
+    @staticmethod
+    async def _terminate_worker_process(proc: asyncio.subprocess.Process | None) -> None:
+        """Stop a worker without changing the established POSIX behavior."""
+        if proc is None or proc.returncode is not None:
+            return
+        if sys.platform == "win32":
+            await process_supervisor.stop_process(proc)
+            return
+        try:
+            proc.kill()
+        except ProcessLookupError:
+            pass
+
     @classmethod
     def ensure_worker_running(cls):
         """Guarantees that the background queue worker task is active and running."""
@@ -182,6 +208,7 @@ class TaskQueueService:
                     *cmd,
                     cwd=str(WORKSPACE_ROOT),
                     env=env,
+                    **cls._subprocess_creation_kwargs(),
                 )
                 state.current_process = proc
                 task_item["pid"] = proc.pid
@@ -193,10 +220,7 @@ class TaskQueueService:
                     print(
                         f"[QueueWorker] Task [{sess_id}] was cancelled during launch. Terminating."
                     )
-                    try:
-                        proc.kill()
-                    except Exception:
-                        pass
+                    await cls._terminate_worker_process(proc)
 
                 # 3. Await subprocess completion
                 returncode = await proc.wait()
@@ -225,10 +249,7 @@ class TaskQueueService:
             except asyncio.CancelledError:
                 print("[QueueWorker] Task received cancellation signal.")
                 if state.current_process and state.current_process.returncode is None:
-                    try:
-                        state.current_process.kill()
-                    except Exception:
-                        pass
+                    await cls._terminate_worker_process(state.current_process)
                 break
             except Exception as e:
                 print(f"[QueueWorker] Unexpected error executing task: {e}")