From 07f59adf10d4d0540a33952e091712736fa4bf4c Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Wed, 30 Sep 2026 15:40:53 -0700
Subject: [PATCH 1/6] python(feat): batch measurement creates during log replay

A test-results log records one CreateTestMeasurement line per measurement, and
replay sent one request for each. Measurements are the bulk of a log, so a
large log took one round trip per measurement to upload.

Replay now coalesces a run of consecutive measurement lines into one
CreateTestMeasurements call. A run ends at the next non-measurement line, at
the size guard, or at the end of the log, so lines still reach the server in
log order and the cursor advances by the whole run once its call returns.

On a 13,631-line log this takes the requests from 13,628 to 4,021, and the
upload from about 12 minutes to about 3.5 at a 53ms round trip. The sidecar is
written once per run rather than once per line, which cuts its cost with them.

The log format does not change, so a log written by an earlier version gets the
same speedup. Step creates and updates still take one request each.
---
 .../low_level_wrappers/_test_results_log.py   |  14 +-
 .../low_level_wrappers/test_results.py        | 248 ++++++++++++---
 .../test_incremental_replay.py                | 299 +++++++++++++++++-
 3 files changed, 499 insertions(+), 62 deletions(-)

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..2d595841ff 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
@@ -92,13 +92,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
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..cf56ebc78c 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
@@ -52,7 +52,7 @@
 )
 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,
@@ -82,6 +82,14 @@
 # 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 leaves a wide margin under the
+# 4MB gRPC message limit. The same number bounds two other things, which is why
+# it is not worth tuning for throughput alone: it is the flush threshold for a
+# run of consecutive measurement log lines, and it is how much progress a crash
+# can cost, since the sidecar is written once per request.
+_MEASUREMENT_CREATE_BATCH_SIZE = 100
+
 
 class _EntryIds(NamedTuple):
     """The entity a replayed log entry acted on, for the audit trace.
@@ -860,6 +868,51 @@ 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, in request order.
+
+        Callers pair the result with whatever they hold per input, so the order
+        and the length of the result both matter.
+
+        The batch response carries IDs rather than entities, so each entity is
+        rebuilt from the proto that was sent plus its assigned ID. Fields the
+        server normalizes are therefore the values as sent, not as stored.
+
+        Raises:
+            RuntimeError: If the response returns an ID count other than the
+                number of measurements sent. IDs are paired with inputs by
+                position, so a short response would attach real IDs to the wrong
+                measurements and send later updates to the wrong rows.
+        """
+        _, 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 _create_measurements_in_batches(
+        self, protos: list[TestMeasurementProto]
+    ) -> list[TestMeasurement]:
+        """Create measurements over as few batch calls as the size guard allows.
+
+        For a caller that holds more measurements than one request should carry,
+        such as a logged batch line of arbitrary length.
+        """
+        created: list[TestMeasurement] = []
+        for chunk in chunked(protos, _MEASUREMENT_CREATE_BATCH_SIZE):
+            created.extend(await self._create_measurement_batch(chunk))
+        return created
+
     async def list_test_measurements(
         self,
         *,
@@ -1220,6 +1273,48 @@ 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,
+    ) -> list[str]:
+        """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, so it is left
+        alone rather than created a second time. A group can be part done like
+        that because a batch upload creates measurements in collapsed order, one
+        at a time, so it can stop partway through what one log line covers.
+
+        Every group goes through the batch endpoint, including a group of one.
+        Choosing an endpoint by group size would mean entities built two ways,
+        since the batch endpoint returns IDs while the single endpoint returns the
+        server's entity, and a resume would send the same measurement through a
+        different endpoint than the attempt before it.
+
+        Returns the real IDs created, in the order they were sent, for the audit
+        trail.
+        """
+        pending = [
+            (logged_id, tm) for logged_id, tm in pairs if not (logged_id and logged_id in id_map)
+        ]
+        if not pending:
+            return []
+        created = 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
+
     async def _replay_create_measurements(
         self,
         json_str: str,
@@ -1234,8 +1329,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)
@@ -1246,26 +1341,13 @@ async def _replay_create_measurements(
                 state.measurements_order.append(meas._id_or_error)
                 created_ids.append(meas._id_or_error)
         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 = 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
@@ -1393,19 +1475,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 +1517,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)
 
