fix(scheduler): raise SchedulerError on empty status output instead of FAILED

src/hpc/cli.py

@@ -14,6 +14,7 @@
 from .sync import SyncManager
 from .job import JobManager, JobStatus
 from .run import RunManager
+from .scheduler import SchedulerError
 
 # Type alias for config option
 ConfigOption = Annotated[
@@ -345,7 +346,16 @@ def status(id: str = typer.Argument(None), config: ConfigOption = None):
     if detail is None or not detail.state:
         # Scheduler does not support detail (PJM), or sacct has not yet
         # recorded this job. Fall back to the existing single-line display.
-        job_status = job_manager.get_job_status(job_id)
+        # SchedulerError (parse-side absence) becomes a friendly message;
+        # other SSHError (real transport/command failure) propagates so
+        # legitimate failures stay visible to the user.
+        try:
+            job_status = job_manager.get_job_status(job_id)
+        except SchedulerError:
+            print(
+                f"Job {job_id}: status unavailable yet (scheduler accounting not ready)"
+            )
+            return
         print(f"Job {job_id}: {job_status.value}")
         return
 

src/hpc/scheduler.py

@@ -5,6 +5,31 @@
 from dataclasses import dataclass
 from enum import Enum
 
+from .ssh import SSHError
+
+
+class SchedulerError(SSHError):
+    """Scheduler returned no usable response.
+
+    Raised by ``parse_status`` when the scheduler's status command exits
+    cleanly but its output contains no data row from which a ``JobStatus``
+    can be derived — typically a freshly-submitted job that has not yet
+    been indexed by the accounting database (Slurm's ``sacct``) or a job
+    that aged out of the active view (PJM's ``pjstat`` post-``EXT``
+    window). Distinguishing this from a real ``JobStatus.FAILED`` lets
+    callers retry / fall through instead of treating the absence of
+    information as a terminal failure.
+
+    Inherits from ``SSHError`` so the three existing transient-catch
+    sites in ``JobManager`` cover it without modification:
+    ``wait_for_job``'s polling loop (continues), ``tail_job_output``'s
+    pre-tail status probe (falls through to ``tail -F``), and the inner
+    status probe inside ``get_job_output`` (swallows so the original
+    "No such file" diagnostic re-raises). CLI surfaces that want to
+    discriminate "no data yet" from a real SSH / command failure catch
+    ``SchedulerError`` specifically before the generic ``SSHError``.
+    """
+
 
 class JobStatus(Enum):
     PENDING = "PENDING"
@@ -149,7 +174,12 @@ def parse_status(self, output: str) -> JobStatus:
             if tokens:
                 lines.append(tokens[0].rstrip("+"))
         if not lines:
-            return JobStatus.FAILED
+            # No parseable row — sacct has not yet indexed the job (fresh
+            # submission, accounting lag). Surface as a transient absence
+            # rather than synthesizing a terminal FAILED.
+            raise SchedulerError(
+                "sacct returned no data row; job may not yet be indexed"
+            )
         statuses = [_STATUS_MAP.get(s, JobStatus.FAILED) for s in lines]
         for status in (
             JobStatus.RUNNING,
@@ -298,9 +328,9 @@ def parse_status(self, output: str) -> JobStatus:
         #     48969021   EXT 0   0
         # The first parseable row wins; multi-row aggregation for array
         # / step jobs is tracked separately (Issue #12) and out of scope
-        # here. An empty / header-only response falls back to ``FAILED``,
-        # the same fallback the pre-EC/SN parser used (Issue #8 owns
-        # the principled fix).
+        # here. An empty / header-only / short-row-only response raises
+        # ``SchedulerError`` so callers can distinguish "no data yet"
+        # from a real terminal failure.
         for raw in output.splitlines():
             tokens = raw.split()
             if len(tokens) < 4 or tokens[0] == "JOB_ID":
@@ -311,7 +341,10 @@ def parse_status(self, output: str) -> JobStatus:
                     JobStatus.COMPLETED if ec == "0" and sn == "0" else JobStatus.FAILED
                 )
             return self._STATUS_MAP.get(st, JobStatus.FAILED)
-        return JobStatus.FAILED
+        raise SchedulerError(
+            "pjstat returned no data row; job may not yet be indexed"
+            " or has aged out of the active view"
+        )
 
 
 # Slurm job state strings as reported by `sacct`. Unknown states fall back

tests/test_cli.py

@@ -139,6 +139,29 @@ def test_status_falls_back_when_detail_unavailable(cli_runner, temp_dir, monkeyp
     assert "ExitCode" not in result.stdout
 
 
+def test_status_prints_friendly_message_when_status_unavailable(
+    cli_runner, temp_dir, monkeypatch
+):
+    """SchedulerError on the fallback ``get_job_status`` becomes a friendly
+    "status unavailable yet" line instead of an SSHError stack."""
+    from hpc.scheduler import SchedulerError
+
+    monkeypatch.chdir(temp_dir)
+    cli_runner.invoke(app, ["init"])
+
+    with patch("hpc.cli.JobManager") as MockJobManager:
+        instance = MockJobManager.return_value
+        instance.get_job_detail.return_value = None
+        instance.get_job_status.side_effect = SchedulerError(
+            "sacct returned no data row"
+        )
+        result = cli_runner.invoke(app, ["status", "12345678"])
+
+    assert result.exit_code == 0
+    assert "Job 12345678: status unavailable yet" in result.stdout
+    assert "scheduler accounting not ready" in result.stdout
+
+
 def test_status_normalizes_decorated_cancelled_state(cli_runner, temp_dir, monkeypatch):
     """``CANCELLED+`` and ``CANCELLED by 12345`` should still trigger detail rendering."""
     from hpc.scheduler import JobDetail

tests/test_job.py

@@ -6,7 +6,7 @@
 import pytest
 
 from hpc.job import JobManager, JobStatus, _extract_prologue_directives
-from hpc.scheduler import JobDetail
+from hpc.scheduler import JobDetail, SchedulerError
 from hpc.ssh import SSHManager, SSHError
 from hpc.config import HpcConfig, ClusterConfig, EnvConfig, SlurmConfig, PjmConfig
 
@@ -290,6 +290,19 @@ def test_get_job_status_uses_sacct(self, mock_ssh_manager, sample_config):
         assert "12345678" in call_args.args[1]
         assert "--noheader" in call_args.args[1]
 
+    def test_get_job_status_raises_scheduler_error_on_empty_output(
+        self, mock_ssh_manager, sample_config
+    ):
+        # Boundary pin: the SchedulerError raised by parse_status reaches
+        # JobManager's public API unchanged (no conversion to a different
+        # type). Since SchedulerError is an SSHError subclass, the three
+        # transient-handling callers still catch it via `except SSHError`.
+        manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config)
+        mock_ssh_manager.run_command.return_value = MagicMock(stdout="")
+
+        with pytest.raises(SchedulerError):
+            manager.get_job_status("12345678")
+
 
 class TestJobManagerDetail:
     def test_get_job_detail_slurm_invokes_sacct_and_returns_detail(
@@ -428,6 +441,26 @@ def test_pjm_error_flag_uses_err_extension(self, mock_ssh_manager):
         call_args = mock_ssh_manager.run_command.call_args
         assert call_args.args[1] == ["/scratch/user/proj/.hpc/runs/test_run/job.err"]
 
+    def test_inner_scheduler_error_does_not_mask_original_no_such_file(
+        self, mock_ssh_manager, sample_config
+    ):
+        # Regression guard: when ``cat`` raises "No such file" and the
+        # inner status probe raises SchedulerError (job not yet indexed),
+        # the existing ``except SSHError: pass`` swallows the inner
+        # SchedulerError (subclass of SSHError) and re-raises the original
+        # cat error. The user must see the original missing-file
+        # diagnostic, not the inner accounting-not-ready one.
+        manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config)
+        mock_ssh_manager.run_command.side_effect = [
+            SSHError("SSH command failed: cat: No such file or directory"),
+            MagicMock(stdout=""),  # status probe; parse_status raises SchedulerError
+        ]
+
+        with pytest.raises(SSHError) as exc_info:
+            manager.get_job_output("test_run", "12345678")
+        assert "No such file" in str(exc_info.value)
+        assert not isinstance(exc_info.value, SchedulerError)
+
 
 class TestJobManagerTailJobOutput:
     def _running_status(self):

tests/test_scheduler.py

@@ -1,6 +1,9 @@
 """Scheduler unit tests."""
 
-from hpc.scheduler import PJM, JobDetail, JobStatus, Slurm
+import pytest
+
+from hpc.scheduler import PJM, JobDetail, JobStatus, SchedulerError, Slurm
+from hpc.ssh import SSHError
 
 
 class TestSlurmStatusCmd:
@@ -97,15 +100,18 @@ def test_aggregation_strips_plus_on_every_row(self):
         output = "COMPLETED+\nCOMPLETED+\nCOMPLETED+\n"
         assert Slurm().parse_status(output) == JobStatus.COMPLETED
 
-    # Empty / unknown — empty output preserves the pre-existing FAILED
-    # behavior; unknown states fall back to FAILED so we stay
+    # Empty / unknown — empty input raises SchedulerError so callers can
+    # distinguish "no data yet" (accounting lag) from a real terminal
+    # FAILED. Unknown but parseable states still map to FAILED to stay
     # conservative with respect to wait termination.
 
-    def test_empty_output(self):
-        assert Slurm().parse_status("") == JobStatus.FAILED
+    def test_empty_output_raises_scheduler_error(self):
+        with pytest.raises(SchedulerError):
+            Slurm().parse_status("")
 
-    def test_whitespace_only_output(self):
-        assert Slurm().parse_status("   \n  \n") == JobStatus.FAILED
+    def test_whitespace_only_output_raises_scheduler_error(self):
+        with pytest.raises(SchedulerError):
+            Slurm().parse_status("   \n  \n")
 
     def test_unknown_status_falls_back_to_failed(self):
         assert Slurm().parse_status("BOGUS_STATE\n") == JobStatus.FAILED
@@ -235,18 +241,18 @@ def test_ccl_maps_to_cancelled(self):
     def test_rjt_maps_to_failed(self):
         assert PJM().parse_status(self._row("RJT")) == JobStatus.FAILED
 
-    def test_empty_output_falls_back_to_failed(self):
-        # Regression guard for the deferred Issue #8: today, empty
-        # pjstat output is conflated with a true failure. The eventual
-        # fix should be a deliberate behavior change rather than an
-        # accidental one.
-        assert PJM().parse_status("") == JobStatus.FAILED
+    def test_empty_output_raises_scheduler_error(self):
+        # Empty pjstat output is a structural absence (accounting lag /
+        # aged-out job), not a real terminal failure; raise so callers
+        # treat it as transient.
+        with pytest.raises(SchedulerError):
+            PJM().parse_status("")
 
-    def test_header_only_output_falls_back_to_failed(self):
-        # Same deferred-#8 guard, header-only shape: ``pjstat`` exited
-        # cleanly but produced no data row (job aged out of the active
-        # view's post-EXT window).
-        assert PJM().parse_status(self._HEADER) == JobStatus.FAILED
+    def test_header_only_output_raises_scheduler_error(self):
+        # ``pjstat`` exited cleanly but produced no data row (job aged
+        # out of the active view's post-EXT window).
+        with pytest.raises(SchedulerError):
+            PJM().parse_status(self._HEADER)
 
     def test_multi_row_input_takes_first_parseable_row(self):
         # Documents today's single-row semantics so the eventual array
@@ -261,6 +267,22 @@ def test_short_row_skipped_before_data_row(self):
         output = f"{self._HEADER}short\n48971221   EXT 0   0\n"
         assert PJM().parse_status(output) == JobStatus.COMPLETED
 
+    def test_unknown_st_falls_back_to_failed(self):
+        # An unrecognized but structurally-valid ST token (4 data
+        # tokens, non-``EXT``) stays on the conservative FAILED
+        # fallback rather than raising — only structural absence
+        # raises SchedulerError.
+        assert PJM().parse_status(self._row("ZZZ")) == JobStatus.FAILED
+
+
+class TestSchedulerErrorInheritance:
+    def test_scheduler_error_is_ssh_error_subclass(self):
+        # The three callers in JobManager (wait_for_job, tail_job_output,
+        # get_job_output) catch SSHError to apply transient/unknown
+        # handling. SchedulerError must inherit so those existing
+        # handlers cover it without modification.
+        assert issubclass(SchedulerError, SSHError)
+
 
 class TestPJMParseJobID:
     def test_parse_job_id_prefers_job_token(self):

tests/test_wait.py

@@ -53,6 +53,25 @@ def test_wait_returns_on_failed(self, mock_ssh_manager, sample_config):
         status = manager.wait_for_job("12345678", interval=0.01)
         assert status == JobStatus.FAILED
 
+    def test_wait_keeps_polling_on_empty_status_output(
+        self, mock_ssh_manager, sample_config
+    ):
+        # Regression guard for https://github.com/ultimatile/hpc/issues/13:
+        # a freshly-submitted job whose sacct row hasn't been indexed yet
+        # returns empty stdout. The old parse_status mapped that to FAILED
+        # and wait_for_job exited prematurely; SchedulerError (subclass
+        # of SSHError) now routes through the retry path so polling
+        # continues.
+        manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config)
+        mock_ssh_manager.run_command.side_effect = [
+            MagicMock(stdout=""),
+            MagicMock(stdout="COMPLETED\n"),
+        ]
+
+        status = manager.wait_for_job("12345678", interval=0.01)
+        assert status == JobStatus.COMPLETED
+        assert mock_ssh_manager.run_command.call_count == 2
+
     def test_wait_adaptive_interval(self, mock_ssh_manager, sample_config):
         manager = JobManager(ssh_manager=mock_ssh_manager, config=sample_config)
         mock_ssh_manager.run_command.side_effect = [