feat(scheduler): properly track SCHEDULED state for batch jobs

- Add _set_job_state_from_process() to check process state and set job
  state to SCHEDULED or RUNNING accordingly
- Add aio_wait_until_running() to Process base class with polling default
- SLURM: implement event-driven aio_wait_until_running() using watcher
- Add clock icon (🕐) for SCHEDULED state in TUI
- Show informative message when trying to view logs for SCHEDULED jobs

This enables proper tracking of jobs that are submitted to batch systems
like SLURM but not yet executing. The TUI now shows a distinct icon and
prevents attempts to view logs that don't exist yet.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: bpiwowar

src/experimaestro/connectors/__init__.py

@@ -138,6 +138,28 @@ async def aio_isrunning(self):
         """True is the process is truly running (I/O)"""
         return (await self.aio_state()).running
 
+    async def aio_wait_until_running(
+        self, poll_interval: float = 5.0
+    ) -> "ProcessState":
+        """Wait until the process transitions from SCHEDULED to RUNNING (or finishes).
+
+        Subclasses (e.g., SLURM) may override with event-driven implementation.
+        Default implementation polls aio_state() at regular intervals.
+
+        Args:
+            poll_interval: How often to poll for state changes (seconds)
+
+        Returns:
+            The new ProcessState (RUNNING or a finished state)
+        """
+        import asyncio
+
+        while True:
+            state = await self.aio_state()
+            if state != ProcessState.SCHEDULED:
+                return state
+            await asyncio.sleep(poll_interval)
+
     async def aio_code(self) -> Optional[int]:
         """Returns a future containing the returned code
 

src/experimaestro/launchers/slurm/base.py

@@ -221,10 +221,14 @@ def __init__(self, launcher: "SlurmLauncher"):
         self.fetched_event = threading.Event()
         self.updating_jobs = threading.Lock()
 
-        # Async waiters: jobid -> list of (asyncio.Future, event_loop)
+        # Async waiters for job completion: jobid -> list of (asyncio.Future, event_loop)
         self.async_waiters: Dict[
             str, List[Tuple[asyncio.Future, asyncio.AbstractEventLoop]]
         ] = {}
+        # Async waiters for state change: jobid -> list of (asyncio.Future, event_loop)
+        self.async_state_waiters: Dict[
+            str, List[Tuple[asyncio.Future, asyncio.AbstractEventLoop]]
+        ] = {}
         self.async_waiters_lock = threading.Lock()
 
         # Async waiters for first fetch: list of (asyncio.Future, event_loop)
@@ -346,9 +350,24 @@ def register_async_waiter(
             self.async_waiters[jobid].append((event, loop))
         return event
 
+    def register_async_state_waiter(
+        self, jobid: str, loop: asyncio.AbstractEventLoop
+    ) -> asyncio.Future:
+        """Register an async waiter for job state change.
+
+        Returns an asyncio.Future that will be set on next state update.
+        """
+        future = loop.create_future()
+        with self.async_waiters_lock:
+            if jobid not in self.async_state_waiters:
+                self.async_state_waiters[jobid] = []
+            self.async_state_waiters[jobid].append((future, loop))
+        return future
+
     def _notify_async_waiters(self):
-        """Notify async waiters for finished jobs"""
+        """Notify async waiters for finished jobs and state changes"""
         with self.async_waiters_lock:
+            # Notify completion waiters
             finished_jobs = []
             for jobid, waiters in self.async_waiters.items():
                 state = self.jobs.get(jobid)
@@ -361,6 +380,13 @@ def _notify_async_waiters(self):
             for jobid in finished_jobs:
                 del self.async_waiters[jobid]
 
+            # Notify state change waiters (always notify with current state)
+            for jobid, waiters in list(self.async_state_waiters.items()):
+                state = self.jobs.get(jobid)
+                for future, loop in waiters:
+                    loop.call_soon_threadsafe(future.set_result, state)
+            self.async_state_waiters.clear()
+
     def _notify_async_fetched_waiters(self):
         """Notify async waiters waiting for the first fetch"""
         with self.async_waiters_lock:
@@ -524,6 +550,29 @@ async def aio_state(self, timeout: float | None = None) -> ProcessState:
             if release_after:
                 SlurmProcessWatcher.release(watcher)
 
+    async def aio_wait_until_running(self) -> ProcessState:
+        """Wait until the job transitions from SCHEDULED to RUNNING (or finishes).
+
+        Uses the SLURM watcher's event-driven notification instead of polling.
+        Returns the new ProcessState (RUNNING or a finished state).
+        """
+        loop = asyncio.get_running_loop()
+
+        with SlurmProcessWatcher.get(self.launcher) as watcher:
+            while True:
+                # Check current state
+                jobinfo = await watcher.aio_getjob(self.jobid)
+                if jobinfo:
+                    if jobinfo.state != ProcessState.SCHEDULED:
+                        return jobinfo.state
+                else:
+                    # Job not found yet, wait for next update
+                    pass
+
+                # Register for state change notification and wait
+                future = watcher.register_async_state_waiter(self.jobid, loop)
+                await future
+
     def kill(self):
         logger.warning("Killing slurm job %s", self.jobid)
         builder = self.launcher.connector.processbuilder()

src/experimaestro/scheduler/base.py

@@ -853,16 +853,45 @@ async def aio_final_state(self, job: Job) -> JobState:
         # Wait for done handler to complete and notify exit condition
         return job.state
 
+    async def _set_job_state_from_process(self, job: Job, process: "Process") -> None:
+        """Set job state based on process state and wait until running.
+
+        Checks the process state and sets job to SCHEDULED or RUNNING accordingly.
+        If SCHEDULED, waits until the process starts running (or finishes).
+
+        Args:
+            job: The job to update
+            process: The process to check state on
+        """
+        from experimaestro.connectors import ProcessState
+
+        # Check initial process state
+        state = await process.aio_state()
+        if state == ProcessState.SCHEDULED:
+            job.set_state(JobState.SCHEDULED)
+            self.notify_job_state(job)
+
+            # Wait until running or finished (uses event-driven for SLURM)
+            state = await process.aio_wait_until_running()
+            if state == ProcessState.RUNNING:
+                job.set_state(JobState.RUNNING)
+                self.notify_job_state(job)
+                logger.info("Job %s started running", job.identifier[:8])
+            # If finished, state will be set later from marker files
+        else:
+            # Process is already running (or finished)
+            job.set_state(JobState.RUNNING)
+            self.notify_job_state(job)
+
     async def _wait_for_job_process(self, job: Job, process: "Process") -> None:
         """Wait for a running job process to complete and update state.
 
         Args:
             job: The job with a running process
             process: The process to wait for
         """
-        # Notify listeners that job is running
-        job.set_state(JobState.RUNNING)
-        self.notify_job_state(job)
+        # Set initial state (SCHEDULED or RUNNING) and wait until running
+        await self._set_job_state_from_process(job, process)
 
         # And now, we wait...
         code = await process.aio_code()
@@ -1066,6 +1095,9 @@ async def aio_start(self, job: Job) -> Optional[JobState]:  # noqa: C901
                     logger.warning("Error while starting job", exc_info=True)
                     return JobState.ERROR
 
+            # Set initial state (SCHEDULED or RUNNING) and wait until running
+            await self._set_job_state_from_process(job, process)
+
             # Wait for job to complete while holding locks
             try:
                 logger.debug("Waiting for job %s process to end", job)

src/experimaestro/tui/app.py

@@ -621,6 +621,16 @@ def on_view_job_logs(self, message: ViewJobLogs) -> None:
         For remote monitoring, switches to log viewer immediately with loading state,
         then starts adaptive sync in background.
         """
+        from experimaestro.scheduler.interfaces import JobState
+
+        # Check if job is scheduled (not yet running) - no logs available
+        if message.job_state == JobState.SCHEDULED:
+            self.notify(
+                "Job is scheduled but not yet running - logs not available",
+                severity="information",
+            )
+            return
+
         job_path = Path(message.job_path)
         job_id = job_path.name
 

src/experimaestro/tui/utils.py

@@ -44,6 +44,8 @@ def get_status_icon(status: str, failure_reason=None, transient=None):
         return "❌"
     elif status == "running":
         return "▶"
+    elif status == "scheduled":
+        return "🕐"  # Scheduled (e.g., in SLURM queue)
     elif status == "waiting":
         return "⌛"  # Waiting for dependencies
     elif status == "unscheduled" and transient is not None and transient.is_transient: