diff --git a/apps/admin_console/core/state.py b/apps/admin_console/core/state.py index e4e73e5..96edea3 100644 --- a/apps/admin_console/core/state.py +++ b/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) diff --git a/apps/admin_console/database/repositories/step_repository.py b/apps/admin_console/database/repositories/step_repository.py index 71d0d7b..9d7f73a 100644 --- a/apps/admin_console/database/repositories/step_repository.py +++ b/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 diff --git a/apps/admin_console/routers/tasks.py b/apps/admin_console/routers/tasks.py index 4b7e818..774d616 100644 --- a/apps/admin_console/routers/tasks.py +++ b/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", + }, + ) diff --git a/apps/admin_console/server.py b/apps/admin_console/server.py index cd3ace9..eaefcf5 100644 --- a/apps/admin_console/server.py +++ b/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__": diff --git a/apps/admin_console/services/ipc_service.py b/apps/admin_console/services/ipc_service.py index a45b8e6..e1c3d08 100644 --- a/apps/admin_console/services/ipc_service.py +++ b/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() diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 516108c..8fe0bc6 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/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}") diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html index 11854d8..1f3dbb4 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.html @@ -69,8 +69,8 @@