diff --git a/nemo_automodel/components/loggers/metric_logger.py b/nemo_automodel/components/loggers/metric_logger.py
index 549587794a..0bc6ba273b 100644
--- a/nemo_automodel/components/loggers/metric_logger.py
+++ b/nemo_automodel/components/loggers/metric_logger.py
@@ -106,7 +106,9 @@ def __init__(
         if not isinstance(buffer_size, int) or buffer_size < 1:
             raise ValueError("buffer_size must be a positive integer")
         self.filepath = os.path.abspath(filepath)
-        self.flush = flush
+        # NOT `self.flush`: an instance attribute of that name shadows the flush() method
+        # below, so callers get a bool back and calling it raises TypeError.
+        self._fsync_on_write = flush
         self.buffer_size = buffer_size
         self.buffer: List[MetricsSample] = []
         self._lock = threading.Lock()
@@ -133,14 +135,37 @@ def _save(self, lines: List[str]) -> None:
         if len(lines) == 0:
             return
         self._fp.write("\n".join(lines) + "\n")
-        if self.flush:
+        if self._fsync_on_write:
+            self._fp.flush()
+            os.fsync(self._fp.fileno())
+
+    def _drain(self) -> None:
+        """Write buffered records out. Caller must hold ``self._lock``."""
+        self._save(self._move_to_cpu(self.buffer))
+        self.buffer = []
+
+    def flush(self) -> None:
+        """Write buffered records out to the file and fsync, without closing it.
+
+        Lets a caller align durability with an external event -- a checkpoint boundary, say --
+        instead of relying on the record count happening to reach ``buffer_size`` there.
+
+        The fsync is unconditional, NOT gated on the ``flush=`` constructor flag. That flag
+        controls whether every buffer-size-triggered write pays for an fsync; this method is
+        an explicit, caller-chosen durability point, so it has to be durable under the
+        default ``flush=False`` too. Draining into the file object alone would leave the
+        records in the process's stdio buffer -- still lost to a crash, and invisible to
+        anything reading the file -- which would make the guarantee empty exactly where it
+        is being relied on.
+        """
+        with self._lock:
+            self._drain()
             self._fp.flush()
             os.fsync(self._fp.fileno())
 
     def close(self) -> None:
         with self._lock:
-            self._save(self._move_to_cpu(self.buffer))
-            self.buffer = []
+            self._drain()
             try:
                 self._fp.flush()
             except Exception:
@@ -173,6 +198,11 @@ def log(self, record: MetricsSample) -> None:
             return
         super().log(record)
 
+    def flush(self) -> None:
+        if self.rank != 0:
+            return
+        super().flush()
+
     def close(self) -> None:
         if self.rank != 0:
             return
diff --git a/nemo_automodel/recipes/retrieval/train_bi_encoder.py b/nemo_automodel/recipes/retrieval/train_bi_encoder.py
index 5eb9f3f33c..f6656941a1 100644
--- a/nemo_automodel/recipes/retrieval/train_bi_encoder.py
+++ b/nemo_automodel/recipes/retrieval/train_bi_encoder.py
@@ -31,7 +31,11 @@
 from nemo_automodel.components.distributed.init_utils import initialize_distributed
 from nemo_automodel.components.distributed.utils import FirstRankPerNode, get_sync_ctx
 from nemo_automodel.components.loggers.log_utils import setup_logging
-from nemo_automodel.components.loggers.metric_logger import MetricsSample, build_metric_logger
+from nemo_automodel.components.loggers.metric_logger import (
+    DEFAULT_BUFFER_SIZE,
+    MetricsSample,
+    build_metric_logger,
+)
 from nemo_automodel.components.loggers.wandb_utils import suppress_wandb_log_messages
 from nemo_automodel.components.optim.precision_warnings import warn_if_torch_adam_with_bf16_params
 from nemo_automodel.components.training.rng import ScopedRNG, StatefulRNG
@@ -384,11 +388,24 @@ def materialize_loader(config):
         )
         self._log_model_and_optimizer_details(self.model_parts, self.optimizer, self.lr_scheduler)
 