+        # Measurements are the bulk of a log, so they go out in batches rather
+        # than one call each. One request per chunk means 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)
+        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 = await self._create_measurement_batch(protos)
+            real_measurements.extend(created)
+            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
@@ -1470,10 +1565,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 +1592,81 @@ async def _incremental_import_log_file(
         state = _ReplayState()
 
         raw_lines = await _read_log_lines(log_path)
+        # A run of consecutive single-measurement lines collects here and reaches
+        # the server as one batch call. Measurements are the bulk of a log, and
+        # one round-trip each is what makes a large log slow to upload.
+        buffered_measurements: list[tuple[str | None, TestMeasurementProto]] = []
+
+        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. That widens an existing
+            exposure: a create is not idempotent, so a response lost after the
+            server committed duplicates the measurements on retry. The window was
+            one measurement per line and is now one run. Closing it needs a
+            request ID the API does not accept yet.
+            """
+            if not buffered_measurements:
+                return
+            first_line = tracking.last_uploaded_line + 1
+            try:
+                real_ids = 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
+
+            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 request_type == "CreateTestMeasurement":
+                # Parsed on the way into the buffer, so the flush has one job. 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(
@@ -1542,6 +1709,9 @@ async def _incremental_import_log_file(
                 idmap=len(id_map),
             )
 
+        # A log ending in measurement lines leaves a run buffered.
+        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.
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..811b184d9c 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,11 +14,12 @@
 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 import test_results as test_results_module
 from sift_client._internal.low_level_wrappers._test_results_log import LogTracking
 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.
@@ -287,16 +288,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 +319,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 +531,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 +558,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 +611,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 +630,262 @@ 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
 
 
 @pytest.mark.asyncio

From 95bade9b91149a6acff3665b99c717faf6d36102 Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Wed, 14 Oct 2026 13:30:47 -0700
Subject: [PATCH 2/6] bulk upload measurements and roconciliation check

---
 .../low_level_wrappers/_test_results_log.py   |   8 +
 .../low_level_wrappers/test_results.py        | 196 ++++++++++++++++--
 .../_internal/pytest_plugin/audit_log.py      |  19 ++
 .../_internal/pytest_plugin/replay_worker.py  |  13 +-
 .../test_incremental_replay.py                | 196 ++++++++++++++++++
 .../_tests/util/test_report_context.py        |  25 ++-
 python/lib/sift_client/errors.py              |   8 +
 .../scripts/import_test_result_log.py         |  11 +-
 .../util/test_results/context_manager.py      |  45 +++-
 9 files changed, 495 insertions(+), 26 deletions(-)

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 2d595841ff..35bb941d90 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
@@ -207,6 +207,14 @@ 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 in a way that left the server's
+    state ambiguous (see ``_create_measurement_batch_reconciling``), and the
+    server turned out to already have some or all of them. Callers use this to
+    decide whether to warn.
+    """
 
 
 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 cf56ebc78c..720d44ac2d 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,7 @@
 import logging
 import uuid
 import warnings
+from datetime import timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar, cast
 
@@ -66,6 +67,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
@@ -90,6 +92,26 @@
 # can cost, since the sidecar is written once per request.
 _MEASUREMENT_CREATE_BATCH_SIZE = 100
 
+# Status codes that leave the server's handling of a request unknown: the
+# request may never have arrived, or it may have been committed with only the
+# response lost. A code outside this set is a definite rejection - the server
+# told us the create did not happen - so reconciling would only spend a round
+# trip finding nothing before the same failure repeats.
+_AMBIGUOUS_STATUS_CODES = frozenset(
+    {StatusCode.UNAVAILABLE, StatusCode.DEADLINE_EXCEEDED, StatusCode.CANCELLED, StatusCode.UNKNOWN}
+)
+
+
+def _is_ambiguous_grpc_failure(exc: Exception) -> bool:
+    """True when *exc* is an RpcError whose status is in ``_AMBIGUOUS_STATUS_CODES``.
+
+    Anything without a ``code()`` - a local bug such as a response-length
+    mismatch, a plain ``TypeError`` - is treated as definite, not ambiguous:
+    there is no server status to be unsure about.
+    """
+    code = getattr(exc, "code", None)
+    return callable(code) and code() in _AMBIGUOUS_STATUS_CODES
+
 
 class _EntryIds(NamedTuple):
     """The entity a replayed log entry acted on, for the audit trace.
@@ -102,11 +124,17 @@ 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, after a batch create's outcome came back unknown.
+    Zero for every entry type except a logged ``CreateTestMeasurements`` line
+    replayed for real; 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):
@@ -900,18 +928,134 @@ async def _create_measurement_batch(
             created.append(TestMeasurement._from_proto(proto))
         return created
 
+    async def _reconcile_measurement_batch(
+        self, protos: list[TestMeasurementProto]
+    ) -> list[TestMeasurement | None]:
+        """Match this batch's measurements against what the server already has.
+
+        There is no request ID the API accepts to confirm a create landed, so
+        this falls back to the closest thing the proto has to a natural key:
+        ``test_step_id``, ``name``, ``timestamp``, and whichever value field is
+        set. Timestamps compare as ``ToDatetime(tzinfo=timezone.utc)`` on both
+        sides, the same conversion ``TestMeasurement._from_proto`` uses.
+
+        Matching is by multiset, not a plain lookup: two measurements in the
+        same batch can legitimately share an identical key, and a plain dict
+        would let both claim the same server-side row. Each key's candidates
+        are consumed one at a time, so a second coincidental duplicate is
+        correctly treated as still missing.
+
+        Returns one slot per input proto, in order; ``None`` where nothing on
+        the server matched it.
+        """
+        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))
+        )
+
+        def value_of(numeric: float | None, string: str | None, boolean: bool | None):
+            if numeric is not None:
+                return numeric
+            if string is not None:
+                return string
+            return boolean
+
+        candidates: dict[tuple[str, str, Any, Any], list[TestMeasurement]] = {}
+        for measurement in existing:
+            key = (
+                measurement.test_step_id,
+                measurement.name,
+                measurement.timestamp,
+                value_of(
+                    measurement.numeric_value, measurement.string_value, measurement.boolean_value
+                ),
+            )
+            candidates.setdefault(key, []).append(measurement)
+
+        matches: list[TestMeasurement | None] = []
+        for proto in protos:
+            if proto.HasField("numeric_value"):
+                value = value_of(proto.numeric_value, None, None)
+            elif proto.HasField("string_value"):
+                value = value_of(None, proto.string_value, None)
+            elif proto.HasField("boolean_value"):
+                value = value_of(None, None, proto.boolean_value)
+            else:
+                value = None
+            key = (
+                proto.test_step_id,
+                proto.name,
+                proto.timestamp.ToDatetime(tzinfo=timezone.utc),
+                value,
+            )
+            bucket = candidates.get(key)
+            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 call's outcome is unknown.
+
+        ``CreateTestMeasurements`` is not idempotent and the API has no request
+        ID to make a retry safe. A failure whose status is in
+        ``_AMBIGUOUS_STATUS_CODES`` could mean the request never reached the
+        server, or that it did and only the response was lost; blindly
+        resending would duplicate the whole batch in the second case. This
+        reconciles by natural key (see ``_reconcile_measurement_batch``) first
+        and only recreates what is genuinely missing. If the reconciliation
+        query itself fails, that exception propagates too.
+
+        A definite rejection (see ``_is_ambiguous_grpc_failure``) propagates
+        unchanged, including a local bug such as the ID-count mismatch
+        ``_create_measurement_batch`` raises itself.
+
+        Returns the measurements in input order, plus how many were matched on
+        the server instead of recreated (0 on the common, no-failure path).
+        """
+        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]
-    ) -> list[TestMeasurement]:
+    ) -> tuple[list[TestMeasurement], int]:
         """Create measurements over as few batch calls as the size guard allows.
 
         For a caller that holds more measurements than one request should carry,
-        such as a logged batch line of arbitrary length.
+        such as a logged batch line of arbitrary length. 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):
-            created.extend(await self._create_measurement_batch(chunk))
-        return created
+            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,
@@ -1279,7 +1423,7 @@ async def _replay_create_measurement_group(
         *,
         id_map: dict[str, str],
         state: _ReplayState,
-    ) -> list[str]:
+    ) -> 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,
@@ -1296,14 +1440,17 @@ async def _replay_create_measurement_group(
         different endpoint than the attempt before it.
 
         Returns the real IDs created, in the order they were sent, for the audit
-        trail.
+        trail, 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 []
-        created = await self._create_measurements_in_batches([tm for _, tm in 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):
@@ -1313,7 +1460,7 @@ async def _replay_create_measurement_group(
             state.measurements_by_id[real_id] = measurement
             state.measurements_order.append(real_id)
             real_ids.append(real_id)
-        return real_ids
+        return real_ids, reconciled_count
 
     async def _replay_create_measurements(
         self,
@@ -1340,19 +1487,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:
             pairs = [
                 (original_ids[i] if i < len(original_ids) else None, tm)
                 for i, tm in enumerate(request.test_measurements)
             ]
-            created_ids = await self._replay_create_measurement_group(
+            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,
@@ -1522,6 +1675,7 @@ def record_created(simulated_id: str, real_id: str, *, save: bool = True) -> Non
         # once per chunk, so a crash loses at most one request's IDs and the
         # resume recreates those measurements.
         real_measurements: list[TestMeasurement] = []
+        reconciled_count = 0
         for chunk_ids in chunked(state.measurements_order, _MEASUREMENT_CREATE_BATCH_SIZE):
             protos = []
             for sim_measurement_id in chunk_ids:
@@ -1533,8 +1687,9 @@ def record_created(simulated_id: str, real_id: str, *, save: bool = True) -> Non
                     sim_measurement, real_step_id
                 )
                 protos.append(measurement_create.to_proto())
-            created = await self._create_measurement_batch(protos)
+            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)
@@ -1549,6 +1704,7 @@ def record_created(simulated_id: str, real_id: str, *, save: bool = True) -> Non
             report=real_report,
             steps=real_steps,
             measurements=real_measurements,
+            reconciled_count=reconciled_count,
         )
 
     # ------------------------------------------------------------------
@@ -1596,22 +1752,23 @@ async def _incremental_import_log_file(
         # the server as one batch call. Measurements are the bulk of a log, and
         # one round-trip each is what makes a large log slow to upload.
         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. That widens an existing
-            exposure: a create is not idempotent, so a response lost after the
-            server committed duplicates the measurements on retry. The window was
-            one measurement per line and is now one run. Closing it needs a
-            request ID the API does not accept yet.
+            every line of the run to be retried next tick.
+            ``_replay_create_measurement_group`` already reconciles an ambiguous
+            failure against the server; this still raises only when that
+            reconciliation itself can't be answered.
             """
             if not buffered_measurements:
                 return
