diff --git a/python/changelog.d/810.feat.md b/python/changelog.d/810.feat.md
new file mode 100644
index 0000000000..186ec3813c
--- /dev/null
+++ b/python/changelog.d/810.feat.md
@@ -0,0 +1 @@
+Test-results log uploads are faster: consecutive measurements are now sent in one `CreateTestMeasurements` call, a retry after an unknown outcome checks the server first and reports it with the new `SiftReplayReconciledWarning`, and pytest now waits up to 120s (was 30s) at session end for the background upload, asking it to stop cleanly if it runs over.
diff --git a/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py b/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py
index aae8ca2592..90986b5424 100644
--- a/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py
+++ b/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py
@@ -54,6 +54,7 @@
 import logging
 import os
 import re
+import threading
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Generator
@@ -92,13 +93,13 @@ class LogTracking:
     file itself is append-only and stores only API-call data lines.
 
     * ``last_uploaded_line`` is the count of data lines that have been
-      successfully replayed against the server, in log order. Each data line
-      corresponds to a single API call, so line granularity matches the atomic
-      unit of work: a line is either fully replayed or must be retried in its
-      entirety. Data lines are strictly append-only, so this counter is stable
-      across runs. A batch upload creates in collapsed order rather than log
-      order and so leaves it at zero; ``complete`` is what marks that upload
-      finished.
+      successfully replayed against the server, in log order. A line is either
+      fully replayed or must be retried in its entirety. Most lines are one API
+      call, and a run of consecutive measurement lines is one batch call, so the
+      counter advances by the whole run once that call returns. Data lines are
+      strictly append-only, so this counter is stable across runs. A batch upload
+      creates in collapsed order rather than log order and so leaves it at zero;
+      ``complete`` is what marks that upload finished.
     * ``id_map`` maps simulated response IDs (created during the original test
       run) to the real IDs assigned by the server during replay. Subsequent
       ``Update*`` entries consult this map to translate IDs, and a resumed
@@ -126,6 +127,15 @@ def backup_path(log_path: str | Path) -> Path:
         sidecar = LogTracking.sidecar_path(log_path)
         return sidecar.with_name(sidecar.name + ".bak")
 
+    @staticmethod
+    def stop_path(log_path: str | Path) -> Path:
+        """Return the file that asks the replay worker to stop (``<log>.stop``).
+
+        Used where the worker cannot be signalled gracefully, which is Windows.
+        """
+        p = Path(log_path)
+        return p.with_name(p.name + ".stop")
+
     @staticmethod
     def archive(log_path: str | Path) -> Path | None:
         """Move an existing sidecar aside to ``<log>.tracking.bak``; return its new path.
@@ -207,6 +217,28 @@ class ReplayResult:
     report: TestReport | None = None
     steps: list[TestStep] = field(default_factory=list)
     measurements: list[TestMeasurement] = field(default_factory=list)
+    reconciled_count: int = 0
+    """How many measurements were matched on the server instead of recreated.
+
+    Nonzero only after a batch create failed with an unknown outcome (see
+    ``_create_measurement_batch_reconciling``) and the server already had some
+    or all of the batch.
+    """
+
+
+# Set by the replay worker's SIGTERM handler; the incremental importer checks it
+# between requests. Only the dedicated worker process ever sets it.
+_replay_stop = threading.Event()
+
+
+def request_replay_stop() -> None:
+    """Ask the incremental importer to stop after the request in flight."""
+    _replay_stop.set()
+
+
+def replay_stop_requested(log_path: str | Path) -> bool:
+    """True once a stop was requested by signal or by the log's stop file."""
+    return _replay_stop.is_set() or LogTracking.stop_path(log_path).exists()
 
 
 async def log_request_to_file(
diff --git a/python/lib/sift_client/_internal/low_level_wrappers/test_results.py b/python/lib/sift_client/_internal/low_level_wrappers/test_results.py
index da4c4bfb8f..4033e2001b 100644
--- a/python/lib/sift_client/_internal/low_level_wrappers/test_results.py
+++ b/python/lib/sift_client/_internal/low_level_wrappers/test_results.py
@@ -3,6 +3,8 @@
 import logging
 import uuid
 import warnings
+from collections import defaultdict
+from datetime import timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar, cast
 
@@ -49,10 +51,11 @@
     _ReplayState,
     log_request_to_file,
     parse_log_data_lines,
+    replay_stop_requested,
 )
 from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
 from sift_client._internal.pytest_plugin.audit_log import log_event
-from sift_client._internal.util.util import caller_stacklevel
+from sift_client._internal.util.util import caller_stacklevel, chunked
 from sift_client.errors import SiftIgnoredInputWarning
 from sift_client.sift_types.test_report import (
     TestMeasurement,
@@ -66,6 +69,7 @@
     TestStepUpdate,
 )
 from sift_client.transport import WithGrpcClient
+from sift_client.util.cel_utils import and_, in_
 
 if TYPE_CHECKING:
     from sift_client.transport.grpc_transport import GrpcClient
@@ -82,6 +86,36 @@
 # handler filters per measurement instead.
 _WHOLE_ENTRY_CREATES = frozenset({"CreateTestReport", "CreateTestStep", "CreateTestMeasurement"})
 
+# Measurements per CreateTestMeasurements request. One measurement with a full
+# description and metadata is about 2.5KB, so 100 stays well under the 4MB gRPC
+# message limit. It is also the flush threshold for a run of measurement log
+# lines and the most progress a crash can cost, since the sidecar is written
+# once per request.
+_MEASUREMENT_CREATE_BATCH_SIZE = 100
+
+# Status codes where the request may never have arrived or may have been
+# committed with only the response lost. Any other code is a definite rejection.
+_AMBIGUOUS_STATUS_CODES = frozenset(
+    {StatusCode.UNAVAILABLE, StatusCode.DEADLINE_EXCEEDED, StatusCode.CANCELLED, StatusCode.UNKNOWN}
+)
+
+
+def _rpc_code(exc: RpcError) -> StatusCode | None:
+    """Status code of *exc*, or None when it has none.
+
+    RpcError itself carries no status; only the Call subclasses the channel
+    raises do, and the async ones are not ``grpc.Call``.
+    """
+    return getattr(exc, "code", lambda: None)()
+
+
+def _is_ambiguous_grpc_failure(exc: Exception) -> bool:
+    """True when *exc* is an RpcError whose status is in ``_AMBIGUOUS_STATUS_CODES``.
+
+    Any other exception, such as a local ``TypeError``, is not ambiguous.
+    """
+    return isinstance(exc, RpcError) and _rpc_code(exc) in _AMBIGUOUS_STATUS_CODES
+
 
 class _EntryIds(NamedTuple):
     """The entity a replayed log entry acted on, for the audit trace.
@@ -94,11 +128,15 @@ class _EntryIds(NamedTuple):
 
     ``skipped`` marks an entry a resume found already on the server, so the
     audit trail distinguishes what this run sent from what it inherited.
+
+    ``reconciled`` counts measurements this entry matched on the server instead
+    of recreating; see ``_create_measurement_batch_reconciling``.
     """
 
     sim_id: str | None = None
     real_id: str | None = None
     skipped: bool = False
+    reconciled: int = 0
 
 
 class TestResultsLowLevelClient(LowLevelClientBase, WithGrpcClient):