+        # buffer_size bounds how many records can be pending, so a hard kill that runs no
+        # cleanup loses at most one checkpoint interval of train metrics; it is NOT what keeps
+        # metrics in step with checkpoints, since it counts records rather than watching
+        # checkpoint boundaries. The explicit flush after each successful checkpoint is what
+        # provides that. flush=True because a buffered write only reaches the file object,
+        # whose own buffer dies with the process. Validation is rare enough to write every
+        # point.
+        train_logger_kwargs = {"flush": True}
+        if self.step_scheduler.ckpt_every_steps > 0:
+            train_logger_kwargs["buffer_size"] = min(DEFAULT_BUFFER_SIZE, self.step_scheduler.ckpt_every_steps)
         self.metric_logger_train = build_metric_logger(
-            pathlib.Path(self.checkpointer.config.checkpoint_dir) / "training.jsonl"
+            pathlib.Path(self.checkpointer.config.checkpoint_dir) / "training.jsonl",
+            **train_logger_kwargs,
         )
         self.metric_logger_valid = build_metric_logger(
-            pathlib.Path(self.checkpointer.config.checkpoint_dir) / "validation.jsonl"
+            pathlib.Path(self.checkpointer.config.checkpoint_dir) / "validation.jsonl",
+            buffer_size=1,
+            flush=True,
         )
         self.loss_average_window = deque(maxlen=self.step_scheduler.loss_average_window_steps)
 
@@ -431,13 +448,37 @@ def run_train_validation_loop(self):
                             train_loss=train_log_data.metrics["loss"],
                             val_loss=val_loss,
                         )
+                        # Flush once the checkpoint save returns, so the metrics describing the
+                        # steps it covers are durable by the time the loop moves on rather than
+                        # waiting for the buffer to fill. This is a post-save boundary, not an
+                        # atomic one: a crash between the save completing and this flush leaves
+                        # the checkpoint on disk with those records still buffered. Buffer size alone
+                        # does not give this: it counts records, and the count only lines up
+                        # with checkpoint steps while training runs from step 0 with periodic
+                        # checkpoints. Resuming at a step that is not a multiple of
+                        # ckpt_every_steps, or a checkpoint taken at an epoch boundary or the
+                        # final step, leaves the two out of phase.
+                        if self.metric_logger_train is not None:
+                            self.metric_logger_train.flush()
+                        if self.metric_logger_valid is not None:
+                            self.metric_logger_valid.flush()
                     self._maybe_collect_garbage()
+
+                    # Poll per step; the StepScheduler iterators stop on the flag.
+                    if self.step_scheduler.sigterm_received:
+                        logger.info(
+                            "Preemption signal received at step %d; stopping cleanly.", self.step_scheduler.step
+                        )
         finally:
             if pbar is not None:
                 pbar.close()
+            # In the finally, not after it: an exception (OOM, NCCL timeout) would
+            # otherwise propagate past these and discard every buffered record.
+            if self.metric_logger_train is not None:
+                self.metric_logger_train.close()
+            if self.metric_logger_valid is not None:
+                self.metric_logger_valid.close()
 
-        self.metric_logger_train.close()
-        self.metric_logger_valid.close()
         self._finalize_and_close_checkpointer()
 
     def _forward_backward_step(self, idx, batch, *, loss_buffer, num_batches, is_train: bool = True):
diff --git a/tests/unit_tests/loggers/test_metric_logger.py b/tests/unit_tests/loggers/test_metric_logger.py
index 7ad88267f7..d641decc0b 100644
--- a/tests/unit_tests/loggers/test_metric_logger.py
+++ b/tests/unit_tests/loggers/test_metric_logger.py
@@ -166,3 +166,55 @@ def test_metric_logger_dist_nonzero_noop(tmp_path, monkeypatch):
     else:
         # If file was not created, that's also acceptable
         assert True
+
+
+def test_flush_method_is_callable_and_not_shadowed_by_the_flag(tmp_path):
+    """The ``flush=`` constructor flag and the ``flush()`` method must coexist.
+
+    Storing the flag as ``self.flush`` replaced the method with a bool on every instance,
+    so the first caller -- the recipe, at its first checkpoint -- got
+    ``TypeError: 'bool' object is not callable``.
+    """
+    logger = MetricLogger(str(tmp_path / "metrics.jsonl"), flush=True, append=False)
+    try:
+        assert callable(logger.flush)
+    finally:
+        logger.close()
+
+
+def test_flush_persists_buffered_records_without_closing(tmp_path):
+    """flush() is what lets a caller align durability with an external event.
+
+    buffer_size is far larger than the number of records, so nothing reaches the file on
+    record count alone; only the explicit flush can put them there, and the file must stay
+    open for further writes afterwards.
+    """
+    logfile = tmp_path / "metrics.jsonl"
+    logger = MetricLogger(str(logfile), append=False, buffer_size=1000)
+    try:
+        logger.log(metric_logger_mod.MetricsSample(step=1, epoch=0, metrics={"loss": 1.0}))
+        logger.log(metric_logger_mod.MetricsSample(step=2, epoch=0, metrics={"loss": 2.0}))
+        assert _read_jsonl(logfile) == [], "buffer_size not reached, so nothing is durable yet"
+
+        logger.flush()
+        rows = _read_jsonl(logfile)
+        assert [r["step"] for r in rows] == [1, 2]
+
+        # still usable: the flush drained the buffer, it did not close the file
+        logger.log(metric_logger_mod.MetricsSample(step=3, epoch=0, metrics={"loss": 3.0}))
+        logger.flush()
+        assert [r["step"] for r in _read_jsonl(logfile)] == [1, 2, 3]
+    finally:
+        logger.close()
+
+
+def test_flush_on_an_empty_buffer_is_a_no_op(tmp_path):
+    """The recipe flushes at every checkpoint, including ones with nothing buffered."""
+    logfile = tmp_path / "metrics.jsonl"
+    logger = MetricLogger(str(logfile), append=False, buffer_size=1000)
+    try:
+        logger.flush()
+        logger.flush()
+        assert logfile.read_text() == ""
+    finally:
+        logger.close()
diff --git a/tests/unit_tests/recipes/test_retrieval_bi_encoder_recipe.py b/tests/unit_tests/recipes/test_retrieval_bi_encoder_recipe.py
index f7a2f599e5..06063bd542 100644
--- a/tests/unit_tests/recipes/test_retrieval_bi_encoder_recipe.py
+++ b/tests/unit_tests/recipes/test_retrieval_bi_encoder_recipe.py
@@ -12,13 +12,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 from collections import deque
 from contextlib import nullcontext
 from types import SimpleNamespace
 