+            nonlocal reconciled_count
             first_line = tracking.last_uploaded_line + 1
             try:
-                real_ids = await self._replay_create_measurement_group(
+                real_ids, batch_reconciled = await self._replay_create_measurement_group(
                     buffered_measurements, id_map=id_map, state=state
                 )
             except Exception as exc:
@@ -1625,6 +1782,7 @@ async def flush_measurements() -> None:
                     error=repr(exc),
                 )
                 raise
+            reconciled_count += batch_reconciled
 
             tracking.last_uploaded_line += len(buffered_measurements)
             tracking.save(log_path)
@@ -1692,6 +1850,7 @@ async def flush_measurements() -> None:
 
             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,
@@ -1728,6 +1887,7 @@ async def flush_measurements() -> None:
             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..26d43b70e4 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,12 @@
 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; see
+# ``_create_measurement_batch_reconciling``. Lives here rather than in
+# ``replay_worker`` or ``context_manager`` because those two 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 +96,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..e349d8f273 100644
--- a/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py
+++ b/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py
@@ -25,7 +25,7 @@
 
 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 RECONCILED_MARKER, log_event
 from sift_client.util.test_results.context_manager import log_replay_instructions
 
 if TYPE_CHECKING:
@@ -77,14 +77,25 @@ 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.
+
+    A tick that reconciled a measurement batch against the server (see
+    ``_create_measurement_batch_reconciling``) still exits cleanly, so the
+    count is summed across ticks and reported on stderr once, at session
+    end - 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 result is not None:
+            total_reconciled += result.reconciled_count
         if received_signal:
             break
     log_event(logger, logging.INFO, "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:
         cleanup_temp_log(log_file)
     return result
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 811b184d9c..14745b024a 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
@@ -888,6 +888,162 @@ async def test_short_batch_response_is_rejected(tmp_path):
     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
 async def test_batch_upload_records_measurements(tmp_path):
     """Measurements created by a batch upload are recorded like steps are.
@@ -916,6 +1072,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.
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..c3e8304bd7 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)
diff --git a/python/lib/sift_client/errors.py b/python/lib/sift_client/errors.py
index b370b110a9..2cf53d5a71 100644
--- a/python/lib/sift_client/errors.py
+++ b/python/lib/sift_client/errors.py
@@ -23,6 +23,14 @@ 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.
+
+    Raised once per run, after a batch measurement create's response was
+    lost 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/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..29e152958a 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,13 @@
 
 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.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 +101,26 @@ 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.
+
+    The worker's own exit code can't carry this: a reconciled run is a clean
+    exit, with nothing in the usual non-zero-exit stderr handling to surface
+    it. ``RECONCILED_MARKER`` is the only channel back.
+
+    Returns 0 for a line that is missing or malformed, rather than raising
+    over a worker detail that was never essential to parse.
+    """
+    for line in stderr_text.splitlines():
+        prefix = f"{RECONCILED_MARKER}:"
+        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.
@@ -483,11 +508,21 @@ def finalize(self) -> bool:
                 )
                 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,

