fix(jobs): cloud log tail survives short replays and silent stalls (MT47)

The HF cloud log tail could stop delivering mid-run and never recover:
progress, charts and ETA froze while the run completed normally (status
polls independently; checkpoints list from the Hub). Hit live on runs
from 2026-07-30/31 and 2026-08-07 (110-min log gap on a completed run).
Introduced by a08ab03 (2026-05-12). Two independent defects, both
unit-reproduced:

- Reconnect accounting: a per-connection line index compared against a
  cross-connection counter silently drops every subsequent line forever
  once a reconnect replays fewer lines than already processed. Replaced
  with content-based bounded replay dedupe.
- No silence timeout: a stalled stream stranded the tail thread
  unobservably. Stream iteration now runs on a reader thread with a
  600s silence timeout raising into the existing reconnect path.

Mitigation: _settle_terminal_metrics snaps done runs to total_steps and
clears eta_seconds on all terminal states (never inventing progress for
failed/interrupted); the frontend gates the ETA label on active runs.
The existing wandb URL scrape is preserved through the rewrite and
pinned by a dedicated test.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Mokuroh54

frontend/src/components/training/monitoring/MonitoringStats.tsx

@@ -138,8 +138,13 @@ const MonitoringStats: React.FC<MonitoringStatsProps> = ({
   const stepLabel = isStarting
     ? "Training starting…"
     : `${trainingStatus.current_step.toLocaleString()} / ${trainingStatus.total_steps.toLocaleString()}`;
+  // Only a RUNNING job has time remaining. `eta_seconds` is the last value the
+  // log parser extrapolated, and it survives on the record after the run ends —
+  // so a finished job whose log stream died mid-flight (MT47) would otherwise
+  // render a confident countdown next to its "Done" badge, which is how a stale
+  // reading reads as a live one.
   const etaLabel =
-    trainingStatus.eta_seconds != null
+    trainingStatus.training_active && trainingStatus.eta_seconds != null
       ? formatTime(trainingStatus.eta_seconds)
       : "—";
 

makermodslab/jobs.py

@@ -312,6 +312,34 @@ def _initial_metrics(config: TrainingRequest) -> TrainingMetrics:
     return TrainingMetrics(current_step=start, total_steps=config.steps)
 
 
+def _settle_terminal_metrics(record: JobRecord) -> None:
+    """Reconcile a finished run's progress with the fact that it finished.
+
+    `metrics` only ever advances when a log line is parsed, so a run whose log
+    stream died mid-flight keeps the last frame it saw forever. That produced
+    the MT47 symptom: a `done` run rendering "3,650 / 10,000" beside a live
+    countdown ("00:53:05 remaining") while its step-10,000 checkpoint sat on the
+    Hub — three surfaces of one record disagreeing.
+
+    Two changes, both about not asserting what we no longer believe:
+
+      * `done` means the trainer reached its target, so progress is the target.
+        Only claimed when a target is actually known (`total_steps > 0`), and
+        only for `done` — a `failed`/`interrupted` run genuinely stopped where
+        the last frame said, and rounding that up to the target would invent
+        training that never happened.
+      * The ETA is cleared for EVERY terminal state. A finished run has no
+        remaining time, whatever the last frame extrapolated.
+
+    A mitigation, not the fix: the root cause is the log tail going silent
+    (MT47), and a repaired tail leaves this a no-op on a healthy run.
+    """
+    if record.metrics.eta_seconds is not None:
+        record.metrics.eta_seconds = None
+    if record.state == "done" and record.metrics.total_steps > 0:
+        record.metrics.current_step = record.metrics.total_steps
+
+
 def _read_log_metrics(path: Path, resume_total: int | None) -> builtins.list[MetricsHistoryPoint]:
     """Parse one job's log.jsonl into (step, loss, lr, grad_norm) points.
 
@@ -3262,6 +3290,7 @@ def _tick(self) -> None:
                 record.state = "done" if rc == 0 else "failed"
                 record.ended_at = time.time()
                 record.exit_code = rc
+                _settle_terminal_metrics(record)
                 if rc != 0 and record.error_message is None:
                     # Prefer a runner-supplied reason (e.g. HF Jobs'
                     # 'Job timeout') over the synthetic exit-code message.

makermodslab/runners/hf_cloud.py

@@ -34,6 +34,7 @@
 import threading
 import time
 import tomllib
+from collections import deque
 from importlib.metadata import requires
 from pathlib import Path
 from queue import Empty, Queue
@@ -574,6 +575,22 @@ def resolve_job_timeout(config: TrainingRequest) -> int | str:
 # (transient network blip during a long training).
 _TAIL_RECONNECT_BACKOFF_S = 5.0
 
+# How long a single connection may deliver NOTHING before we abandon it and
+# reconnect (MT47). `fetch_job_logs(follow=True)` can block inside one read
+# forever — no line, no StopIteration, no exception — and a plain `for` over it
+# has no way to notice. Generous on purpose: a job still QUEUED/BUILDING is
+# legitimately silent for minutes, and a needless reconnect is cheap now that
+# the replay is deduped by content rather than by position.
+_TAIL_SILENCE_TIMEOUT_S = 600.0
+
+# How many recently-emitted lines are remembered for replay de-duplication on
+# reconnect (MT47). Bounds memory while covering a realistic replay window: a
+# 2.5-hour cloud run's log.jsonl held ~275 lines. If a replay ever exceeds this,
+# the oldest lines fall out of the window and are re-emitted — duplicated log
+# lines, which is the DELIBERATE failure direction: the previous positional
+# scheme failed the other way and went permanently silent.
+_TAIL_DEDUPE_WINDOW = 1000
+
 
 def resolve_wandb_api_key() -> str | None:
     """Look up the host's wandb API key for forwarding to a cloud job.
@@ -629,9 +646,12 @@ def __init__(
         # registry can surface it to the UI instead of a synthetic exit code.
         self._terminal_message: str | None = None
         self._wandb_run_url: str | None = None
-        # Count of log lines processed across (possibly multiple) SSE
-        # connections, so reconnects skip the replayed prefix.
-        self._lines_processed: int = 0
+        # The most recently emitted log lines, for de-duplicating the prefix an
+        # SSE reconnect replays (MT47). Content, not position: `seen`-vs-total
+        # counting assumed every reconnect replays the whole log from line 1,
+        # and silently dropped every subsequent line whenever it didn't.
+        self._recent_lines: deque[str] = deque(maxlen=_TAIL_DEDUPE_WINDOW)
+        self._recent_line_set: set[str] = set()
 
     def start(self, job_id: str, config: TrainingRequest, output_dir: str) -> None:
         # output_dir is the host-local path the registry pins for local jobs;
@@ -827,30 +847,101 @@ def _ensure_dataset_on_hub(self, repo_id: str) -> None:
             raise RuntimeError(msg) from exc
         self._log_line(f"[upload] dataset {repo_id} uploaded.")
 
+    def _is_replayed(self, stripped: str) -> bool:
+        """Whether this line was already emitted, so a reconnect's replayed
+        prefix isn't teed to disk and the UI twice (MT47).
+
+        Content-based and bounded, deliberately replacing the positional
+        `seen <= _lines_processed` scheme this used to use. That scheme was only
+        correct if EVERY reconnect replayed the whole log from line 1; when a
+        reconnect replayed less than that (or nothing at all, following from
+        "now"), the per-connection counter never caught up with the
+        cross-connection total and every subsequent line was skipped — silently,
+        forever, while the job ran happily to completion.
+
+        The tradeoff runs the other way now: a line repeated legitimately within
+        the window is dropped, and a replay longer than the window is re-emitted.
+        Both are cosmetic. Going mute is not.
+        """
+        if stripped in self._recent_line_set:
+            return True
+        evicted = self._recent_lines[0] if len(self._recent_lines) == self._recent_lines.maxlen else None
+        self._recent_lines.append(stripped)
+        # deque(maxlen=…) drops the oldest on append; mirror that in the set,
+        # but only if the evicted text isn't still present later in the window.
+        if evicted is not None and evicted not in self._recent_lines:
+            self._recent_line_set.discard(evicted)
+        self._recent_line_set.add(stripped)
+        return False
+
+    def _iter_job_logs(self):
+        """Yield raw log lines, abandoning a connection that goes silent (MT47).
+
+        `fetch_job_logs(follow=True)` can block inside a single read
+        indefinitely — no line, no StopIteration, no exception — when the SSE
+        connection is half-open (NAT eviction, laptop sleep, proxy idle
+        timeout). A plain `for` over that iterator has no way to notice: it
+        cannot even observe `_stop_event`, because the loop body never runs.
+
+        So the blocking iteration happens on a reader thread and is consumed
+        through a queue with a timeout. On silence we raise, which the caller
+        already handles as "reconnect". The reader thread is abandoned rather
+        than joined — it is stuck in exactly the read we gave up on — but it is
+        a daemon and dies with the process. That is not a new leak: before this,
+        a stalled read stranded the whole tail loop the same way, and stranded
+        it permanently.
+        """
+        assert self._hf_job_id is not None
+        queue: Queue = Queue()
+        done = object()
+
+        def _reader() -> None:
+            try:
+                for raw in self._api.fetch_job_logs(job_id=self._hf_job_id, follow=True):
+                    queue.put(raw)
+                    if self._stop_event.is_set():
+                        break
+            except Exception as exc:  # surfaced on the consuming thread
+                queue.put(exc)
+            finally:
+                queue.put(done)
+
+        threading.Thread(target=_reader, name=f"hf-job-{self._hf_job_id}-sse", daemon=True).start()
+
+        while True:
+            try:
+                item = queue.get(timeout=_TAIL_SILENCE_TIMEOUT_S)
+            except Empty as exc:
+                raise TimeoutError(f"no log output for {_TAIL_SILENCE_TIMEOUT_S:.0f}s; reconnecting") from exc
+            if item is done:
+                return
+            if isinstance(item, BaseException):
+                raise item
+            yield item
+
     def _tail_loop(self) -> None:
         """Stream HfApi.fetch_job_logs, teeing each line to disk and the
-        in-memory queue. Reconnects on stream end or transient error while
-        the status poller still says the job is alive — SSE death is no
-        longer fatal. Exits when _stop_event is set (status poller saw a
-        terminal stage, or stop() was called).
+        in-memory queue. Reconnects on stream end, transient error, or a
+        silent connection while the status poller still says the job is alive
+        — SSE death is no longer fatal. Exits when _stop_event is set (status
+        poller saw a terminal stage, or stop() was called).
         """
         assert self._hf_job_id is not None
         try:
             while not self._stop_event.is_set():
                 clean_end = False
                 try:
-                    seen = 0
-                    for raw in self._api.fetch_job_logs(job_id=self._hf_job_id, follow=True):
+                    for raw in self._iter_job_logs():
                         if self._stop_event.is_set():
                             return
-                        seen += 1
-                        # Skip the replayed prefix from a reconnect.
-                        if seen <= self._lines_processed:
-                            continue
-                        self._lines_processed = seen
                         stripped = raw.rstrip()
                         if not stripped:
                             continue
+                        # Drop the prefix a reconnect replayed. Deliberately
+                        # AFTER the blank-line skip and on the stripped text, so
+                        # the window holds exactly what was emitted.
+                        if self._is_replayed(stripped):
+                            continue
                         parse_metrics_into(stripped, self._metrics)
                         if self._wandb_run_url is None:
                             url = extract_wandb_run_url(stripped)

tests/test_jobs.py

@@ -2868,3 +2868,81 @@ def test_start_allows_matching_feature_space(tmp_path) -> None:
     ):
         record = reg.start(cfg, JobTarget(runner="local"))
     assert record.state == "running"
+
+
+# ---------------------------------------------------------------------------
+# MT47 mitigation: a terminal record must not advertise progress it no longer
+# believes, nor a countdown for a run that has already stopped.
+# ---------------------------------------------------------------------------
+
+
+def _record_with_metrics(state, **metrics):
+    from makermodslab.jobs import JobRecord, TrainingMetrics
+    from makermodslab.train import TrainingRequest
+
+    return JobRecord(
+        id="J",
+        name="j",
+        state=state,
+        config=TrainingRequest(dataset_repo_id="d", steps=10000),
+        output_dir="/tmp/j",
+        started_at=0.0,
+        metrics=TrainingMetrics(**metrics),
+    )
+
+
+def test_settle_terminal_metrics_snaps_a_done_run_to_its_target() -> None:
+    """The live symptom: a `done` cloud run whose log stream died at step 3,650
+    kept rendering "3,650 / 10,000" beside a step-10,000 checkpoint. `done`
+    means the target was reached, so progress is the target."""
+    from makermodslab.jobs import _settle_terminal_metrics
+
+    record = _record_with_metrics("done", current_step=3650, total_steps=10000, eta_seconds=3185.0)
+
+    _settle_terminal_metrics(record)
+
+    assert record.metrics.current_step == 10000
+    assert record.metrics.eta_seconds is None
+
+
+def test_settle_terminal_metrics_never_invents_progress_for_a_failed_run() -> None:
+    """A failed or interrupted run genuinely stopped where the last frame said.
+    Rounding that up to the target would claim training that never happened —
+    and would poison the resume flow, which reads the step to decide what is
+    left to do."""
+    from makermodslab.jobs import _settle_terminal_metrics
+
+    for state in ("failed", "interrupted"):
+        record = _record_with_metrics(state, current_step=3650, total_steps=10000, eta_seconds=3185.0)
+
+        _settle_terminal_metrics(record)
+
+        assert record.metrics.current_step == 3650, state
+        # The ETA still goes: nothing terminal has time remaining.
+        assert record.metrics.eta_seconds is None, state
+
+
+def test_settle_terminal_metrics_leaves_an_unknown_target_alone() -> None:
+    """total_steps == 0 means tqdm never spoke, which the UI reads as
+    "Training starting…". Snapping to it would assert 0/0 as a finished run."""
+    from makermodslab.jobs import _settle_terminal_metrics
+
+    record = _record_with_metrics("done", current_step=0, total_steps=0)
+
+    _settle_terminal_metrics(record)
+
+    assert record.metrics.current_step == 0
+    assert record.metrics.total_steps == 0
+
+
+def test_settle_terminal_metrics_is_a_noop_on_a_healthy_finished_run() -> None:
+    """With the log tail repaired this is the normal case, and it must not
+    change anything — the mitigation exists for the broken-stream case only."""
+    from makermodslab.jobs import _settle_terminal_metrics
+
+    record = _record_with_metrics("done", current_step=10000, total_steps=10000, current_loss=0.04)
+
+    _settle_terminal_metrics(record)
+
+    assert record.metrics.current_step == 10000
+    assert record.metrics.current_loss == 0.04