From 4927c4f0b9c678d9054a6131e5644838f6444553 Mon Sep 17 00:00:00 2001 From: somew1nd Date: Wed, 19 Aug 2026 00:47:45 -0700 Subject: [PATCH] Stabilize task execution and UI lifecycle --- apps/admin_console/core/state.py | 32 + .../database/repositories/step_repository.py | 64 +- apps/admin_console/routers/tasks.py | 53 +- apps/admin_console/server.py | 118 +- apps/admin_console/services/ipc_service.py | 152 +- .../services/task_queue_service.py | 37 +- .../agent-stream/agent-stream.component.html | 34 +- .../agent-stream/agent-stream.component.scss | 61 + .../agent-stream/agent-stream.component.ts | 56 +- .../chat-interface.component.html | 8 +- .../chat-interface.component.scss | 6 + .../chat-interface.component.ts | 14 +- .../src/app/core/models/session.model.ts | 5 +- .../src/app/services/agent.service.ts | 135 +- artemis/config/runtime.py | 3 +- artemis/core/diagnostics/emulator_manager.py | 18 +- artemis/data_engine/engine.py | 117 +- artemis/interfaces/cli/commands/mcp.py | 201 +- artemis/interfaces/cli/commands/ui.py | 47 +- artemis/runtime/__init__.py | 3 + artemis/runtime/device_lock.py | 189 + artemis/sdk/agent.py | 13 +- artemis/services/llm.py | 25 +- mcp_server/tools/task_manager.py | 14 +- .../admin_console/test_server_lifecycle.py | 138 + .../admin_console/test_step_repository.py | 45 + .../test_system_emulator_endpoints.py | 19 + .../admin_console/test_task_queue_service.py | 28 +- tests/unit/data_engine/test_engine.py | 86 +- tests/unit/runtime/test_device_lock.py | 110 + tests/unit/test_cli.py | 110 +- tests/unit/test_config_package.py | 5 +- tests/unit/test_llm_pause.py | 37 + tests/unit/test_ui_command.py | 41 + uv.lock | 4895 +++++++++++++++++ 35 files changed, 6680 insertions(+), 239 deletions(-) create mode 100644 artemis/runtime/device_lock.py create mode 100644 tests/unit/admin_console/test_server_lifecycle.py create mode 100644 tests/unit/admin_console/test_step_repository.py create mode 100644 tests/unit/runtime/test_device_lock.py create mode 100644 tests/unit/test_llm_pause.py create mode 100644 tests/unit/test_ui_command.py create mode 100644 uv.lock 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 @@
{{ session.start_time * 1000 | date:'shortTime' }} -
diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss index c598619..3da10c9 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.scss @@ -610,6 +610,12 @@ $uniform-gap: 8px; animation: pulse-black-dot-small 1.5s infinite ease-in-out; } } + &.paused { + background: #fff7ed; + color: #c2410c; + border-color: #fdba74; + &::before { background-color: #f97316; } + } &.completed { background: #fafafa; color: #1f1f1f; @@ -2336,6 +2342,61 @@ $uniform-gap: 8px; color: #ef4444; } } + + &.is-paused { + background: #fff7ed; + color: #c2410c; + border: 1px solid #fed7aa; + + .status-icon { + color: #f97316; + } + } + } + + &.llm-failure-header { + flex-wrap: wrap; + + .btn-retry-llm { + flex: 0 0 auto; + min-height: 26px; + padding: 3px 9px; + display: inline-flex; + align-items: center; + justify-content: center; + gap: 4px; + box-sizing: border-box; + border: 1px solid #bfdbfe; + border-radius: 7px; + background: #ffffff; + color: #2563eb; + font-family: inherit; + font-size: 11px; + font-weight: 700; + line-height: 1; + cursor: pointer; + box-shadow: 0 1px 2px rgba(15, 23, 42, 0.04); + transition: background 0.18s ease, border-color 0.18s ease, box-shadow 0.18s ease, transform 0.18s ease; + + .material-symbols-outlined { + font-size: 15px; + } + + &:hover { + background: #eff6ff; + border-color: #93c5fd; + box-shadow: 0 2px 6px rgba(37, 99, 235, 0.12); + } + + &:active { + transform: scale(0.97); + } + + &:focus-visible { + outline: 2px solid rgba(37, 99, 235, 0.35); + outline-offset: 2px; + } + } } .action-expand-icon { diff --git a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts index 1cc042f..83e71fd 100644 --- a/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts +++ b/apps/showcase_ui/src/app/components/agent-stream/agent-stream.component.ts @@ -168,13 +168,13 @@ export class AgentStreamComponent implements AfterViewInit { public activeQueue = computed(() => { const list = this.agentService.sessions().filter((s) => { const status = this.getTaskStatus(s); - return status === 'running' || status === 'pending'; + return status === 'running' || status === 'paused' || status === 'pending'; }); return list.sort((a, b) => { const statusA = this.getTaskStatus(a); const statusB = this.getTaskStatus(b); - if (statusA === 'running') return -1; - if (statusB === 'running') return 1; + if (statusA === 'running' || statusA === 'paused') return -1; + if (statusB === 'running' || statusB === 'paused') return 1; return a.start_time - b.start_time; }); }); @@ -244,6 +244,28 @@ export class AgentStreamComponent implements AfterViewInit { return groupBlocksToPhases(this.consolidatedBlocks(), currentSession?.start_time || 0); }); + private retryablePauseTrace = computed(() => { + if (!this.agentService.isPaused()) return null; + + const blocks = this.consolidatedBlocks(); + for (let blockIndex = blocks.length - 1; blockIndex >= 0; blockIndex--) { + const tools = blocks[blockIndex]?.data?.generic_tools; + if (!Array.isArray(tools)) continue; + + for (let toolIndex = tools.length - 1; toolIndex >= 0; toolIndex--) { + const tool = tools[toolIndex]; + if ( + tool?.type === 'llm_call' + && tool?.status === 'failed' + && tool?.payload?.pause === true + ) { + return tool; + } + } + } + return null; + }); + public hasVisibleBlocks(phase: any): boolean { if (!phase || !phase.blocks) return false; return phase.blocks.some((b: any) => this.hasVisibleContent(b)); @@ -447,13 +469,13 @@ export class AgentStreamComponent implements AfterViewInit { this.isNotesDropdownOpen.set(false); } - public getTaskStatus(session: Session): 'running' | 'completed' | 'pending' | 'failed' | 'cancelled' { - if (session.session_id === this.agentService.runningSessionId() && this.agentService.agentStatus() === 'running') { - return 'running'; + public getTaskStatus(session: Session): 'running' | 'paused' | 'completed' | 'pending' | 'failed' | 'cancelled' { + if (session.session_id === this.agentService.runningSessionId() && (this.agentService.agentStatus() === 'running' || this.agentService.agentStatus() === 'paused')) { + return this.agentService.agentStatus() as 'running' | 'paused'; } if (session.status) { const s = session.status.toLowerCase(); - if (s === 'running' || s === 'completed' || s === 'pending' || s === 'failed' || s === 'cancelled') { + if (s === 'running' || s === 'paused' || s === 'completed' || s === 'pending' || s === 'failed' || s === 'cancelled') { return s as any; } } @@ -819,20 +841,12 @@ export class AgentStreamComponent implements AfterViewInit { } public isTerminalLLMFailure(tool: any): boolean { - if (!tool) return false; - for (const block of this.consolidatedBlocks()) { - if (block?.data?.generic_tools) { - const tools = block.data.generic_tools; - const failedLLMs = tools.filter((t: any) => t && t.type === 'llm_call' && t.status === 'failed'); - if (failedLLMs.length > 0) { - const lastFailed = failedLLMs[failedLLMs.length - 1]; - if (lastFailed && lastFailed.trace_id === tool.trace_id) { - return this.agentService.isPaused() || block.id === this.consolidatedBlocks()[this.consolidatedBlocks().length - 1]?.id; - } - } - } - } - return false; + return !!tool && this.retryablePauseTrace() === tool; + } + + public retryPausedTask(event: Event): void { + event.stopPropagation(); + this.agentService.resumeTask(); } public getLLMErrorText(tool: any): string { diff --git a/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html b/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html index 9f1535d..8c569f2 100644 --- a/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html +++ b/apps/showcase_ui/src/app/components/chat-interface/chat-interface.component.html @@ -78,8 +78,8 @@
{{ session.start_time * 1000 | date:'shortTime' }} -