From 3fb86dac6bfe0e3c4686b9ff9351717e031c2f85 Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Wed, 14 Oct 2026 14:51:12 -0700
Subject: [PATCH 3/6] chagenlog

---
 python/changelog.d/810.feat.md | 3 +++
 1 file changed, 3 insertions(+)
 create mode 100644 python/changelog.d/810.feat.md

diff --git a/python/changelog.d/810.feat.md b/python/changelog.d/810.feat.md
new file mode 100644
index 0000000000..b8c2fe0761
--- /dev/null
+++ b/python/changelog.d/810.feat.md
@@ -0,0 +1,3 @@
+Uploading a test-results log is faster. A run of consecutive measurement lines now reaches the server through `CreateTestMeasurements` as one call, rather than one call per measurement. On a 13,631-line log this cut the calls from 13,628 to 4,021 and the upload from about 12 minutes to about 3.5. Step creates and updates still take one call each.
+
+  A batch measurement create that fails after the server already committed the request, but lost the response, no longer risks duplicating the batch on retry: the retry checks the server first by test step, name, timestamp, and value, and only recreates what's genuinely missing, reporting it once per run as the new `SiftReplayReconciledWarning`.

From 6b614862cc5f8058702f397bf8cfb41278f7b97b Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Wed, 14 Oct 2026 17:16:23 -0700
Subject: [PATCH 4/6] clean up pass

---
 python/changelog.d/810.feat.md                |   4 +-
 .../low_level_wrappers/_test_results_log.py   |   7 +-
 .../low_level_wrappers/test_results.py        | 128 ++++++------------
 .../_internal/pytest_plugin/audit_log.py      |   8 +-
 .../_internal/pytest_plugin/replay_worker.py  |   8 +-
 python/lib/sift_client/errors.py              |   5 +-
 .../util/test_results/context_manager.py      |   7 +-
 7 files changed, 56 insertions(+), 111 deletions(-)