@@ -860,6 +898,133 @@ async def create_test_measurements(
         response = cast("CreateTestMeasurementsResponse", response)
         return response.measurements_created_count, list(response.measurement_ids)
 
+    async def _create_measurement_batch(
+        self, protos: list[TestMeasurementProto]
+    ) -> list[TestMeasurement]:
+        """Create measurements in one batch call, returned in request order.
+
+        The response carries only IDs, so each entity is rebuilt from the proto
+        that was sent. Fields the server normalizes hold the values as sent.
+
+        Raises:
+            RuntimeError: If the response has a different number of IDs than
+                measurements sent.
+        """
+        _, real_ids = await self.create_test_measurements(
+            request=CreateTestMeasurementsRequest(test_measurements=protos)
+        )
+        if len(real_ids) != len(protos):
+            raise RuntimeError(
+                f"CreateTestMeasurements returned {len(real_ids)} IDs for "
+                f"{len(protos)} measurements; cannot match IDs to measurements."
+            )
+        created: list[TestMeasurement] = []
+        for proto, real_id in zip(protos, real_ids):
+            proto.measurement_id = real_id
+            created.append(TestMeasurement._from_proto(proto))
+        return created
+
+    async def _reconcile_measurement_batch(
+        self, protos: list[TestMeasurementProto]
+    ) -> list[TestMeasurement | None]:
+        """Match a batch's measurements against what the server already has.
+
+        The key is ``test_step_id``, ``name``, ``timestamp``, and whichever
+        value field is set. Each server row matches at most one proto, so
+        identical measurements in one batch need identical rows on the server.
+        Best effort: a value the server stores differently from what was sent
+        does not match.
+
+        Returns one slot per input proto, in order; ``None`` where nothing on
+        the server matched.
+        """
+        step_ids = sorted({proto.test_step_id for proto in protos})
+        names = sorted({proto.name for proto in protos})
+        existing = await self.list_all_test_measurements(
+            query_filter=and_(in_("test_step_id", step_ids), in_("name", names))
+        )
+
+        candidates: defaultdict[tuple[str, str, Any, Any], list[TestMeasurement]] = defaultdict(
+            list
+        )
+        for measurement in existing:
+            value: float | str | bool | None = measurement.numeric_value
+            if value is None:
+                value = measurement.string_value
+            if value is None:
+                value = measurement.boolean_value
+            key = (measurement.test_step_id, measurement.name, measurement.timestamp, value)
+            candidates[key].append(measurement)
+
+        matches: list[TestMeasurement | None] = []
+        for proto in protos:
+            which = proto.WhichOneof("value")
+            value = getattr(proto, which) if which else None
+            key = (
+                proto.test_step_id,
+                proto.name,
+                proto.timestamp.ToDatetime(tzinfo=timezone.utc),
+                value,
+            )
+            bucket = candidates.get(key)  # get, so a miss does not insert an empty list
+            matches.append(bucket.pop(0) if bucket else None)
+        return matches
+
+    async def _create_measurement_batch_reconciling(
+        self, protos: list[TestMeasurementProto]
+    ) -> tuple[list[TestMeasurement], int]:
+        """Create a batch, reconciling against the server if the outcome is unknown.
+
+        ``CreateTestMeasurements`` is not idempotent. After an ambiguous failure
+        (see ``_is_ambiguous_grpc_failure``), this matches the batch against the
+        server (see ``_reconcile_measurement_batch``) and recreates only what is
+        missing. Any other exception, and any failure of the reconciliation
+        query or the recreate, propagates.
+
+        Returns the measurements in input order, plus how many were matched on
+        the server instead of recreated.
+        """
+        try:
+            return await self._create_measurement_batch(protos), 0
+        except Exception as exc:
+            if not _is_ambiguous_grpc_failure(exc):
+                raise
+            matches = await self._reconcile_measurement_batch(protos)
+            missing = [proto for proto, match in zip(protos, matches) if match is None]
+            reconciled = len(protos) - len(missing)
+            log_event(
+                logger,
+                logging.WARNING,
+                "replay.reconciled",
+                total=len(protos),
+                reconciled=reconciled,
+                missing=len(missing),
+                error=repr(exc),
+            )
+            if not missing:
+                return cast("list[TestMeasurement]", matches), reconciled
+            recreated = iter(await self._create_measurement_batch(missing))
+            created = [match if match is not None else next(recreated) for match in matches]
+            return created, reconciled
+
+    async def _create_measurements_in_batches(
+        self, protos: list[TestMeasurementProto]
+    ) -> tuple[list[TestMeasurement], int]:
+        """Create measurements in chunks of ``_MEASUREMENT_CREATE_BATCH_SIZE``.
+
+        Returns the created measurements plus how many were reconciled rather
+        than recreated; see ``_create_measurement_batch_reconciling``.
+        """
+        created: list[TestMeasurement] = []
+        reconciled_count = 0
+        for chunk in chunked(protos, _MEASUREMENT_CREATE_BATCH_SIZE):
+            chunk_created, chunk_reconciled = await self._create_measurement_batch_reconciling(
+                chunk
+            )
+            created.extend(chunk_created)
+            reconciled_count += chunk_reconciled
+        return created, reconciled_count
+
     async def list_test_measurements(
         self,
         *,
@@ -1098,10 +1263,7 @@ async def _resume_report(self, log_path: Path, tracking: LogTracking) -> TestRep
         try:
             return await self.get_test_report(real_report_id)
         except RpcError as exc:
-            # RpcError itself carries no status; only the Call subclasses the
-            # channel actually raises do, and the async ones are not grpc.Call.
-            code = getattr(exc, "code", lambda: None)()
-            if code != StatusCode.NOT_FOUND:
+            if _rpc_code(exc) != StatusCode.NOT_FOUND:
                 raise
             raise ValueError(
                 f"Test report {real_report_id}, from the interrupted upload recorded in "
@@ -1220,6 +1382,43 @@ async def _replay_create_measurement(
         state.measurements_order.append(measurement._id_or_error)
         return _EntryIds(response_id, measurement._id_or_error)
 
+    async def _replay_create_measurement_group(
+        self,
+        pairs: list[tuple[str | None, TestMeasurementProto]],
+        *,
+        id_map: dict[str, str],
+        state: _ReplayState,
+    ) -> tuple[list[str], int]:
+        """Create a group of logged measurements, recording what each one became.
+
+        ``pairs`` holds the logged response ID, where the line carried one,
+        against the measurement proto to send. An entry whose logged ID is already
+        in the sidecar map reached the server on an earlier attempt and is not
+        created again.
+
+        Returns the real IDs created, in the order they were sent, plus how many
+        were reconciled rather than recreated (see
+        ``_create_measurement_batch_reconciling``).
+        """
+        pending = [
+            (logged_id, tm) for logged_id, tm in pairs if not (logged_id and logged_id in id_map)
+        ]
+        if not pending:
+            return [], 0
+        created, reconciled_count = await self._create_measurements_in_batches(
+            [tm for _, tm in pending]
+        )
+
+        real_ids: list[str] = []
+        for (logged_id, _), measurement in zip(pending, created):
+            real_id = measurement._id_or_error
+            if logged_id:
+                id_map[logged_id] = real_id
+            state.measurements_by_id[real_id] = measurement
+            state.measurements_order.append(real_id)
+            real_ids.append(real_id)
+        return real_ids, reconciled_count
+
     async def _replay_create_measurements(
         self,
         json_str: str,
@@ -1234,8 +1433,8 @@ async def _replay_create_measurements(
         for tm in request.test_measurements:
             tm.test_step_id = self._map_id(id_map, tm.test_step_id)
         original_ids = response_id.split(",") if response_id else []
-        created_ids: list[str] = []
         if simulate:
+            created_ids: list[str] = []
             # Batch endpoint has no simulate path; fan out to per-measurement simulate calls.
             for i, tm_proto in enumerate(request.test_measurements):
                 single_req = CreateTestMeasurementRequest(test_measurement=tm_proto)
@@ -1245,32 +1444,25 @@ async def _replay_create_measurements(
                 state.measurements_by_id[meas._id_or_error] = meas
                 state.measurements_order.append(meas._id_or_error)
                 created_ids.append(meas._id_or_error)
+            reconciled_count = 0
         else:
-            # Batch replay creates measurements one at a time, so an interrupted
-            # run can leave part of a batch line already on the server. Re-sending
-            # the whole line would duplicate those, so send only what is missing.
-            pending: list[tuple[str | None, TestMeasurementProto]] = []
-            for i, tm in enumerate(request.test_measurements):
-                logged_id = original_ids[i] if i < len(original_ids) else None
-                if logged_id and logged_id in id_map:
-                    continue
-                pending.append((logged_id, tm))
-            real_ids: list[str] = []
-            if pending:
-                _, real_ids = await self.create_test_measurements(
-                    request=CreateTestMeasurementsRequest(
-                        test_measurements=[tm for _, tm in pending]
-                    )
-                )
-            for (logged_id, _), real_id in zip(pending, real_ids):
-                if logged_id:
-                    id_map[logged_id] = real_id
-                created_ids.append(real_id)
+            pairs = [
+                (original_ids[i] if i < len(original_ids) else None, tm)
+                for i, tm in enumerate(request.test_measurements)
+            ]
+            created_ids, reconciled_count = await self._replay_create_measurement_group(
+                pairs, id_map=id_map, state=state
+            )
         # Batch line covers many measurements; comma-join both sides so the
         # audit row still names every entity (fields are space-free, so commas
         # keep it one token). A resume that found the whole line already on the
         # server creates nothing, which is what marks the entry skipped.
-        return _EntryIds(response_id, ",".join(created_ids) or None, skipped=not created_ids)
+        return _EntryIds(
+            response_id,
+            ",".join(created_ids) or None,
+            skipped=not created_ids,
+            reconciled=reconciled_count,
+        )
 
     async def _replay_update_report(
         self,
@@ -1393,19 +1585,23 @@ async def _batch_import_log_file(self, log_path: Path) -> ReplayResult:
         logged_by_simulated = {simulated: logged for logged, simulated in id_map.items()}
         real_id_map: dict[str, str] = {}
 
-        def record_created(simulated_id: str, real_id: str) -> None:
+        def record_created(simulated_id: str, real_id: str, *, save: bool = True) -> None:
             """Note a real entity against both the in-run map and the sidecar.
 
             The sidecar is saved per entity so an upload interrupted at any point
             is resumable. That is one small atomic rewrite per created entity;
             the incremental path already pays the same cost per log line.
+
+            ``save=False`` records the entity without writing, for a caller that
+            creates a batch in one call and writes once for the whole batch.
             """
             real_id_map[simulated_id] = real_id
             logged_id = logged_by_simulated.get(simulated_id)
             if not logged_id:
                 return
             tracking.id_map[logged_id] = real_id
-            tracking.save(log_path)
+            if save:
+                tracking.save(log_path)
 
         real_report = await self._create_report_from_simulated(state.report)
         real_report_id = real_report._id_or_error
@@ -1431,18 +1627,27 @@ def record_created(simulated_id: str, real_id: str) -> None:
             real_steps.append(real_step)
             record_created(sim_step_id, real_step._id_or_error)
 
+        # The sidecar is written once per chunk, so a crash loses at most one
+        # request's IDs and the resume recreates those measurements.
         real_measurements: list[TestMeasurement] = []
-        for sim_measurement_id in state.measurements_order:
-            sim_measurement = state.measurements_by_id[sim_measurement_id]
-            real_step_id = real_id_map.get(
-                sim_measurement.test_step_id, sim_measurement.test_step_id
-            )
-            measurement_create = self._measurement_create_from_simulated(
-                sim_measurement, real_step_id
-            )
-            real_measurement = await self.create_test_measurement(measurement_create)
-            real_measurements.append(real_measurement)
-            record_created(sim_measurement_id, real_measurement._id_or_error)
+        reconciled_count = 0
+        for chunk_ids in chunked(state.measurements_order, _MEASUREMENT_CREATE_BATCH_SIZE):
+            protos = []
+            for sim_measurement_id in chunk_ids:
+                sim_measurement = state.measurements_by_id[sim_measurement_id]
+                real_step_id = real_id_map.get(
+                    sim_measurement.test_step_id, sim_measurement.test_step_id
+                )
+                measurement_create = self._measurement_create_from_simulated(
+                    sim_measurement, real_step_id
+                )
+                protos.append(measurement_create.to_proto())
+            created, chunk_reconciled = await self._create_measurement_batch_reconciling(protos)
+            real_measurements.extend(created)
+            reconciled_count += chunk_reconciled
+            for sim_measurement_id, real_measurement in zip(chunk_ids, created):
+                record_created(sim_measurement_id, real_measurement._id_or_error, save=False)
+            tracking.save(log_path)
 
         # Everything in the log reached the server. The cursor stays at zero
         # because batch created in collapsed order, not log order; the flag is
@@ -1454,6 +1659,7 @@ def record_created(simulated_id: str, real_id: str) -> None:
             report=real_report,
             steps=real_steps,
             measurements=real_measurements,
+            reconciled_count=reconciled_count,
         )
 
     # ------------------------------------------------------------------
@@ -1470,10 +1676,10 @@ async def _incremental_import_log_file(
 
         Resumes from ``LogTracking.last_uploaded_line`` (loaded from the
         ``<log>.tracking`` sidecar) so already-uploaded entries are skipped on
-        subsequent ticks rather than re-sent to the server. Each data line is a
-        single atomic API call; if replay of a line fails,
-        ``last_uploaded_line`` is not advanced so the whole line is retried
-        next tick.
+        subsequent ticks rather than re-sent to the server. Most data lines are a
+        single atomic API call, and a run of consecutive measurement lines is one
+        batch call covering the run. If a call fails, ``last_uploaded_line`` is
+        not advanced, so every line it covered is retried next tick.
 
         A batch upload records what it created but keeps its cursor at zero,
         since it creates in collapsed order rather than log order. Finishing one
@@ -1497,9 +1703,86 @@ async def _incremental_import_log_file(
         state = _ReplayState()
 
         raw_lines = await _read_log_lines(log_path)
+
+        def stop_requested() -> bool:
+            # Only the live worker follows a stop request. A resume of a final
+            # log must walk to the end, or it would mark unsent data complete.
+            return not resuming and replay_stop_requested(log_path)
+
+        # A run of consecutive single-measurement lines collects here and is sent
+        # as one batch call.
+        buffered_measurements: list[tuple[str | None, TestMeasurementProto]] = []
+        reconciled_count = 0
+
+        async def flush_measurements() -> None:
+            """Send the buffered run as one group and move the cursor past it.
+
+            The cursor advances only after the call returns, so a failure leaves
+            every line of the run to be retried next tick.
+            """
+            nonlocal reconciled_count
+            if not buffered_measurements:
+                return
+            first_line = tracking.last_uploaded_line + 1
+            try:
+                real_ids, batch_reconciled = await self._replay_create_measurement_group(
+                    buffered_measurements, id_map=id_map, state=state
+                )
+            except Exception as exc:
+                log_event(
+                    logger,
+                    logging.WARNING,
+                    "replay.error",
+                    line=first_line,
+                    lines=len(buffered_measurements),
+                    type="CreateTestMeasurement",
+                    error=repr(exc),
+                )
+                raise
+            reconciled_count += batch_reconciled
+
+            tracking.last_uploaded_line += len(buffered_measurements)
+            tracking.save(log_path)
+            # One row per run rather than per line. ``line`` stays the cursor, as
+            # on every other row, and ``first_line`` names where the run started
+            # so a reader can attribute the whole span.
+            log_event(
+                logger,
+                logging.DEBUG,
+                "replay.upload",
+                type="CreateTestMeasurement",
+                line=tracking.last_uploaded_line,
+                first_line=first_line,
+                count=len(buffered_measurements),
+                sim_id=",".join(sid for sid, _ in buffered_measurements if sid) or "-",
+                real_id=",".join(real_ids) or "-",
+                skipped="yes" if not real_ids else "no",
+                idmap=len(id_map),
+            )
+            buffered_measurements.clear()
+
         for request_type, response_id, json_str in parse_log_data_lines(
             raw_lines, start_line=tracking.last_uploaded_line
         ):
+            if stop_requested():
+                break
+            if request_type == "CreateTestMeasurement":
+                # The step ID is resolvable now because a CreateTestStep line
+                # flushes the buffer before it is imported, so every step a
+                # buffered measurement can name is already in the ID map.
+                buffered_request = CreateTestMeasurementRequest()
+                json_format.Parse(json_str, buffered_request)
+                buffered_request.test_measurement.test_step_id = self._map_id(
+                    id_map, buffered_request.test_measurement.test_step_id
+                )
+                buffered_measurements.append((response_id, buffered_request.test_measurement))
+                if len(buffered_measurements) >= _MEASUREMENT_CREATE_BATCH_SIZE:
+                    await flush_measurements()
+                continue
+            # The cursor is a line count, so lines must reach the server in log
+            # order. Flush the run before the line that ended it.
+            await flush_measurements()
+
             line_number = tracking.last_uploaded_line + 1
             try:
                 entry_ids = await self._import_entry(
@@ -1525,6 +1808,7 @@ async def _incremental_import_log_file(
 
             tracking.last_uploaded_line += 1
             tracking.save(log_path)
+            reconciled_count += entry_ids.reconciled
             # One line per replayed entity: what was sent, the sim->real id of
             # the entity it acted on (the new entity for a create, the target
             # for an update), whether a resume found it already on the server,
@@ -1542,10 +1826,17 @@ async def _incremental_import_log_file(
                 idmap=len(id_map),
             )
 
+        # A stop leaves any buffered run unsent; the cursor has not moved past it,
+        # so the next run resends it. Otherwise a log ending in measurement lines
+        # leaves a run buffered.
+        stopped = stop_requested()
+        if not stopped:
+            await flush_measurements()
+
         # On a resume tick the CreateTestReport line was consumed on an earlier
         # tick, so state.report is expected to be None; the report already exists
         # on the server. Only a genuine first pass over an empty log is an error.
-        if state.report is None and not report_already_created:
+        if state.report is None and not report_already_created and not stopped:
             raise ValueError("No CreateTestReport found in log file")
 
         if resuming:
@@ -1558,6 +1849,7 @@ async def _incremental_import_log_file(
             report=state.report,
             steps=[state.steps_by_id[sid] for sid in state.steps_order],
             measurements=[state.measurements_by_id[mid] for mid in state.measurements_order],
+            reconciled_count=reconciled_count,
         )
 
     # ------------------------------------------------------------------
diff --git a/python/lib/sift_client/_internal/pytest_plugin/audit_log.py b/python/lib/sift_client/_internal/pytest_plugin/audit_log.py
index d79bb0d865..7192508144 100644
--- a/python/lib/sift_client/_internal/pytest_plugin/audit_log.py
+++ b/python/lib/sift_client/_internal/pytest_plugin/audit_log.py
@@ -40,6 +40,10 @@
 EVENT_WIDTH = 16
 # Tag so a re-entered pytest_configure (or both processes) doesn't double-attach.
 HANDLER_TAG = "sift_audit"
+# Prefix for the replay worker's stderr line reporting reconciled measurements
+# on an otherwise clean (exit 0) run. Lives here because ``replay_worker`` and
+# ``context_manager`` already import each other.
+RECONCILED_MARKER = "SIFT_REPLAY_RECONCILED"
 # Field names whose values should be redacted in audit logs to avoid logging
 # sensitive data. This is a best-effort blacklist; callers should avoid passing
 # sensitive values in the first place.
@@ -90,6 +94,19 @@ def log_event(logger: logging.Logger, level: int, event: str, **fields: object)
     logger.log(level, "%-*s %s", EVENT_WIDTH, event, body)
 
 
+def reconciled_warning_message(subject: str, count: int) -> str:
+    """Message for ``SiftReplayReconciledWarning``, shared by both warn sites.
+
+    ``subject`` names what did the reconciling, so the import script and the
+    pytest plugin's background worker can each say which one it was.
+    """
+    return (
+        f"{subject} reconciled {count} measurement(s) against the server rather than "
+        "recreating them, after a batch create's outcome came back unknown. No action "
+        "needed; this is a report that the recovery happened."
+    )
+
+
 class ColumnFormatter(logging.Formatter):
     """Formatter for the columnar file log.
 
diff --git a/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py b/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py
index 8dcaf19172..80892e20e7 100644
--- a/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py
+++ b/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py
@@ -18,14 +18,19 @@
 import os
 import select
 import shutil
+import signal
 import sys
 import tempfile
 from pathlib import Path
 from typing import TYPE_CHECKING
 
 from sift_client import SiftClient, SiftConnectionConfig
-from sift_client._internal.low_level_wrappers._test_results_log import LogTracking
-from sift_client._internal.pytest_plugin.audit_log import log_event
+from sift_client._internal.low_level_wrappers._test_results_log import (
+    LogTracking,
+    replay_stop_requested,
+    request_replay_stop,
+)
+from sift_client._internal.pytest_plugin.audit_log import RECONCILED_MARKER, log_event
 from sift_client.util.test_results.context_manager import log_replay_instructions
 
 if TYPE_CHECKING:
@@ -77,15 +82,33 @@ def _incremental_import_loop(
 
     When ``keep_log`` is False (audit logging off) the temp log is deleted on a
     clean finish; with audit logging on it's retained alongside the audit trail.
+
+    SIGTERM, or the log's stop file on Windows, stops the loop after the request
+    in flight, with the sidecar saved. The log is kept either way, since it is
+    not fully uploaded.
+
+    Measurements reconciled against the server (see
+    ``_create_measurement_batch_reconciling``) are summed across ticks and
+    reported once on stderr at session end, which is the only channel back to
+    the parent process.
     """
     result = None
+    total_reconciled = 0
     while True:
         received_signal, _, _ = select.select([sys.stdin], [], [], 1.0)
         result = client.test_results.import_log_file(log_file, incremental=True)
-        if received_signal:
+        if result is not None:
+            total_reconciled += result.reconciled_count
+        if received_signal or replay_stop_requested(log_file):
             break
-    log_event(logger, logging.INFO, "replay.complete", log=log_file)
-    if not keep_log:
+    stopped = replay_stop_requested(log_file)
+    log_event(
+        logger, logging.INFO, "replay.stopped" if stopped else "replay.complete", log=log_file
+    )
+    if total_reconciled:
+        log_event(logger, logging.WARNING, "replay.reconciled_total", count=total_reconciled)
+        print(f"{RECONCILED_MARKER}:{total_reconciled}", file=sys.stderr, flush=True)
+    if not keep_log and not stopped:
         cleanup_temp_log(log_file)
     return result
 
@@ -128,6 +151,9 @@ def main() -> None:
         )
     )
 
+    # Sent by the parent when the session-end wait times out.
+    signal.signal(signal.SIGTERM, lambda *_: request_replay_stop())
+
     # The worker is spawned with --audit-log only when audit logging is on, so
     # its presence is the signal to retain the buffer after a clean upload.
     try:
diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py
index 847f5d474a..ccc67781c3 100644
--- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py
+++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py
@@ -14,12 +14,17 @@
 import logging
 from contextlib import contextmanager
 from datetime import datetime, timezone
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from grpc import RpcError, StatusCode
 
-from sift_client._internal.low_level_wrappers._test_results_log import LogTracking
+from sift_client._internal.low_level_wrappers import test_results as test_results_module
+from sift_client._internal.low_level_wrappers._test_results_log import (
+    LogTracking,
+    _replay_stop,
+    request_replay_stop,
+)
 from sift_client._internal.low_level_wrappers.test_results import (
     # Aliased so pytest doesn't try to collect the `Test`-prefixed client as a suite.
     TestResultsLowLevelClient as ResultsLowLevelClient,
@@ -287,16 +292,29 @@ def _answer_real_creates(client, *, report_id, step_ids=(), measurement_ids=()):
     with ``simulate=True``, so a blanket mock would swallow those too. Returns
     the list that records the name of each real create, in order; the report is
     recorded as ``"report"`` since a report create carries no step name.
+
+    Replay creates measurements through the batch endpoint only, so
+    ``measurement_ids`` is drained by ``create_test_measurements``, one ID per
+    measurement in the request, and one name per measurement is recorded. Supply
+    as many IDs as the log holds measurements; running short raises
+    ``StopIteration``.
     """
     created: list[str] = []
 
-    def answer(name, canned_ids, describe):
+    def answer(name, canned, describe, *, many=False):
         real = getattr(client, name)
-        remaining = iter(canned_ids)
+        remaining = iter(canned)
 
         async def call(*args, **kwargs):
             if kwargs.get("simulate") or kwargs.get("log_file"):
                 return await real(*args, **kwargs)
+            if many:
+                # The batch endpoint answers one call with a count and one ID per
+                # measurement sent, rather than a single entity.
+                protos = kwargs["request"].test_measurements
+                created.extend(describe(proto) for proto in protos)
+                real_ids = [next(remaining) for _ in protos]
+                return len(real_ids), real_ids
             created.append(describe(*args))
             return next(remaining)
 
@@ -305,9 +323,10 @@ async def call(*args, **kwargs):
     answer("create_test_report", [_make_report(report_id)], lambda *_: "report")
     answer("create_test_step", [_make_step(sid) for sid in step_ids], lambda create: create.name)
     answer(
-        "create_test_measurement",
-        [_make_measurement(mid) for mid in measurement_ids],
-        lambda create: create.name,
+        "create_test_measurements",
+        list(measurement_ids),
+        lambda proto: proto.name,
+        many=True,
     )
     return created
 
@@ -516,11 +535,11 @@ async def test_resume_into_deleted_report_explains_the_override(tmp_path):
     client.create_test_step.assert_not_awaited()
 
 
-async def _build_measurement_log(client, log_file, *, batched):
-    """Write a log holding a report, one step, and three measurements.
+async def _build_measurement_log(client, log_file, *, batched, count=3):
+    """Write a log holding a report, one step, and ``count`` measurements.
 
     ``batched`` picks how the measurements are logged: one ``CreateTestMeasurements``
-    line covering all three, or a separate ``CreateTestMeasurement`` line each.
+    line covering all of them, or a separate ``CreateTestMeasurement`` line each.
     """
     report = await client.create_test_report(test_report=_report_create(), log_file=log_file)
     step = await client.create_test_step(
@@ -543,7 +562,7 @@ async def _build_measurement_log(client, log_file, *, batched):
             timestamp=T0,
             numeric_value=float(index),
         )
-        for index in (1, 2, 3)
+        for index in range(1, count + 1)
     ]
     if batched:
         _, measurement_ids = await client.create_test_measurements(
@@ -596,7 +615,11 @@ async def test_resume_sends_only_the_missing_part_of_a_batched_line(tmp_path):
 
 @pytest.mark.asyncio
 async def test_resume_skips_a_measurement_already_created(tmp_path):
-    """A measurement logged on its own line is skipped once it is in the id map."""
+    """A measurement logged on its own line is skipped once it is in the id map.
+
+    The one measurement left over goes through the batch endpoint like any other
+    group, so a resume does not send it somewhere a fuller run would not.
+    """
     log_file = tmp_path / "single_measurements.jsonl"
     client = ResultsLowLevelClient(grpc_client=MagicMock())
     report, step, measurement_ids = await _build_measurement_log(client, log_file, batched=False)
@@ -611,14 +634,418 @@ async def test_resume_skips_a_measurement_already_created(tmp_path):
     ).save(log_file)
 
     client.get_test_report = AsyncMock(return_value=_make_report("real-report"))
-    client.create_test_measurement = AsyncMock(return_value=_make_measurement("real-meas-3"))
+    client.create_test_measurement = AsyncMock()
+    client.create_test_measurements = AsyncMock(return_value=(1, ["real-meas-3"]))
 
     await client.import_log_file(log_file)
 
-    client.create_test_measurement.assert_awaited_once()
-    sent = client.create_test_measurement.await_args.kwargs["request"]
-    assert sent.test_measurement.name == "m3"
-    assert sent.test_measurement.test_step_id == "real-step"
+    client.create_test_measurement.assert_not_awaited()
+    client.create_test_measurements.assert_awaited_once()
+    sent = client.create_test_measurements.await_args.kwargs["request"]
+    assert [m.name for m in sent.test_measurements] == ["m3"]
+    assert sent.test_measurements[0].test_step_id == "real-step"
+
+
+@pytest.mark.asyncio
+async def test_measurement_run_goes_out_as_one_call(tmp_path):
+    """Consecutive single-measurement lines reach the server as one batch call.
+
+    One round-trip per measurement is what made a large log slow to upload: the
+    field report that prompted this had 10,412 of them.
+    """
+    log_file = tmp_path / "run_of_measurements.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, measurement_ids = await _build_measurement_log(client, log_file, batched=False)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    client.create_test_measurement = AsyncMock()
+    client.create_test_measurements = AsyncMock(
+        return_value=(3, ["real-meas-1", "real-meas-2", "real-meas-3"])
+    )
+
+    result = await client.import_log_file(log_file, incremental=True)
+
+    client.create_test_measurement.assert_not_awaited()
+    client.create_test_measurements.assert_awaited_once()
+    sent = client.create_test_measurements.await_args.kwargs["request"]
+    assert [m.name for m in sent.test_measurements] == ["m1", "m2", "m3"]
+    tracking = LogTracking.load(log_file)
+    # The cursor covers all three lines, and each logged ID names what it became.
+    assert tracking.last_uploaded_line == 5
+    assert [tracking.id_map[mid] for mid in measurement_ids] == [
+        "real-meas-1",
+        "real-meas-2",
+        "real-meas-3",
+    ]
+    # The batch endpoint returns IDs only, so the entities are rebuilt from what
+    # was sent; the replay result still reports them.
+    assert [m.name for m in result.measurements] == ["m1", "m2", "m3"]
+    assert [m.id_ for m in result.measurements] == ["real-meas-1", "real-meas-2", "real-meas-3"]
+
+
+@pytest.mark.asyncio
+async def test_measurement_run_saves_the_sidecar_once(tmp_path):
+    """The run costs one sidecar write, not one per measurement.
+
+    The sidecar rewrites the whole ID map on every save, so a write per
+    measurement is the other half of what made a large log slow.
+    """
+    log_file = tmp_path / "one_save.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    client.create_test_measurements = AsyncMock(
+        return_value=(3, ["real-meas-1", "real-meas-2", "real-meas-3"])
+    )
+
+    saves = 0
+    real_save = LogTracking.save
+
+    def counting_save(self, path):
+        nonlocal saves
+        saves += 1
+        real_save(self, path)
+
+    with patch.object(LogTracking, "save", counting_save):
+        await client.import_log_file(log_file, incremental=True)
+
+    assert saves == 1
+
+
+@pytest.mark.asyncio
+async def test_measurement_run_logs_one_audit_row_for_the_span(tmp_path):
+    """One ``replay.upload`` row covers the run, naming the span it advanced.
+
+    Every other row reports a single line, so a reader needs the count and the
+    first line to attribute a run's measurements to the right part of the log.
+    """
+    log_file = tmp_path / "audit_run.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, measurement_ids = await _build_measurement_log(client, log_file, batched=False)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    client.create_test_measurements = AsyncMock(
+        return_value=(3, ["real-meas-1", "real-meas-2", "real-meas-3"])
+    )
+
+    with _captured_replay_logs() as messages:
+        await client.import_log_file(log_file, incremental=True)
+
+    rows = [m for m in messages if m.startswith("replay.upload") and "CreateTestMeasurement" in m]
+    assert len(rows) == 1
+    row = rows[0]
+    assert "count=3" in row
+    assert "first_line=3" in row
+    assert "line=5" in row
+    assert "skipped=no" in row
+    # Both sides comma-joined, so the row still names every measurement.
+    assert f"sim_id={','.join(measurement_ids)}" in row
+    assert "real_id=real-meas-1,real-meas-2,real-meas-3" in row
+
+
+@pytest.mark.asyncio
+async def test_run_of_one_measurement_still_uses_the_batch_endpoint(tmp_path):
+    """Run length must not pick the endpoint, even for a run of one.
+
+    The two endpoints build entities differently: the batch one returns IDs, so
+    the entity is rebuilt from what was sent, while the single one returns the
+    server's copy. Choosing by size would make that depend on how many
+    measurements a step happened to record.
+    """
+    log_file = tmp_path / "single_run.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False, count=1)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    client.create_test_measurement = AsyncMock()
+    client.create_test_measurements = AsyncMock(return_value=(1, ["real-meas-1"]))
+
+    await client.import_log_file(log_file, incremental=True)
+
+    client.create_test_measurement.assert_not_awaited()
+    client.create_test_measurements.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_measurement_run_is_flushed_at_the_step_boundary(tmp_path):
+    """The line that ends a run is sent after the run, not before it.
+
+    The cursor is a single line count, so a line applied out of order would be
+    marked done before the lines ahead of it.
+    """
+    log_file = tmp_path / "step_boundary.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, step, _ = await _build_measurement_log(client, log_file, batched=False, count=2)
+    step_update = StepUpdate(status=TestStatus.FAILED)
+    step_update.resource_id = step.id_
+    await client.update_test_step(update=step_update, log_file=log_file)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+
+    order: list[str] = []
+
+    async def create_measurements(*_, **__):
+        order.append("measurements")
+        return 2, ["real-meas-1", "real-meas-2"]
+
+    async def update_step(*_, **__):
+        order.append("update_step")
+        return _make_step("real-step")
+
+    client.create_test_measurements = create_measurements
+    client.update_test_step = update_step
+
+    await client.import_log_file(log_file, incremental=True)
+
+    assert order == ["measurements", "update_step"]
+
+
+@pytest.mark.asyncio
+async def test_run_larger_than_the_batch_size_is_split(tmp_path, monkeypatch):
+    """The size guard splits a long run instead of sending one huge request."""
+    log_file = tmp_path / "split_run.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False, count=5)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+
+    sent_sizes: list[int] = []
+    minted = iter(f"real-meas-{index}" for index in range(1, 6))
+
+    async def create_measurements(*_, request, **__):
+        sent_sizes.append(len(request.test_measurements))
+        return len(request.test_measurements), [next(minted) for _ in request.test_measurements]
+
+    client.create_test_measurements = create_measurements
+
+    monkeypatch.setattr(test_results_module, "_MEASUREMENT_CREATE_BATCH_SIZE", 2)
+    await client.import_log_file(log_file, incremental=True)
+
+    # Two full batches as the buffer hits the cap, then the fifth measurement is
+    # left over at the end of the walk.
+    assert sent_sizes == [2, 2, 1]
+    assert LogTracking.load(log_file).last_uploaded_line == 7
+
+
+@pytest.mark.asyncio
+async def test_resume_sends_only_the_rest_of_a_measurement_run(tmp_path):
+    """A run whose first measurements already reached the server sends the rest."""
+    log_file = tmp_path / "partial_run.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, measurement_ids = await _build_measurement_log(
+        client, log_file, batched=False, count=4
+    )
+
+    LogTracking(
+        last_uploaded_line=2,
+        id_map={report.id_: "real-report", measurement_ids[0]: "real-meas-1"},
+    ).save(log_file)
+    client.create_test_measurements = AsyncMock(
+        return_value=(3, ["real-meas-2", "real-meas-3", "real-meas-4"])
+    )
+
+    await client.import_log_file(log_file, incremental=True)
+
+    sent = client.create_test_measurements.await_args.kwargs["request"]
+    assert [m.name for m in sent.test_measurements] == ["m2", "m3", "m4"]
+    # The one that already existed keeps the ID the interrupted run recorded.
+    assert LogTracking.load(log_file).id_map[measurement_ids[0]] == "real-meas-1"
+
+
+@pytest.mark.asyncio
+async def test_failed_measurement_run_leaves_the_cursor_alone(tmp_path):
+    """A run that fails is retried whole, so the cursor must not move past it."""
+    log_file = tmp_path / "failed_run.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    client.create_test_measurements = AsyncMock(side_effect=_PermissionDeniedError())
+
+    with pytest.raises(RpcError):
+        await client.import_log_file(log_file, incremental=True)
+
+    tracking = LogTracking.load(log_file)
+    assert tracking.last_uploaded_line == 2
+    assert tracking.id_map == {report.id_: "real-report"}
+
+
+@pytest.mark.asyncio
+async def test_short_batch_response_is_rejected(tmp_path):
+    """Fewer IDs than measurements sent must raise, not mis-pair the IDs.
+
+    IDs are matched to measurements by position. Accepting a short response would
+    record real IDs against the wrong measurements, so a later update would
+    target the wrong row.
+    """
+    log_file = tmp_path / "short_response.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    client.create_test_measurements = AsyncMock(return_value=(2, ["real-meas-1", "real-meas-2"]))
+
+    with pytest.raises(RuntimeError, match="returned 2 IDs for 3 measurements"):
+        await client.import_log_file(log_file, incremental=True)
+
+    assert LogTracking.load(log_file).last_uploaded_line == 2
+
+
+class _UnavailableError(RpcError):
+    """Stand-in for a response lost after the server already committed."""
+
+    def code(self) -> StatusCode:
+        return StatusCode.UNAVAILABLE
+
+
+def _server_measurement(id_: str, *, name: str, numeric_value: float) -> TestMeasurement:
+    return TestMeasurement(
+        id_=id_,
+        test_step_id="real-step",
+        name=name,
+        passed=True,
+        timestamp=T0,
+        numeric_value=numeric_value,
+        measurement_type=TestMeasurementType.DOUBLE,
+    )
+
+
+@pytest.mark.asyncio
+async def test_definite_rejection_does_not_attempt_reconciliation(tmp_path):
+    """A definite rejection skips the reconciliation query entirely.
+
+    PERMISSION_DENIED already tells us the create did not happen; querying
+    first would only spend a round trip finding nothing before the same
+    failure repeats.
+    """
+    log_file = tmp_path / "definite_rejection.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    client.create_test_measurements = AsyncMock(side_effect=_PermissionDeniedError())
+    client.list_all_test_measurements = AsyncMock()
+
+    with pytest.raises(RpcError):
+        await client.import_log_file(log_file, incremental=True)
+
+    client.list_all_test_measurements.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_ambiguous_measurement_failure_reconciles_against_the_server(tmp_path):
+    """A lost-ack failure is resolved by matching the server, not blindly resent.
+
+    ``create_test_measurements`` raising UNAVAILABLE leaves the outcome
+    unknown: the server may have already committed. The server already having
+    all three means nothing more should be created.
+    """
+    log_file = tmp_path / "ambiguous.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, step, measurement_ids = await _build_measurement_log(
+        client, log_file, batched=False, count=3
+    )
+
+    LogTracking(
+        last_uploaded_line=2,
+        id_map={report.id_: "real-report", step.id_: "real-step"},
+    ).save(log_file)
+    client.create_test_measurements = AsyncMock(side_effect=_UnavailableError())
+    client.list_all_test_measurements = AsyncMock(
+        return_value=[
+            _server_measurement(f"real-meas-{i}", name=f"m{i}", numeric_value=float(i))
+            for i in range(1, 4)
+        ]
+    )
+
+    await client.import_log_file(log_file, incremental=True)
+
+    client.create_test_measurements.assert_awaited_once()  # never resent the batch itself
+    tracking = LogTracking.load(log_file)
+    assert tracking.last_uploaded_line == 5
+    assert [tracking.id_map[mid] for mid in measurement_ids] == [
+        "real-meas-1",
+        "real-meas-2",
+        "real-meas-3",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_ambiguous_measurement_failure_recreates_what_is_still_missing(tmp_path):
+    """Only the measurements the server does not already have are recreated."""
+    log_file = tmp_path / "ambiguous_partial.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, step, measurement_ids = await _build_measurement_log(
+        client, log_file, batched=False, count=3
+    )
+
+    LogTracking(
+        last_uploaded_line=2,
+        id_map={report.id_: "real-report", step.id_: "real-step"},
+    ).save(log_file)
+    client.create_test_measurements = AsyncMock(
+        side_effect=[_UnavailableError(), (1, ["real-meas-3"])]
+    )
+    client.list_all_test_measurements = AsyncMock(
+        return_value=[
+            _server_measurement("real-meas-1", name="m1", numeric_value=1.0),
+            _server_measurement("real-meas-2", name="m2", numeric_value=2.0),
+        ]
+    )
+
+    await client.import_log_file(log_file, incremental=True)
+
+    assert client.create_test_measurements.await_count == 2
+    recreated_protos = client.create_test_measurements.await_args.kwargs[
+        "request"
+    ].test_measurements
+    assert [p.name for p in recreated_protos] == ["m3"]
+    tracking = LogTracking.load(log_file)
+    assert [tracking.id_map[mid] for mid in measurement_ids] == [
+        "real-meas-1",
+        "real-meas-2",
+        "real-meas-3",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_reconcile_measurement_batch_matches_by_natural_key():
+    """Matching is step + name + timestamp + value; a value mismatch stays missing."""
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    client.list_all_test_measurements = AsyncMock(
+        return_value=[_server_measurement("real-1", name="m", numeric_value=1.0)]
+    )
+    matching = TestMeasurementCreate(
+        name="m", test_step_id="real-step", passed=True, timestamp=T0, numeric_value=1.0
+    ).to_proto()
+    mismatched = TestMeasurementCreate(
+        name="m", test_step_id="real-step", passed=True, timestamp=T0, numeric_value=2.0
+    ).to_proto()
+
+    matches = await client._reconcile_measurement_batch([matching, mismatched])
+
+    assert matches[0] is not None
+    assert matches[0].id_ == "real-1"
+    assert matches[1] is None
+
+
+@pytest.mark.asyncio
+async def test_reconcile_measurement_batch_does_not_double_match_a_coincidental_duplicate():
+    """Two buffered measurements sharing a key must not both claim the one server row."""
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    client.list_all_test_measurements = AsyncMock(
+        return_value=[_server_measurement("real-1", name="m", numeric_value=1.0)]
+    )
+    proto = TestMeasurementCreate(
+        name="m", test_step_id="real-step", passed=True, timestamp=T0, numeric_value=1.0
+    ).to_proto()
+
+    matches = await client._reconcile_measurement_batch([proto, proto])
+
+    assert matches[0] is not None
+    assert matches[0].id_ == "real-1"
+    assert matches[1] is None
 
 
 @pytest.mark.asyncio
@@ -649,6 +1076,46 @@ async def test_batch_upload_records_measurements(tmp_path):
     ]
 
 
+@pytest.mark.asyncio
+async def test_batch_upload_reconciles_an_ambiguous_measurement_failure(tmp_path):
+    """The batch/default path reconciles too, not just the incremental one.
+
+    ``_batch_import_log_file`` creates measurements through
+    ``_create_measurement_batch_reconciling`` as well, so an ambiguous failure
+    there must recover the same way.
+    """
+    log_file = tmp_path / "batch_ambiguous.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    _, _, measurement_ids = await _build_measurement_log(client, log_file, batched=False, count=3)
+
+    # Report and step must still simulate correctly during the collapse pass,
+    # so this leaves them to the pass-through helper and only overrides
+    # measurements, which the simulate pass for an unbatched log never calls
+    # through the plural endpoint being mocked below.
+    _answer_real_creates(client, report_id="real-report", step_ids=["real-step"])
+    client.create_test_measurements = AsyncMock(
+        side_effect=[_UnavailableError(), (1, ["real-meas-3"])]
+    )
+    client.list_all_test_measurements = AsyncMock(
+        return_value=[
+            _server_measurement("real-meas-1", name="m1", numeric_value=1.0),
+            _server_measurement("real-meas-2", name="m2", numeric_value=2.0),
+        ]
+    )
+
+    result = await client.import_log_file(log_file)
+
+    assert [m.id_ for m in result.measurements] == ["real-meas-1", "real-meas-2", "real-meas-3"]
+    assert result.reconciled_count == 2
+    tracking = LogTracking.load(log_file)
+    assert tracking.complete is True
+    assert [tracking.id_map[mid] for mid in measurement_ids] == [
+        "real-meas-1",
+        "real-meas-2",
+        "real-meas-3",
+    ]
+
+
 @pytest.mark.asyncio
 async def test_batch_upload_tolerates_an_untagged_create(tmp_path):
     """A create logged without a response ID is uploaded but cannot be recorded.
@@ -895,3 +1362,103 @@ def test_cleanup_temp_log_legacy_flat_layout(tmp_path, monkeypatch):
     assert not log.exists()
     assert not tracking.exists()
     assert other.exists()
+
+
+@pytest.mark.asyncio
+async def test_stop_request_ends_replay_after_the_request_in_flight(tmp_path):
+    """A stop request halts replay between requests, with the cursor at what was sent."""
+    log_file = tmp_path / "stop.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False, count=150)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+
+    async def create_and_request_stop(*, request):
+        sent = len(request.test_measurements)
+        request_replay_stop()
+        return sent, [f"real-{i}" for i in range(sent)]
+
+    client.create_test_measurements = AsyncMock(side_effect=create_and_request_stop)
+    try:
+        await client.import_log_file(log_file, incremental=True)
+    finally:
+        _replay_stop.clear()
+
+    # The first full batch went out; the other 50 measurements wait for the next run.
+    client.create_test_measurements.assert_awaited_once()
+    assert LogTracking.load(log_file).last_uploaded_line == 2 + 100
+
+
+@pytest.mark.asyncio
+async def test_stop_file_ends_replay_before_any_request(tmp_path):
+    """The stop file stops replay where a signal cannot reach the worker (Windows)."""
+    log_file = tmp_path / "stop_file.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    report, _, _ = await _build_measurement_log(client, log_file, batched=False)
+
+    LogTracking(last_uploaded_line=2, id_map={report.id_: "real-report"}).save(log_file)
+    LogTracking.stop_path(log_file).touch()
+    client.create_test_measurements = AsyncMock()
+
+    await client.import_log_file(log_file, incremental=True)
+
+    client.create_test_measurements.assert_not_awaited()
+    assert LogTracking.load(log_file).last_uploaded_line == 2
+
+
+@pytest.mark.asyncio
+async def test_resume_ignores_a_stop_file(tmp_path):
+    """A leftover stop file must not cut short the recovery upload of a final log.
+
+    Stopping early would end the walk before the unsent lines and still mark the
+    upload complete.
+    """
+    log_file = tmp_path / "resume_with_stop_file.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+    await _build_measurement_log(client, log_file, batched=False)
+    created = _answer_real_creates(
+        client,
+        report_id="real-report",
+        step_ids=["real-step"],
+        measurement_ids=["real-meas-1", "real-meas-2", "real-meas-3"],
+    )
+    LogTracking.stop_path(log_file).touch()
+
+    await client.import_log_file(log_file)
+
+    assert created == ["report", "s1", "m1", "m2", "m3"]
+    assert LogTracking.load(log_file).complete
+
+
+def test_worker_loop_keeps_the_log_after_a_stop(monkeypatch, tmp_path):
+    """A stopped worker has not uploaded everything, so the log is left for the resume."""
+    from sift_client._internal.pytest_plugin import replay_worker
+
+    cleaned: list[str] = []
+    monkeypatch.setattr(replay_worker, "cleanup_temp_log", cleaned.append)
+    monkeypatch.setattr(replay_worker.select, "select", lambda *_: ([], [], []))
+    log = tmp_path / "stopped.jsonl"
+    LogTracking.stop_path(log).touch()
+    client = MagicMock()
+    client.test_results.import_log_file.return_value = MagicMock(reconciled_count=0)
+
+    replay_worker._incremental_import_loop(client, str(log), keep_log=False)
+
+    assert cleaned == []
+
+
+def test_worker_loop_reports_reconciled_total_and_cleans_up(monkeypatch, tmp_path, capsys):
+    """A clean finish reports the summed reconciled count on stderr, then reclaims the log."""
+    from sift_client._internal.pytest_plugin import replay_worker
+
+    cleaned: list[str] = []
+    monkeypatch.setattr(replay_worker, "cleanup_temp_log", cleaned.append)
+    monkeypatch.setattr(replay_worker.select, "select", lambda *_: ([object()], [], []))
+    client = MagicMock()
+    client.test_results.import_log_file.return_value = MagicMock(reconciled_count=4)
+    log = str(tmp_path / "done.jsonl")
+
+    replay_worker._incremental_import_loop(client, log, keep_log=False)
+
+    assert "SIFT_REPLAY_RECONCILED:4" in capsys.readouterr().err
+    assert cleaned == [log]
diff --git a/python/lib/sift_client/_tests/scripts/__init__.py b/python/lib/sift_client/_tests/scripts/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/python/lib/sift_client/_tests/scripts/test_import_test_result_log.py b/python/lib/sift_client/_tests/scripts/test_import_test_result_log.py
new file mode 100644
index 0000000000..f3a73a1bfe
--- /dev/null
+++ b/python/lib/sift_client/_tests/scripts/test_import_test_result_log.py
@@ -0,0 +1,49 @@
+"""Tests for the ``import-test-result-log`` command's reporting."""
+
+from __future__ import annotations
+
+import sys
+import warnings
+from unittest.mock import MagicMock
+
+import pytest
+
+from sift_client.errors import SiftReplayReconciledWarning
+from sift_client.scripts import import_test_result_log
+
+
+def _run_main(monkeypatch, tmp_path, *, reconciled_count: int) -> None:
+    log_file = tmp_path / "log.jsonl"
+    log_file.write_text("")
+    result = MagicMock(reconciled_count=reconciled_count, steps=[], measurements=[])
+    result.report = None
+    client = MagicMock()
+    client.test_results.import_log_file.return_value = result
+    monkeypatch.setattr(import_test_result_log, "SiftClient", lambda **_: client)
+    monkeypatch.setattr(import_test_result_log, "cleanup_temp_log", lambda _: None)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "import-test-result-log",
+            str(log_file),
+            "--grpc-url",
+            "test.invalid:0",
+            "--rest-url",
+            "http://test.invalid",
+            "--api-key",
+            "test",
+        ],
+    )
+    import_test_result_log.main()
+
+
+def test_reconciled_measurements_are_reported_as_a_warning(monkeypatch, tmp_path) -> None:
+    with pytest.warns(SiftReplayReconciledWarning, match="reconciled 2 measurement"):
+        _run_main(monkeypatch, tmp_path, reconciled_count=2)
+
+
+def test_nothing_reconciled_stays_silent(monkeypatch, tmp_path) -> None:
+    with warnings.catch_warnings():
+        warnings.simplefilter("error", SiftReplayReconciledWarning)
+        _run_main(monkeypatch, tmp_path, reconciled_count=0)
diff --git a/python/lib/sift_client/_tests/util/test_report_context.py b/python/lib/sift_client/_tests/util/test_report_context.py
index 7033f5eee2..c2d8b1fd21 100644
--- a/python/lib/sift_client/_tests/util/test_report_context.py
+++ b/python/lib/sift_client/_tests/util/test_report_context.py
@@ -16,7 +16,7 @@
 import pytest
 
 from sift_client import SiftClient, SiftConnectionConfig
-from sift_client.errors import SiftWarning
+from sift_client.errors import SiftReplayReconciledWarning, SiftWarning
 from sift_client.util.test_results import ReportContext
 
 
@@ -65,6 +65,29 @@ def test_worker_clean_exit_is_silent() -> None:
     assert rc._import_proc.returncode == 0
 
 
+def test_worker_clean_exit_with_reconciled_measurements_warns() -> None:
+    """A reconciled run exits 0, so the marker on stderr is the only signal.
+
+    ``_import_proc.returncode`` can't carry this: 0 is also what a run with
+    nothing to reconcile returns. ``RECONCILED_MARKER`` on stderr is what
+    `__exit__` has to check regardless of exit code.
+    """
+    rc = _make_context(
+        [
+            sys.executable,
+            "-c",
+            "import sys; sys.stderr.write('SIFT_REPLAY_RECONCILED:3'); sys.exit(0)",
+        ]
+    )
+    with pytest.warns(SiftReplayReconciledWarning) as recorded:
+        with rc:
+            pass
+    assert rc._import_proc is not None
+    assert rc._import_proc.returncode == 0
+    messages = "\n".join(str(w.message) for w in recorded)
+    assert "reconciled 3 measurement(s)" in messages
+
+
 def test_worker_timeout_kills_and_warns() -> None:
     """Worker still running at session end → kill + SiftWarning, no raise (case 2)."""
     rc = _make_context([sys.executable, "-c", "import time; time.sleep(30)"], timeout=0.2)
@@ -120,3 +143,100 @@ def test_replay_command_runs_module_through_current_interpreter() -> None:
         "-m",
         "sift_client._internal.pytest_plugin.replay_worker",
     ]
+
+
+_IGNORE_SIGTERM = (
+    "import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(30)"
+)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="SIGTERM is a hard kill on Windows")
+def test_worker_that_ignores_sigterm_is_killed_after_the_grace_period() -> None:
+    """The stop request is polite; a worker that does not honor it is still killed."""
+    rc = _make_context([sys.executable, "-c", _IGNORE_SIGTERM], timeout=0.5)
+    rc._import_proc_stop_grace = 0.3
+    with pytest.warns(SiftWarning):
+        with rc:
+            pass
+    assert rc._import_proc is not None
+    assert rc._import_proc.returncode is not None
+
+
+def test_windows_timeout_asks_the_worker_to_stop_through_a_file(monkeypatch, tmp_path) -> None:
+    """terminate() is a hard kill on Windows, so the stop request is a file the worker polls."""
+    import types
+
+    from sift_client.util.test_results import context_manager
+
+    log = tmp_path / "log.jsonl"
+    watcher = (
+        "import os, sys, time\n"
+        "deadline = time.time() + 20\n"
+        "while not os.path.exists(sys.argv[1]) and time.time() < deadline:\n"
+        "    time.sleep(0.05)\n"
+        "sys.exit(0 if os.path.exists(sys.argv[1]) else 3)\n"
+    )
+    stop_file = str(log) + ".stop"
+    rc = ReportContext(_make_simulate_client(), name="test", log_file=str(log))
+    rc._build_replay_command = lambda: [sys.executable, "-c", watcher, stop_file]  # type: ignore[method-assign]
+    rc._import_proc_timeout = 0.5
+    rc._import_proc_stop_grace = 10.0
+    monkeypatch.setattr(
+        context_manager, "sys", types.SimpleNamespace(**{**vars(sys), "platform": "win32"})
+    )
+    with pytest.warns(SiftWarning):
+        with rc:
+            pass
+    assert rc._import_proc is not None
+    # The worker saw the file and left on its own, rather than being killed.
+    assert rc._import_proc.returncode == 0
+    # The file does not outlive the run, or it would stop the recovery upload.
+    assert not (tmp_path / "log.jsonl.stop").exists()
+
+
+def test_stale_stop_file_is_removed_when_the_worker_starts(tmp_path) -> None:
+    log = tmp_path / "log.jsonl"
+    stop_file = tmp_path / "log.jsonl.stop"
+    stop_file.touch()
+    check = "import os, sys; sys.exit(1 if os.path.exists(sys.argv[1]) else 0)"
+    rc = ReportContext(_make_simulate_client(), name="test", log_file=str(log))
+    rc._build_replay_command = lambda: [sys.executable, "-c", check, str(stop_file)]  # type: ignore[method-assign]
+    rc._import_proc_timeout = 5.0
+    with rc:
+        pass
+    assert rc._import_proc is not None
+    assert rc._import_proc.returncode == 0
+    assert not stop_file.exists()
+
+
+def test_pending_upload_timeout_is_set_only_while_a_worker_can_be_waited_on() -> None:
+    rc = _make_context([sys.executable, "-c", "pass"], timeout=5.0)
+    assert rc.pending_upload_timeout is None
+    with rc:
+        assert rc.pending_upload_timeout == 5.0
+    assert rc.pending_upload_timeout is None
+
+
+@pytest.mark.parametrize(
+    ("timeout", "verbose", "expected_lines"),
+    [(120.0, 0, 1), (None, 0, 0), (120.0, -1, 0)],
+    ids=["waiting", "no-worker", "quiet"],
+)
+def test_pending_upload_message(timeout, verbose, expected_lines) -> None:
+    import types
+
+    from sift_client.pytest_plugin import _announce_pending_upload
+
+    lines: list[str] = []
+    reporter = types.SimpleNamespace(write_line=lines.append)
+    config = types.SimpleNamespace(
+        pluginmanager=types.SimpleNamespace(get_plugin=lambda _name: reporter),
+        option=types.SimpleNamespace(verbose=verbose),
+    )
+    context = types.SimpleNamespace(pending_upload_timeout=timeout)
+
+    _announce_pending_upload(config, context)  # type: ignore[arg-type]
+
+    assert len(lines) == expected_lines
+    if expected_lines:
+        assert "120s" in lines[0]
diff --git a/python/lib/sift_client/errors.py b/python/lib/sift_client/errors.py
index b370b110a9..773657274b 100644
--- a/python/lib/sift_client/errors.py
+++ b/python/lib/sift_client/errors.py
@@ -23,6 +23,15 @@ class SiftReportLifecycleWarning(SiftWarning):
     """Warning for test-results updates that are out of the intended lifecycle order."""
 
 
+class SiftReplayReconciledWarning(SiftWarning):
+    """A test-results upload recovered from an ambiguous retry.
+
+    Issued once per run, after a batch measurement create failed with an
+    unknown outcome and the retry found some or all of the batch already on
+    the server.
+    """
+
+
 class SiftCredentialsError(ValueError):
     """Raised when Sift credentials cannot be resolved.
 
diff --git a/python/lib/sift_client/pytest_plugin.py b/python/lib/sift_client/pytest_plugin.py
index a143064ba5..e3f7d35097 100644
--- a/python/lib/sift_client/pytest_plugin.py
+++ b/python/lib/sift_client/pytest_plugin.py
@@ -665,9 +665,19 @@ def pytest_sessionfinish(session: pytest.Session, exitstatus: int):
     yield
     finalize_parents()
     if REPORT_CONTEXT is not None:
+        _announce_pending_upload(session.config, REPORT_CONTEXT)
         REPORT_CONTEXT.finalize()
 
 
+def _announce_pending_upload(config: pytest.Config, context: ReportContext) -> None:
+    """Tell the terminal the session is waiting on the background upload, so it does not look hung."""
+    timeout = context.pending_upload_timeout
+    reporter = config.pluginmanager.get_plugin("terminalreporter")
+    if timeout is None or reporter is None or _verbosity(config) < 0:
+        return
+    reporter.write_line(f"Sift: uploading remaining test results (waiting up to {timeout:.0f}s)...")
+
+
 def _verbosity(config: pytest.Config) -> int:
     """Global verbosity (the ``-v`` count minus ``-q``), pytest 7/8/9-compatible.
 
diff --git a/python/lib/sift_client/scripts/import_test_result_log.py b/python/lib/sift_client/scripts/import_test_result_log.py
index 78876dfaca..051738f7da 100644
--- a/python/lib/sift_client/scripts/import_test_result_log.py
+++ b/python/lib/sift_client/scripts/import_test_result_log.py
@@ -5,13 +5,15 @@
 import argparse
 import logging
 import os
+import warnings
 from pathlib import Path
 from typing import TYPE_CHECKING
 
 from sift_client import SiftClient, SiftConnectionConfig
 from sift_client._internal.low_level_wrappers._test_results_log import LogTracking
-from sift_client._internal.pytest_plugin.audit_log import log_event
+from sift_client._internal.pytest_plugin.audit_log import log_event, reconciled_warning_message
 from sift_client._internal.pytest_plugin.replay_worker import cleanup_temp_log
+from sift_client.errors import SiftReplayReconciledWarning
 from sift_client.util.test_results.context_manager import log_replay_instructions
 
 if TYPE_CHECKING:
@@ -110,6 +112,13 @@ def main() -> None:
         log_replay_instructions(args.log_file)
         raise
 
+    if result and result.reconciled_count:
+        warnings.warn(
+            reconciled_warning_message("import-test-result-log", result.reconciled_count),
+            SiftReplayReconciledWarning,
+            stacklevel=2,
+        )
+
     if result:
         _print_result(result)
 
diff --git a/python/lib/sift_client/util/test_results/context_manager.py b/python/lib/sift_client/util/test_results/context_manager.py
index e50eca2110..9e95f03366 100644
--- a/python/lib/sift_client/util/test_results/context_manager.py
+++ b/python/lib/sift_client/util/test_results/context_manager.py
@@ -16,8 +16,14 @@
 
 import numpy as np
 
-from sift_client._internal.pytest_plugin.audit_log import _make_session_dir, log_event
-from sift_client.errors import SiftReportLifecycleWarning, SiftWarning
+from sift_client._internal.low_level_wrappers._test_results_log import LogTracking
+from sift_client._internal.pytest_plugin.audit_log import (
+    RECONCILED_MARKER,
+    _make_session_dir,
+    log_event,
+    reconciled_warning_message,
+)
+from sift_client.errors import SiftReplayReconciledWarning, SiftReportLifecycleWarning, SiftWarning
 from sift_client.sift_types.test_report import (
     ErrorInfo,
     NumericBounds,
@@ -96,6 +102,21 @@ def log_replay_instructions(log_file: str | Path | None) -> None:
     )
 
 
+def _parse_reconciled_count(stderr_text: str) -> int:
+    """Read the replay worker's reconciled-measurement count off its stderr.
+
+    Returns 0 if the ``RECONCILED_MARKER`` line is missing or malformed.
+    """
+    prefix = f"{RECONCILED_MARKER}:"
+    for line in stderr_text.splitlines():
+        if line.startswith(prefix):
+            try:
+                return int(line[len(prefix) :])
+            except ValueError:
+                return 0
+    return 0
+
+
 @contextmanager
 def _quiet_fork_stderr():
     """Redirect fd 2 to /dev/null across a ``fork()`` to discard gRPC's prefork notices.
@@ -230,7 +251,10 @@ class ReportContext(AbstractContextManager):
     # the JSONL backlog at session end before killing it. Tests substitute
     # a smaller value (via ``_make_context`` patching) so they don't wait
     # the full window for the timeout branch to trigger.
-    _import_proc_timeout: float = 30.0
+    _import_proc_timeout: float = 120.0
+    # Seconds the worker gets to stop after SIGTERM, which lets it finish the
+    # request in flight and save its progress, before it is killed.
+    _import_proc_stop_grace: float = 30.0
 
     def __init__(
         self,
@@ -391,6 +415,8 @@ def _open_import_proc(self):
         # Redact the API key before logging; never log the raw argv.
         api_key = self.client.grpc_client._config.api_key
         safe = ["***" if a == api_key else a for a in self._build_replay_command()]
+        # A stop request left by an earlier run on the same log path.
+        LogTracking.stop_path(self.log_file).unlink(missing_ok=True)
         log_event(logger, logging.INFO, "replay.start", log=self.log_file)
         log_event(logger, logging.DEBUG, "replay.cmd", argv=" ".join(safe))
         try:
@@ -453,8 +479,9 @@ def finalize(self) -> bool:
             # `import-test-result-log <path>`:
             #   1. Exits cleanly (returncode 0). Silent.
             #   2. Still running after the grace window (TimeoutExpired).
-            #      Healthy worker with a large backlog; kill and surface
-            #      replay instructions. 30 seconds is enough for a normal
+            #      Healthy worker with a large backlog; ask it to stop so it
+            #      saves its progress, kill it if it does not, and surface
+            #      replay instructions. The default window is enough for a normal
             #      test suite to drain; pathological backlogs should opt
             #      into inline mode (`--no-sift-log-file`) instead.
             #   3. Exited with non-zero. Connection failures and API call
@@ -464,8 +491,22 @@ def finalize(self) -> bool:
             try:
                 _, stderr_bytes = self._import_proc.communicate(timeout=self._import_proc_timeout)
             except subprocess.TimeoutExpired:
-                self._import_proc.kill()
-                self._import_proc.wait()
+                stop_file = (
+                    LogTracking.stop_path(self.log_file) if self.log_file is not None else None
+                )
+                if sys.platform == "win32" and stop_file is not None:
+                    # terminate() is a hard kill on Windows, so ask through a
+                    # file the importer checks between requests.
+                    stop_file.touch()
+                else:
+                    self._import_proc.terminate()
+                try:
+                    self._import_proc.communicate(timeout=self._import_proc_stop_grace)
+                except subprocess.TimeoutExpired:
+                    self._import_proc.kill()
+                    self._import_proc.wait()
+                if stop_file is not None:
+                    stop_file.unlink(missing_ok=True)
                 self.replay_incomplete = True
                 log_event(
                     logger,
@@ -476,18 +517,28 @@ def finalize(self) -> bool:
                 )
                 warnings.warn(
                     f"Sift import worker did not exit in "
-                    f"{self._import_proc_timeout}s; killing it. "
+                    f"{self._import_proc_timeout}s; stopping it. "
                     "Local log file is preserved for manual replay.",
                     SiftWarning,
                     stacklevel=2,
                 )
                 log_replay_instructions(self.log_file)
                 return True  # Ensures the session is marked as passed in pytest
+            stderr_text = (
+                stderr_bytes.decode("utf-8", errors="replace").strip() if stderr_bytes else ""
+            )
+            # A reconciled run still exits 0, so this has to run before the
+            # returncode branch below rather than only in its failure half.
+            reconciled_count = _parse_reconciled_count(stderr_text)
+            if reconciled_count:
+                log_event(logger, logging.WARNING, "replay.reconciled", count=reconciled_count)
+                warnings.warn(
+                    reconciled_warning_message("Sift import worker", reconciled_count),
+                    SiftReplayReconciledWarning,
+                    stacklevel=2,
+                )
             if self._import_proc.returncode != 0:
                 self.replay_incomplete = True
-                stderr_text = (
-                    stderr_bytes.decode("utf-8", errors="replace").strip() if stderr_bytes else ""
-                )
                 log_event(
                     logger,
                     logging.WARNING,
@@ -508,6 +559,13 @@ def finalize(self) -> bool:
 
         return True
 
+    @property
+    def pending_upload_timeout(self) -> float | None:
+        """Seconds ``finalize`` will wait for the import worker, or None if it will not wait."""
+        if self._finalized or self._import_proc is None:
+            return None
+        return self._import_proc_timeout
+
     @property
     def is_simulated(self) -> bool:
         """True when this context's report came from the simulate path.