+import pytest
 import torch
 
 from nemo_automodel.components.distributed.config import DDPConfig, FSDP2Config
+from nemo_automodel.components.loggers.metric_logger import MetricLogger, MetricsSample
 from nemo_automodel.recipes.retrieval import train_bi_encoder
 from nemo_automodel.recipes.retrieval.train_bi_encoder import (
     TrainBiEncoderRecipe,
@@ -259,3 +262,180 @@ def _fake_scale_grads_and_clip_grad_norm(*args, **kwargs):
     recipe._run_train_optim_step([{}], max_grad_norm=1.0)
 
     assert captured["use_torch_clip_grad_norm"] is False
+
+
+class _RecordingMetricLogger:
+    """Buffers like MetricLogger but records what actually reached "disk" and when."""
+
+    def __init__(self):
+        self.buffer = []
+        self.persisted = []
+        self.closed = False
+
+    def log(self, record):
+        self.buffer.append(record)
+
+    def flush(self):
+        self.persisted.extend(self.buffer)
+        self.buffer = []
+
+    def close(self):
+        self.flush()
+        self.closed = True
+
+
+class _LoopStepScheduler:
+    """Minimal StepScheduler stand-in driving a fixed number of steps.
+
+    ``sigterm_at`` makes the signal appear at that step, mirroring the real scheduler's
+    sticky flag: once raised it stays raised, and the epoch/step iterators stop on it.
+    """
+
+    def __init__(self, n_steps, ckpt_steps=(), sigterm_at=None, val_steps=()):
+        self.step = 0
+        self.epoch = 0
+        self._n = n_steps
+        self._ckpt = set(ckpt_steps)
+        self._val = set(val_steps)
+        self._sigterm_at = sigterm_at
+        self.sigterm_flag = False
+        self.steps_run = []
+
+    @property
+    def epochs(self):
+        for e in range(1):
+            if self.sigterm_received:
+                return
+            yield e
+
+    def set_epoch(self, epoch):
+        self.epoch = epoch
+
+    def __iter__(self):
+        while self.step < self._n:
+            self.step += 1
+            if self.sigterm_flag:
+                return
+            self.steps_run.append(self.step)
+            yield [{}]
+
+    @property
+    def sigterm_received(self):
+        if self.sigterm_flag:
+            return True
+        if self._sigterm_at is not None and self.step >= self._sigterm_at:
+            self.sigterm_flag = True
+        return self.sigterm_flag
+
+    @property
+    def is_ckpt_step(self):
+        return self.step in self._ckpt or self.sigterm_received
+
+    @property
+    def is_val_step(self):
+        return self.step in self._val
+
+
+def _make_loop_recipe(step_scheduler, *, raise_at=None):
+    recipe = TrainBiEncoderRecipe.__new__(TrainBiEncoderRecipe)
+    recipe.model_parts = [torch.nn.Linear(1, 1)]
+    recipe.step_scheduler = step_scheduler
+    recipe.dataloader = SimpleNamespace(dataset=SimpleNamespace())
+    recipe.val_dataloader = None
+    recipe.max_grad_norm = 1.0
+    recipe.timestamp = 0.0
+    recipe.metric_logger_train = _RecordingMetricLogger()
+    recipe.metric_logger_valid = _RecordingMetricLogger()
+    recipe.saved_at = []
+
+    def _optim_step(batches, max_grad_norm):
+        if raise_at is not None and step_scheduler.step == raise_at:
+            raise RuntimeError("simulated CUDA OOM")
+        return SimpleNamespace(metrics={"loss": 1.0})
+
+    recipe._run_train_optim_step = _optim_step
+    recipe.log_train_metrics = lambda d: recipe.metric_logger_train.log(d)
+    recipe._make_progress_bar = lambda: None
+    recipe._update_progress_bar = lambda pbar, metrics: None
+    recipe._maybe_collect_garbage = lambda: None
+    recipe._finalize_and_close_checkpointer = lambda: None
+    recipe.save_checkpoint = lambda *a, **k: recipe.saved_at.append(step_scheduler.step)
+    return recipe
+
+
+def test_sigterm_on_checkpoint_step_saves_then_stops_the_loop():
+    """A signal on a scheduled checkpoint step: the checkpoint is written, then the
+    post-checkpoint poll stops the loop instead of running further steps."""
+    sched = _LoopStepScheduler(n_steps=10, ckpt_steps={4}, sigterm_at=4)
+    recipe = _make_loop_recipe(sched)
+
+    recipe.run_train_validation_loop()
+
+    assert sched.steps_run == [1, 2, 3, 4], "loop must stop after the signalled step"
+    assert recipe.saved_at == [4], "checkpoint must still be taken"
+    assert recipe.metric_logger_train.closed
+    assert len(recipe.metric_logger_train.persisted) == 4, "every step's metrics must survive"
+
+
+def test_metrics_persisted_when_the_loop_raises():
+    """An exception mid-loop must still persist buffered metrics, because the loggers are
+    closed in the finally rather than after it."""
+    sched = _LoopStepScheduler(n_steps=10)
+    recipe = _make_loop_recipe(sched, raise_at=3)
+
+    with pytest.raises(RuntimeError, match="simulated CUDA OOM"):
+        recipe.run_train_validation_loop()
+
+    assert recipe.metric_logger_train.closed, "logger must be closed from the finally"
+    assert recipe.metric_logger_valid.closed
+    # steps 1 and 2 logged before the raise; both must have reached disk
+    assert len(recipe.metric_logger_train.persisted) == 2
+
+
+def _durable_steps(path):
+    """Steps whose records are readable from the file right now, by anything else."""
+    if not path.exists():
+        return []
+    return [json.loads(line)["step"] for line in path.read_text().splitlines() if line.strip()]
+
+
+def test_checkpoint_flushes_metrics_to_disk_mid_run(tmp_path):
+    """The durability guarantee comes from the explicit flush, not from record counting.
+
+    Uses a REAL MetricLogger and reads the real file, because the property under test is
+    whether records are on disk at a point in time -- which a stand-in that records calls
+    cannot show. It is also the only way this test can fail if the recipe's flush calls are
+    deleted: the earlier version sampled the count before the flush and then checked a total
+    that ``close()`` in the finally satisfies on its own, so it passed either way.
+
+    The checkpoint lands at step 3, which no record-count boundary coincides with -- the
+    situation after resuming at a step that is not a multiple of ckpt_every_steps, or at an
+    epoch-boundary checkpoint. buffer_size is far above the length of the run, so nothing
+    reaches the file on count alone.
+    """
+    logfile = tmp_path / "train_metrics.jsonl"
+    sched = _LoopStepScheduler(n_steps=6, ckpt_steps={3})
+    recipe = _make_loop_recipe(sched)
+    recipe.metric_logger_train = MetricLogger(str(logfile), append=False, buffer_size=1000)
+    recipe.log_train_metrics = lambda d: recipe.metric_logger_train.log(
+        MetricsSample(step=sched.step, epoch=0, metrics={"loss": 1.0})
+    )
+
+    # Observe the file at the TOP of each step, so step N sees the state left by step N-1.
+    seen = {}
+    inner_step = recipe._run_train_optim_step
+
+    def _observing_step(batches, max_grad_norm):
+        seen[sched.step] = _durable_steps(logfile)
+        return inner_step(batches, max_grad_norm)
+
+    recipe._run_train_optim_step = _observing_step
+    recipe.run_train_validation_loop()
+
+    assert recipe.saved_at == [3]
+    assert seen[3] == [], "nothing durable before the checkpoint: the buffer is nowhere near full"
+    # THE ASSERTION THAT BITES: only the post-checkpoint flush can have put these on disk,
+    # and it is checked mid-run, before close() drains anything.
+    assert seen[4] == [1, 2, 3], "the checkpoint flush must have made the covered steps durable"
+    assert seen[6] == [1, 2, 3], "and no further flush happens until the run ends"
+    assert _durable_steps(logfile) == [1, 2, 3, 4, 5, 6], "close() drains the remainder"