diff --git a/python/changelog.d/810.feat.md b/python/changelog.d/810.feat.md
index b8c2fe0761..150eacaca6 100644
--- a/python/changelog.d/810.feat.md
+++ b/python/changelog.d/810.feat.md
@@ -1,3 +1 @@
-Uploading a test-results log is faster. A run of consecutive measurement lines now reaches the server through `CreateTestMeasurements` as one call, rather than one call per measurement. On a 13,631-line log this cut the calls from 13,628 to 4,021 and the upload from about 12 minutes to about 3.5. Step creates and updates still take one call each.
-
-  A batch measurement create that fails after the server already committed the request, but lost the response, no longer risks duplicating the batch on retry: the retry checks the server first by test step, name, timestamp, and value, and only recreates what's genuinely missing, reporting it once per run as the new `SiftReplayReconciledWarning`.
+Test-results log uploads are faster: consecutive measurements are now sent in one `CreateTestMeasurements` call, and a retry after an unknown outcome checks the server first and reports it with the new `SiftReplayReconciledWarning`.
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 35bb941d90..d795dd79f4 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
@@ -210,10 +210,9 @@ class ReplayResult:
     reconciled_count: int = 0
     """How many measurements were matched on the server instead of recreated.
 
-    Nonzero only after a batch create failed in a way that left the server's
-    state ambiguous (see ``_create_measurement_batch_reconciling``), and the
-    server turned out to already have some or all of them. Callers use this to
-    decide whether to warn.
+    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.
     """
 
 
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 720d44ac2d..b11d94f770 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
@@ -85,18 +85,14 @@
 _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 leaves a wide margin under the
-# 4MB gRPC message limit. The same number bounds two other things, which is why
-# it is not worth tuning for throughput alone: it is the flush threshold for a
-# run of consecutive measurement log lines, and it is how much progress a crash
-# can cost, since the sidecar is written once per request.
+# 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 that leave the server's handling of a request unknown: the
-# request may never have arrived, or it may have been committed with only the
-# response lost. A code outside this set is a definite rejection - the server
-# told us the create did not happen - so reconciling would only spend a round
-# trip finding nothing before the same failure repeats.
+# 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}
 )
@@ -105,9 +101,7 @@
 def _is_ambiguous_grpc_failure(exc: Exception) -> bool:
     """True when *exc* is an RpcError whose status is in ``_AMBIGUOUS_STATUS_CODES``.
 
-    Anything without a ``code()`` - a local bug such as a response-length
-    mismatch, a plain ``TypeError`` - is treated as definite, not ambiguous:
-    there is no server status to be unsure about.
+    An exception without a ``code()``, such as a local ``TypeError``, is not ambiguous.
     """
     code = getattr(exc, "code", None)
     return callable(code) and code() in _AMBIGUOUS_STATUS_CODES
@@ -125,10 +119,8 @@ 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, after a batch create's outcome came back unknown.
-    Zero for every entry type except a logged ``CreateTestMeasurements`` line
-    replayed for real; see ``_create_measurement_batch_reconciling``.
+    ``reconciled`` counts measurements this entry matched on the server instead
+    of recreating; see ``_create_measurement_batch_reconciling``.
     """
 
     sim_id: str | None = None
@@ -899,20 +891,14 @@ async def create_test_measurements(
     async def _create_measurement_batch(
         self, protos: list[TestMeasurementProto]
     ) -> list[TestMeasurement]:
-        """Create measurements in one batch call, in request order.
+        """Create measurements in one batch call, returned in request order.
 
-        Callers pair the result with whatever they hold per input, so the order
-        and the length of the result both matter.
-
-        The batch response carries IDs rather than entities, so each entity is
-        rebuilt from the proto that was sent plus its assigned ID. Fields the
-        server normalizes are therefore the values as sent, not as stored.
+        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 returns an ID count other than the
-                number of measurements sent. IDs are paired with inputs by
-                position, so a short response would attach real IDs to the wrong
-                measurements and send later updates to the wrong rows.
+            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)
@@ -931,22 +917,14 @@ async def _create_measurement_batch(
     async def _reconcile_measurement_batch(
         self, protos: list[TestMeasurementProto]
     ) -> list[TestMeasurement | None]:
-        """Match this batch's measurements against what the server already has.
-
-        There is no request ID the API accepts to confirm a create landed, so
-        this falls back to the closest thing the proto has to a natural key:
-        ``test_step_id``, ``name``, ``timestamp``, and whichever value field is
-        set. Timestamps compare as ``ToDatetime(tzinfo=timezone.utc)`` on both
-        sides, the same conversion ``TestMeasurement._from_proto`` uses.
+        """Match a batch's measurements against what the server already has.
 
-        Matching is by multiset, not a plain lookup: two measurements in the
-        same batch can legitimately share an identical key, and a plain dict
-        would let both claim the same server-side row. Each key's candidates
-        are consumed one at a time, so a second coincidental duplicate is
-        correctly treated as still missing.
+        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.
 
         Returns one slot per input proto, in order; ``None`` where nothing on
-        the server matched it.
+        the server matched.
         """
         step_ids = sorted({proto.test_step_id for proto in protos})
         names = sorted({proto.name for proto in protos})
