diff --git a/apps/admin_console/database/repositories/session_repository.py b/apps/admin_console/database/repositories/session_repository.py index a3937e7..07c59cb 100644 --- a/apps/admin_console/database/repositories/session_repository.py +++ b/apps/admin_console/database/repositories/session_repository.py @@ -332,6 +332,9 @@ class SessionRepository: conn.commit() return count except Exception: + # Reconciliation is best-effort at startup, but a silent abort + # would leave every orphaned "running" row untouched -- log it. + logger.warning("Orphan-session reconciliation aborted", exc_info=True) return 0 def cleanup_orphans_on_startup(self) -> int: diff --git a/apps/admin_console/services/media_service.py b/apps/admin_console/services/media_service.py index 7b8940c..e5ef8c6 100644 --- a/apps/admin_console/services/media_service.py +++ b/apps/admin_console/services/media_service.py @@ -156,7 +156,13 @@ class MediaService: if not d.exists(): continue try: - for item in d.iterdir(): + entries = list(d.iterdir()) + except OSError as exc: + logger.warning("Video index scan failed under %s: %s", d, exc) + continue + for item in entries: + # One unreadable entry must not abort the rest of the directory. + try: if item.is_dir(): for ext in [".mp4", ".mkv", ".webm"]: vfile = item / f"recording{ext}" @@ -172,9 +178,8 @@ class MediaService: url = cls.path_to_video_url(item) idx[item.stem] = url idx[item.name] = url - except OSError as exc: - # Unreadable directory entry: index the rest of the tree. - logger.warning("Video index scan failed under %s: %s", d, exc) + except OSError as exc: + logger.warning("Skipping unindexable video entry %s: %s", item, exc) return idx @classmethod diff --git a/apps/admin_console/services/task_queue_service.py b/apps/admin_console/services/task_queue_service.py index 9881052..a9c4bcf 100644 --- a/apps/admin_console/services/task_queue_service.py +++ b/apps/admin_console/services/task_queue_service.py @@ -588,26 +588,46 @@ class TaskQueueService: manual_stop, ) if should_persist: - session_repo.update_session_status(sess_id, new_status, time.time()) - print(f"[QueueWorker] Updated session {sess_id} status to '{new_status}'") + if session_repo.update_session_status(sess_id, new_status, time.time()): + print(f"[QueueWorker] Updated session {sess_id} status to '{new_status}'") + else: + # The DB row is the fallback MCP pollers reconcile + # against when status.json is stale, so a failed DB + # write must not pass silently. + logger.error( + "[QueueWorker] Could not persist terminal DB status " + "'%s' for session %s", + new_status, + sess_id, + ) else: print( f"[QueueWorker] Preserved authoritative session {sess_id} " f"status '{new_status}'" ) - try: - if trace_store.read_status(str(sess_id)): - canonical_mcp_status = "completed" if new_status in ("completed", "success") else new_status - trace_store.update_trace_status( - str(sess_id), - canonical_mcp_status, + # A stale status.json would leave MCP pollers seeing "running" + # until their next DB reconcile, so retry transient write + # failures before giving up. + for attempt in range(3): + try: + if trace_store.read_status(str(sess_id)): + canonical_mcp_status = "completed" if new_status in ("completed", "success") else new_status + trace_store.update_trace_status( + str(sess_id), + canonical_mcp_status, + ) + break + except OSError as exc: + if attempt < 2: + await asyncio.sleep(0.5 * (attempt + 1)) + continue + logger.error( + "[QueueWorker] Could not persist terminal MCP status " + "for %s after %d attempts: %s", + sess_id, + attempt + 1, + exc, ) - except OSError as exc: - # A stale status.json would leave MCP pollers seeing - # "running" forever, so surface the failure. - print( - f"[QueueWorker] Could not persist terminal MCP status for {sess_id}: {exc}" - ) rec_info = session_repo.get_video_recording_for_session(sess_id) rec_status = (rec_info or {}).get("status") recovered_video_path = None diff --git a/mcp_server/notifiers/agentapi.py b/mcp_server/notifiers/agentapi.py index 8ca6a15..324e66f 100644 --- a/mcp_server/notifiers/agentapi.py +++ b/mcp_server/notifiers/agentapi.py @@ -172,7 +172,8 @@ class AgentApiNotifier(BaseNotifier): env["ANTIGRAVITY_CSRF_TOKEN"], ) ) - except (psutil.AccessDenied, psutil.NoSuchProcess, Exception): + except psutil.Error: + # This process vanished or denies access: scan the rest. continue except ImportError: import glob diff --git a/mcp_server/tools/task_manager.py b/mcp_server/tools/task_manager.py index d38be80..878d853 100644 --- a/mcp_server/tools/task_manager.py +++ b/mcp_server/tools/task_manager.py @@ -58,23 +58,31 @@ def _pid_alive(pid: int | None) -> bool: return pid_is_alive(pid_int) -def _session_tracked_by_lock(trace_id: str) -> bool: - """True when the device-lock layer still tracks this session (queued or active owner).""" +def _session_tracked_by_lock(trace_id: str) -> bool | None: + """Whether the device-lock layer still tracks this session (queued or active owner). + + Returns ``True`` when tracked, ``False`` when both probes succeeded and the + session is definitively absent, and ``None`` when a probe failed — an + unreadable queue or lock dir may hide a live session, so the caller must + not treat a failed probe as evidence of death (same "uncertain defaults to + alive" stance as ``pid_is_alive``). + """ + uncertain = False try: for q_item in DeviceExecutionLock.get_queued_tasks(): if str(q_item.get("session_id")) == trace_id: return True except Exception as exc: - # An unreadable queue may hide a live session; the caller then leans - # on DB/pid evidence, so surface the probe failure. + uncertain = True print(f"Could not read device queue for {trace_id}: {exc}", file=sys.stderr) try: for owner in DeviceExecutionLock.get_active_owners().values(): if owner.session_id and str(owner.session_id) == trace_id: return True except Exception as exc: + uncertain = True print(f"Could not read device owners for {trace_id}: {exc}", file=sys.stderr) - return False + return None if uncertain else False def _reconcile_task_state( @@ -154,8 +162,11 @@ def _reconcile_task_state( else: is_alive = _pid_alive(pid) if not is_alive: - if _session_tracked_by_lock(trace_id): - # Still queued for a device or owned by a live lock holder. + tracked = _session_tracked_by_lock(trace_id) + if tracked or tracked is None: + # Still queued for a device or owned by a live lock holder -- + # or the lock probe failed, in which case the session may be + # hidden and must not be declared dead on this poll. is_alive = True elif db_status in ("running", "pending") and db_pid: # The runner registered itself and its process is gone: truly dead. diff --git a/mcp_server/tools/task_runner.py b/mcp_server/tools/task_runner.py index 90f66a1..627cb2c 100644 --- a/mcp_server/tools/task_runner.py +++ b/mcp_server/tools/task_runner.py @@ -302,9 +302,24 @@ def mobile_run_task( "stderr_log": stderr_log, } - # Check if the task was already enqueued despite a slow or None response: - try: - queued_tasks = DeviceExecutionLock.get_queued_tasks() + # Check if the task was already enqueued despite a slow or None + # response. The probe reads a shared lock file, so transient + # read failures get a few retries; if it never succeeds the + # verdict is "unknown" -- reporting "failed" here for a task + # that did land in the queue would invite a duplicate submit. + probe_failed = False + for probe_attempt in range(3): + try: + queued_tasks = DeviceExecutionLock.get_queued_tasks() + except Exception as exc: + if probe_attempt < 2: + time.sleep(0.5) + continue + probe_failed = True + logging.getLogger("mcp_server").warning( + f"Could not verify daemon queue for '{trace_id}': {exc}" + ) + break for q in queued_tasks: if str(q.get("session_id")) == str(trace_id): trace_store.update_trace_status( @@ -323,12 +338,24 @@ def mobile_run_task( "stdout_log": str(trace_store.get_trace_stdout_log_path(trace_id)), "stderr_log": str(trace_store.get_trace_stderr_log_path(trace_id)), } - except Exception as exc: - # The queue probe is a rescue path for a slow daemon reply; - # its failure downgrades the verdict to "failed" below. - logging.getLogger("mcp_server").warning( - f"Could not verify daemon queue for '{trace_id}': {exc}" - ) + break + + if probe_failed: + # The task may or may not have been enqueued; do not mark + # the trace failed and do not let the caller resubmit + # blindly. + return { + "trace_id": trace_id, + "status": "unknown", + "message": ( + f"Artemis Daemon gave no enqueue confirmation for task " + f"'{task_desc}' and the queue could not be inspected. The " + f"task may still have been enqueued. Poll " + f"mobile_manage_task(action='status', trace_id='{trace_id}') " + f"before resubmitting; if it still reports 'pending' after " + f"about a minute, treat it as not enqueued." + ), + } # When Daemon is running, refuse to launch a conflicting standalone runner on the same device logging.getLogger("mcp_server").warning(