@@ -996,23 +974,16 @@ def value_of(numeric: float | None, string: str | None, boolean: bool | None):
     async def _create_measurement_batch_reconciling(
         self, protos: list[TestMeasurementProto]
     ) -> tuple[list[TestMeasurement], int]:
-        """Create a batch, reconciling against the server if the call's outcome is unknown.
+        """Create a batch, reconciling against the server if the outcome is unknown.
 
-        ``CreateTestMeasurements`` is not idempotent and the API has no request
-        ID to make a retry safe. A failure whose status is in
-        ``_AMBIGUOUS_STATUS_CODES`` could mean the request never reached the
-        server, or that it did and only the response was lost; blindly
-        resending would duplicate the whole batch in the second case. This
-        reconciles by natural key (see ``_reconcile_measurement_batch``) first
-        and only recreates what is genuinely missing. If the reconciliation
-        query itself fails, that exception propagates too.
-
-        A definite rejection (see ``_is_ambiguous_grpc_failure``) propagates
-        unchanged, including a local bug such as the ID-count mismatch
-        ``_create_measurement_batch`` raises itself.
+        ``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 (0 on the common, no-failure path).
+        the server instead of recreated.
         """
         try:
             return await self._create_measurement_batch(protos), 0
@@ -1040,12 +1011,10 @@ async def _create_measurement_batch_reconciling(
     async def _create_measurements_in_batches(
         self, protos: list[TestMeasurementProto]
     ) -> tuple[list[TestMeasurement], int]:
-        """Create measurements over as few batch calls as the size guard allows.
+        """Create measurements in chunks of ``_MEASUREMENT_CREATE_BATCH_SIZE``.
 
-        For a caller that holds more measurements than one request should carry,
-        such as a logged batch line of arbitrary length. Returns the created
-        measurements plus how many were reconciled rather than recreated; see
-        ``_create_measurement_batch_reconciling``.
+        Returns the created measurements plus how many were reconciled rather
+        than recreated; see ``_create_measurement_batch_reconciling``.
         """
         created: list[TestMeasurement] = []
         reconciled_count = 0
@@ -1428,19 +1397,11 @@ async def _replay_create_measurement_group(
 
         ``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, so it is left
-        alone rather than created a second time. A group can be part done like
-        that because a batch upload creates measurements in collapsed order, one
-        at a time, so it can stop partway through what one log line covers.
-
-        Every group goes through the batch endpoint, including a group of one.
-        Choosing an endpoint by group size would mean entities built two ways,
-        since the batch endpoint returns IDs while the single endpoint returns the
-        server's entity, and a resume would send the same measurement through a
-        different endpoint than the attempt before it.
-
-        Returns the real IDs created, in the order they were sent, for the audit
-        trail, plus how many were reconciled rather than recreated (see
+        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 = [
@@ -1670,10 +1631,8 @@ def record_created(simulated_id: str, real_id: str, *, save: bool = True) -> Non
             real_steps.append(real_step)
             record_created(sim_step_id, real_step._id_or_error)
 
-        # Measurements are the bulk of a log, so they go out in batches rather
-        # than one call each. One request per chunk means the sidecar is written
-        # once per chunk, so a crash loses at most one request's IDs and the
-        # resume recreates those measurements.
+        # 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] = []
         reconciled_count = 0
         for chunk_ids in chunked(state.measurements_order, _MEASUREMENT_CREATE_BATCH_SIZE):
@@ -1748,9 +1707,8 @@ async def _incremental_import_log_file(
         state = _ReplayState()
 
         raw_lines = await _read_log_lines(log_path)
-        # A run of consecutive single-measurement lines collects here and reaches
-        # the server as one batch call. Measurements are the bulk of a log, and
-        # one round-trip each is what makes a large log slow to upload.
+        # 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
 
@@ -1759,9 +1717,6 @@ async def flush_measurements() -> None:
 
             The cursor advances only after the call returns, so a failure leaves
             every line of the run to be retried next tick.
-            ``_replay_create_measurement_group`` already reconciles an ambiguous
-            failure against the server; this still raises only when that
-            reconciliation itself can't be answered.
             """
             if not buffered_measurements:
                 return
@@ -1808,10 +1763,9 @@ async def flush_measurements() -> None:
             raw_lines, start_line=tracking.last_uploaded_line
         ):
             if request_type == "CreateTestMeasurement":
-                # Parsed on the way into the buffer, so the flush has one job. 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.
+                # 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(
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 26d43b70e4..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,11 +40,9 @@
 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; see
-# ``_create_measurement_batch_reconciling``. Lives here rather than in
-# ``replay_worker`` or ``context_manager`` because those two already import
-# each other.
+# 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
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 e349d8f273..9fb9889f04 100644
--- a/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py
+++ b/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py
@@ -78,10 +78,10 @@ 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.
 
-    A tick that reconciled a measurement batch against the server (see
-    ``_create_measurement_batch_reconciling``) still exits cleanly, so the
-    count is summed across ticks and reported on stderr once, at session
-    end - the only channel back to the parent process.
+    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
diff --git a/python/lib/sift_client/errors.py b/python/lib/sift_client/errors.py
index 2cf53d5a71..773657274b 100644
--- a/python/lib/sift_client/errors.py
+++ b/python/lib/sift_client/errors.py
@@ -26,8 +26,9 @@ class SiftReportLifecycleWarning(SiftWarning):
 class SiftReplayReconciledWarning(SiftWarning):
     """A test-results upload recovered from an ambiguous retry.
 
-    Raised once per run, after a batch measurement create's response was
-    lost and the retry found some or all of the batch already on the server.
+    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.
     """
 
 
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 29e152958a..d196ac32f7 100644
--- a/python/lib/sift_client/util/test_results/context_manager.py
+++ b/python/lib/sift_client/util/test_results/context_manager.py
@@ -104,12 +104,7 @@ 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.
 
-    The worker's own exit code can't carry this: a reconciled run is a clean
-    exit, with nothing in the usual non-zero-exit stderr handling to surface
-    it. ``RECONCILED_MARKER`` is the only channel back.
-
-    Returns 0 for a line that is missing or malformed, rather than raising
-    over a worker detail that was never essential to parse.
+    Returns 0 if the ``RECONCILED_MARKER`` line is missing or malformed.
     """
     for line in stderr_text.splitlines():
         prefix = f"{RECONCILED_MARKER}:"

From 76f0f0e3b688f5564bdc59262d690c6fc67636a1 Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Thu, 15 Oct 2026 14:40:44 -0700
Subject: [PATCH 5/6] improve log replay termination

---
 python/changelog.d/810.feat.md                |   2 +-
 .../low_level_wrappers/_test_results_log.py   |  25 +++++
 .../low_level_wrappers/test_results.py        |  82 ++++++++------
 .../_internal/pytest_plugin/replay_worker.py  |  23 +++-
 .../test_incremental_replay.py                | 106 +++++++++++++++++-
 .../sift_client/_tests/scripts/__init__.py    |   0
 .../scripts/test_import_test_result_log.py    |  49 ++++++++
 .../_tests/util/test_report_context.py        |  97 ++++++++++++++++
 python/lib/sift_client/pytest_plugin.py       |  10 ++
 .../util/test_results/context_manager.py      |  38 +++++--
 10 files changed, 382 insertions(+), 50 deletions(-)
 create mode 100644 python/lib/sift_client/_tests/scripts/__init__.py
 create mode 100644 python/lib/sift_client/_tests/scripts/test_import_test_result_log.py

diff --git a/python/changelog.d/810.feat.md b/python/changelog.d/810.feat.md
index 150eacaca6..186ec3813c 100644
--- a/python/changelog.d/810.feat.md
+++ b/python/changelog.d/810.feat.md
@@ -1 +1 @@
-Test-results log uploads are faster: consecutive measurements are now sent in one `CreateTestMeasurements` call, and a retry after an unknown outcome checks the server first and reports it with the new `SiftReplayReconciledWarning`.
+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 d795dd79f4..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
@@ -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.
@@ -216,6 +226,21 @@ class ReplayResult:
     """
 
 
+# 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(
     log_file: str | Path,
     request_type: str,
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 b11d94f770..651921784f 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,7 @@
 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
@@ -50,6 +51,7 @@
     _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
@@ -98,13 +100,21 @@
 )
 
 
+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``.
 
-    An exception without a ``code()``, such as a local ``TypeError``, is not ambiguous.
+    Any other exception, such as a local ``TypeError``, is not ambiguous.
     """
-    code = getattr(exc, "code", None)
-    return callable(code) and code() in _AMBIGUOUS_STATUS_CODES
+    return isinstance(exc, RpcError) and _rpc_code(exc) in _AMBIGUOUS_STATUS_CODES
 
 
 class _EntryIds(NamedTuple):
@@ -922,6 +932,8 @@ async def _reconcile_measurement_batch(
         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.
@@ -932,42 +944,29 @@ async def _reconcile_measurement_batch(
             query_filter=and_(in_("test_step_id", step_ids), in_("name", names))
         )
 
-        def value_of(numeric: float | None, string: str | None, boolean: bool | None):
-            if numeric is not None:
-                return numeric
-            if string is not None:
-                return string
-            return boolean
-
-        candidates: dict[tuple[str, str, Any, Any], list[TestMeasurement]] = {}
+        candidates: defaultdict[tuple[str, str, Any, Any], list[TestMeasurement]] = defaultdict(
+            list
+        )
         for measurement in existing:
-            key = (
-                measurement.test_step_id,
-                measurement.name,
-                measurement.timestamp,
-                value_of(
-                    measurement.numeric_value, measurement.string_value, measurement.boolean_value
-                ),
-            )
-            candidates.setdefault(key, []).append(measurement)
+            value = 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:
-            if proto.HasField("numeric_value"):
-                value = value_of(proto.numeric_value, None, None)
-            elif proto.HasField("string_value"):
-                value = value_of(None, proto.string_value, None)
-            elif proto.HasField("boolean_value"):
-                value = value_of(None, None, proto.boolean_value)
-            else:
-                value = None
+            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)
+            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
 
@@ -1264,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 "
@@ -1707,6 +1703,12 @@ 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]] = []
@@ -1718,9 +1720,9 @@ async def flush_measurements() -> None:
             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
-            nonlocal reconciled_count
             first_line = tracking.last_uploaded_line + 1
             try:
                 real_ids, batch_reconciled = await self._replay_create_measurement_group(
@@ -1762,6 +1764,8 @@ async def flush_measurements() -> None:
         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
@@ -1822,13 +1826,17 @@ async def flush_measurements() -> None:
                 idmap=len(id_map),
             )
 
-        # A log ending in measurement lines leaves a run buffered.
-        await flush_measurements()
+        # 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:
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 9fb9889f04..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,13 +18,18 @@
 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.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
 
@@ -78,6 +83,10 @@ 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
@@ -90,13 +99,16 @@ def _incremental_import_loop(
         result = client.test_results.import_log_file(log_file, incremental=True)
         if result is not None:
             total_reconciled += result.reconciled_count
-        if received_signal:
+        if received_signal or replay_stop_requested(log_file):
             break
-    log_event(logger, logging.INFO, "replay.complete", log=log_file)
+    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:
+    if not keep_log and not stopped:
         cleanup_temp_log(log_file)
     return result
 
@@ -139,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 14745b024a..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
@@ -20,7 +20,11 @@
 from grpc import RpcError, StatusCode
 
 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
+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,
@@ -1358,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 c3e8304bd7..c2d8b1fd21 100644
--- a/python/lib/sift_client/_tests/util/test_report_context.py
+++ b/python/lib/sift_client/_tests/util/test_report_context.py
@@ -143,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/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/util/test_results/context_manager.py b/python/lib/sift_client/util/test_results/context_manager.py
index d196ac32f7..6de0885e01 100644
--- a/python/lib/sift_client/util/test_results/context_manager.py
+++ b/python/lib/sift_client/util/test_results/context_manager.py
@@ -16,6 +16,7 @@
 
 import numpy as np
 
+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,
@@ -106,8 +107,8 @@ def _parse_reconciled_count(stderr_text: str) -> int:
 
     Returns 0 if the ``RECONCILED_MARKER`` line is missing or malformed.
     """
+    prefix = f"{RECONCILED_MARKER}:"
     for line in stderr_text.splitlines():
-        prefix = f"{RECONCILED_MARKER}:"
         if line.startswith(prefix):
             try:
                 return int(line[len(prefix) :])
@@ -250,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,
@@ -411,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:
@@ -473,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
@@ -484,8 +491,18 @@ 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()
+                if sys.platform == "win32":
+                    # terminate() is a hard kill on Windows, so ask through a
+                    # file the importer checks between requests.
+                    LogTracking.stop_path(self.log_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()
+                LogTracking.stop_path(self.log_file).unlink(missing_ok=True)
                 self.replay_incomplete = True
                 log_event(
                     logger,
@@ -496,7 +513,7 @@ 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,
@@ -538,6 +555,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.

From a37e0ee3ae45d5a20be682a72825c717bbdc8c6b Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Thu, 15 Oct 2026 14:48:11 -0700
Subject: [PATCH 6/6] lint

---
 .../_internal/low_level_wrappers/test_results.py       |  2 +-
 .../sift_client/util/test_results/context_manager.py   | 10 +++++++---
 2 files changed, 8 insertions(+), 4 deletions(-)

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 651921784f..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
@@ -948,7 +948,7 @@ async def _reconcile_measurement_batch(
             list
         )
         for measurement in existing:
-            value = measurement.numeric_value
+            value: float | str | bool | None = measurement.numeric_value
             if value is None:
                 value = measurement.string_value
             if value is None:
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 6de0885e01..9e95f03366 100644
--- a/python/lib/sift_client/util/test_results/context_manager.py
+++ b/python/lib/sift_client/util/test_results/context_manager.py
@@ -491,10 +491,13 @@ def finalize(self) -> bool:
             try:
                 _, stderr_bytes = self._import_proc.communicate(timeout=self._import_proc_timeout)
             except subprocess.TimeoutExpired:
-                if sys.platform == "win32":
+                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.
-                    LogTracking.stop_path(self.log_file).touch()
+                    stop_file.touch()
                 else:
                     self._import_proc.terminate()
                 try:
@@ -502,7 +505,8 @@ def finalize(self) -> bool:
                 except subprocess.TimeoutExpired:
                     self._import_proc.kill()
                     self._import_proc.wait()
-                LogTracking.stop_path(self.log_file).unlink(missing_ok=True)
+                if stop_file is not None:
+                    stop_file.unlink(missing_ok=True)
                 self.replay_incomplete = True
                 log_event(
                     logger,