From e11121f8394eb93a98264d5bad59e276a3511109 Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Wed, 30 Sep 2026 12:13:52 -0700
Subject: [PATCH 1/3] handle unknown pydantic keys

---
 python/CHANGELOG.md                           |  5 ++
 .../low_level_wrappers/test_results.py        | 73 ++++++++++++++++++-
 .../test_incremental_replay.py                | 62 ++++++++++++++++
 .../_tests/sift_types/test_base.py            | 43 +++++++++++
 python/lib/sift_client/errors.py              | 24 ++++++
 python/lib/sift_client/sift_types/_base.py    | 30 ++++++++
 6 files changed, 233 insertions(+), 4 deletions(-)

diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md
index 8decb8a361..78b39a4947 100644
--- a/python/CHANGELOG.md
+++ b/python/CHANGELOG.md
@@ -3,6 +3,11 @@ All notable changes to this project will be documented in this file.
 
 This project adheres to [Semantic Versioning](http://semver.org/).
 
+## Unreleased
+
+- Fix a test-results log that could never finish uploading. An update whose fields were all `None`, or whose keys were not fields of the update model, produced an empty field mask. The client logged it without complaint, then the API rejected it at import. The replay cursor only advances past a line that succeeded, so every retry stopped at the same line and stranded the rest of the log. Such an update is now a no-op that is neither logged nor sent, and a logged one is skipped at import. Re-run `import-test-result-log` to finish a log written by an earlier version.
+- A key that is not a field of a create or update model now warns, instead of being dropped in silence. The warning suggests a near match when the key resembles a real field. Add `error::sift_client.errors.SiftWarning` to `filterwarnings` to make it fail the call.
+
 ## [v0.22.1] - September 29, 2026
 
 - Fix streaming ingestion processes crashing with SIGSEGV or SIGABRT at interpreter exit. The `sift-stream-bindings` tokio runtime was never shut down, so a runtime thread that finished work while Python was finalizing re-entered the interpreter and killed the process after the program had completed. The bindings now stop the runtime from an `atexit` hook and expose `shutdown(timeout=5.0)` for callers that exit with `os._exit()`. If runtime threads outlive the timeout, `shutdown()` issues a `RuntimeWarning`. Requires `sift-stream-bindings` 0.5.2. ([#804](https://github.com/sift-stack/sift/pull/804))
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 e0d33dec8c..8571754c27 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
@@ -2,8 +2,9 @@
 
 import logging
 import uuid
+import warnings
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Callable, NamedTuple, TypeVar, cast
 
 from google.protobuf import json_format
 from grpc import RpcError, StatusCode
@@ -51,6 +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.errors import SiftWarning, caller_stacklevel
 from sift_client.sift_types.test_report import (
     TestMeasurement,
     TestMeasurementCreate,
@@ -122,6 +124,33 @@ def _mark_simulated(instance: _EntityT) -> _EntityT:
         instance.__dict__["_simulated"] = True
         return instance
 
+    @classmethod
+    def _no_op_update(
+        cls,
+        entity_name: str,
+        existing: _EntityT | None,
+        simulated: Callable[[], _EntityT],
+    ) -> _EntityT:
+        """Short-circuit an update whose field mask is empty.
+
+        An empty mask asks the API to change nothing, and the API rejects it.
+        Logging one is worse than sending it: at import the entry stops the
+        replay at its line, and no retry gets past it.
+
+        Warn, since either every field was ``None`` or the keys were dropped as
+        unknown (see ``ModelCreateUpdateBase._warn_on_unknown_keys``). Return
+        ``existing`` when the caller has it, otherwise a response built from the
+        request and stamped simulated, since it never reached Sift.
+        """
+        warnings.warn(
+            f"Update to {entity_name} requested no field changes; ignored.",
+            SiftWarning,
+            stacklevel=caller_stacklevel(),
+        )
+        if existing is not None:
+            return existing
+        return cls._mark_simulated(simulated())
+
     @staticmethod
     def simulate_create_test_report_response(
         request: CreateTestReportRequest,
@@ -534,7 +563,8 @@ async def update_test_report(
             simulate: If True, return a simulated response without making an API call.
 
         Returns:
-            The updated TestReport.
+            The updated TestReport, or the report unchanged when the update names
+            no fields (see ``_no_op_update``).
         """
         if request is None:
             if update is None:
@@ -542,6 +572,13 @@ async def update_test_report(
             test_report_proto, field_mask = update.to_proto_with_mask()
             request = UpdateTestReportRequest(test_report=test_report_proto, update_mask=field_mask)
 
+        if not request.update_mask.paths:
+            return self._no_op_update(
+                "TestReport",
+                existing,
+                lambda: self.simulate_update_test_report_response(request),
+            )
+
         if log_file is not None or simulate:
             if log_file is not None:
                 await log_request_to_file(log_file, "UpdateTestReport", request)
@@ -689,7 +726,8 @@ async def update_test_step(
             simulate: If True, return a simulated response without making an API call.
 
         Returns:
-            The updated TestStep.
+            The updated TestStep, or the step unchanged when the update names no
+            fields (see ``_no_op_update``).
         """
         if request is None:
             if update is None:
@@ -700,6 +738,13 @@ async def update_test_step(
                 field_mask.paths.append("error_info")
             request = UpdateTestStepRequest(test_step=test_step_proto, update_mask=field_mask)
 
+        if not request.update_mask.paths:
+            return self._no_op_update(
+                "TestStep",
+                existing,
+                lambda: self.simulate_update_test_step_response(request),
+            )
+
         if log_file is not None or simulate:
             if log_file is not None:
                 await log_request_to_file(log_file, "UpdateTestStep", request)
@@ -892,7 +937,8 @@ async def update_test_measurement(
             simulate: If True, return a simulated response without making an API call.
 
         Returns:
-            The updated TestMeasurement.
+            The updated TestMeasurement, or the measurement unchanged when the
+            update names no fields (see ``_no_op_update``).
         """
         if request is None:
             if update is None:
@@ -902,6 +948,13 @@ async def update_test_measurement(
                 test_measurement=test_measurement_proto, update_mask=field_mask
             )
 
+        if not request.update_mask.paths:
+            return self._no_op_update(
+                "TestMeasurement",
+                existing,
+                lambda: self.simulate_update_test_measurement_response(request),
+            )
+
         if log_file is not None or simulate:
             if log_file is not None:
                 await log_request_to_file(log_file, "UpdateTestMeasurement", request)
@@ -1226,6 +1279,12 @@ async def _replay_update_report(
         orig_report_id = request.test_report.test_report_id
         mapped_report_id = self._map_id(id_map, orig_report_id)
         request.test_report.test_report_id = mapped_report_id
+        # An empty mask asks the server to change nothing, and the API rejects it.
+        # Clients before the write-time guard logged such entries, and the cursor
+        # only advances past a line that succeeded, so one of them stopped every
+        # retry at the same place. Nothing to apply here, so count it skipped.
+        if not request.update_mask.paths:
+            return _EntryIds(orig_report_id or None, mapped_report_id or None, skipped=True)
         # Batch/simulate replays the whole log in order, so a missing report means
         # the log is malformed. Incremental replay may have created the report on an
         # earlier tick (its real ID lives in id_map), so state.report is legitimately
@@ -1251,6 +1310,9 @@ async def _replay_update_step(
         orig_step_id = request.test_step.test_step_id
         mapped_step_id = self._map_id(id_map, orig_step_id)
         request.test_step.test_step_id = mapped_step_id
+        # No paths means nothing to apply; see _replay_update_report.
+        if not request.update_mask.paths:
+            return _EntryIds(orig_step_id or None, mapped_step_id or None, skipped=True)
         existing_step = state.steps_by_id.get(mapped_step_id)
         if simulate and existing_step is None:
             raise ValueError(f"UpdateTestStep for unknown step: {orig_step_id}")
@@ -1275,6 +1337,9 @@ async def _replay_update_measurement(
         orig_meas_id = request.test_measurement.measurement_id
         mapped_meas_id = self._map_id(id_map, orig_meas_id)
         request.test_measurement.measurement_id = mapped_meas_id
+        # No paths means nothing to apply; see _replay_update_report.
+        if not request.update_mask.paths:
+            return _EntryIds(orig_meas_id or None, mapped_meas_id or None, skipped=True)
         existing_meas = state.measurements_by_id.get(mapped_meas_id)
         if simulate and existing_meas is None:
             raise ValueError(f"UpdateTestMeasurement for unknown measurement: {orig_meas_id}")
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 63bbc85333..6a6d1f7281 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
@@ -24,6 +24,7 @@
     # Aliased so pytest doesn't try to collect the `Test`-prefixed client as a suite.
     TestResultsLowLevelClient as ResultsLowLevelClient,
 )
+from sift_client.errors import SiftWarning
 from sift_client.sift_types.test_report import (
     TestMeasurement,
     TestMeasurementCreate,
@@ -723,6 +724,67 @@ async def test_resume_propagates_errors_other_than_a_missing_report(tmp_path):
         await client.import_log_file(log_file)
 
 
+@pytest.mark.asyncio
+async def test_resume_skips_an_update_with_an_empty_mask(tmp_path):
+    """A logged update carrying no field paths is skipped, not sent.
+
+    Clients before the write-time guard logged one whenever every field of an
+    update was None, or its keys were dropped as unknown. The API rejects an
+    empty mask, and the cursor only advances past a line that succeeded, so such
+    a line stranded the rest of the log on every retry.
+    """
+    log_file = tmp_path / "empty_mask.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+
+    report = await client.create_test_report(test_report=_report_create(), log_file=log_file)
+    # Written by hand: the guard in update_test_report now refuses to log this.
+    with log_file.open("a") as handle:
+        handle.write(
+            f'[UpdateTestReport] {{"testReport":{{"testReportId":"{report.id_}"}},'
+            '"updateMask":""}\n'
+        )
+    update = TestReportUpdate(status=TestStatus.FAILED)
+    update.resource_id = report.id_
+    await client.update_test_report(update=update, log_file=log_file)
+
+    LogTracking(last_uploaded_line=1, id_map={report.id_: "real-report"}).save(log_file)
+
+    client.update_test_report = AsyncMock(return_value=_make_report("real-report"))
+
+    await client.import_log_file(log_file, incremental=True)
+
+    # Only the real update was sent; the empty-mask line never reached the API.
+    client.update_test_report.assert_awaited_once()
+    sent = client.update_test_report.await_args.kwargs["request"]
+    assert sent.test_report.status == TestStatus.FAILED.value
+    # Both remaining lines are behind the cursor, so a later tick re-sends neither.
+    assert LogTracking.load(log_file).last_uploaded_line == 3
+
+
+@pytest.mark.asyncio
+async def test_update_with_no_fields_is_not_logged(tmp_path):
+    """An update that changes nothing writes no log entry and returns the entity.
+
+    The API rejects a no-op update, so logging one plants an entry that can never
+    replay. The caller gets the entity back unchanged, plus a warning.
+    """
+    log_file = tmp_path / "no_fields.jsonl"
+    client = ResultsLowLevelClient(grpc_client=MagicMock())
+
+    report = await client.create_test_report(test_report=_report_create(), log_file=log_file)
+    lines_before = log_file.read_text().count("\n")
+
+    update = TestReportUpdate(run_id=None)
+    update.resource_id = report.id_
+    with pytest.warns(SiftWarning, match="requested no field changes"):
+        returned = await client.update_test_report(
+            update=update, log_file=log_file, existing=report
+        )
+
+    assert returned is report
+    assert log_file.read_text().count("\n") == lines_before
+
+
 @pytest.mark.asyncio
 async def test_incremental_and_new_report_are_rejected_together(tmp_path):
     """The two flags contradict each other, so asking for both is an error.
diff --git a/python/lib/sift_client/_tests/sift_types/test_base.py b/python/lib/sift_client/_tests/sift_types/test_base.py
index a0c3cfc584..2264c79761 100644
--- a/python/lib/sift_client/_tests/sift_types/test_base.py
+++ b/python/lib/sift_client/_tests/sift_types/test_base.py
@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
+import warnings
 from datetime import datetime, timezone
 from typing import ClassVar
 from unittest.mock import MagicMock
 
 import pytest
+from pydantic import ConfigDict, ValidationError
 from sift.calculated_channels.v2.calculated_channels_pb2 import (
     CalculatedChannel as CalculatedChannelProto,
 )
@@ -14,6 +16,7 @@
     CreateCalculatedChannelRequest,
 )
 
+from sift_client.errors import SiftWarning
 from sift_client.sift_types._base import (
     BaseType,
     MappingHelper,
@@ -226,6 +229,46 @@ def test_update_requires_resource_id(self):
             model.to_proto_with_mask()
 
 
+class TestUnknownKeys:
+    """Tests for the warning on input keys that are not fields of the model."""
+
+    def test_unknown_key_warns_and_is_ignored(self):
+        """An unrecognized key contributes nothing to the field mask."""
+        with pytest.warns(SiftWarning, match="Unknown field `tags` for SimpleUpdateModel"):
+            model = SimpleUpdateModel.model_validate({"tags": ["a"], "name": "new_name"})
+
+        model.resource_id = "test_id"
+        _, mask = model.to_proto_with_mask()
+        assert mask.paths == ["name"]
+
+    def test_unknown_key_suggests_a_near_match(self):
+        """A misspelled or camelCase key is usually a typo for a real field."""
+        with pytest.warns(SiftWarning, match=r"did you mean `description`\?"):
+            SimpleUpdateModel.model_validate({"descriptionn": "x"})
+
+    def test_unknown_key_on_a_create_model_warns(self):
+        """A create drops the key too, which loses the value instead of no-opping."""
+        with pytest.warns(SiftWarning, match="Unknown field `unit` for SimpleCreateModel"):
+            SimpleCreateModel.model_validate({"name": "n", "unit": "volts"})
+
+    def test_known_keys_do_not_warn(self):
+        """The common case stays quiet, including a field explicitly set to None."""
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", SiftWarning)
+            SimpleUpdateModel.model_validate({"name": "new_name", "description": None})
+
+    def test_extra_forbid_model_is_left_to_raise(self):
+        """A model that opts into extra="forbid" reports the key itself."""
+
+        class StrictUpdateModel(SimpleUpdateModel):
+            model_config = ConfigDict(extra="forbid")
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", SiftWarning)
+            with pytest.raises(ValidationError):
+                StrictUpdateModel.model_validate({"tags": ["a"]})
+
+
 class TestMappingHelper:
     """Tests for MappingHelper functionality."""
 
diff --git a/python/lib/sift_client/errors.py b/python/lib/sift_client/errors.py
index 657c6a9488..c87f1350a8 100644
--- a/python/lib/sift_client/errors.py
+++ b/python/lib/sift_client/errors.py
@@ -1,7 +1,31 @@
 from __future__ import annotations
 
+import inspect
+from pathlib import Path
 from typing import NoReturn
 
+_PACKAGE_ROOT = str(Path(__file__).resolve().parent)
+
+
+def caller_stacklevel() -> int:
+    """Return the ``warnings.warn`` stacklevel of the first frame outside the SDK.
+
+    A fixed stacklevel points at SDK or pydantic internals, which tells the
+    caller nothing about which of their lines caused the warning.
+
+    Call this from the function that calls ``warnings.warn``.
+    """
+    frame = inspect.currentframe()
+    frame = frame.f_back if frame is not None else None  # the function calling warn()
+    level = 1
+    while frame is not None:
+        filename = frame.f_code.co_filename
+        if not filename.startswith(_PACKAGE_ROOT) and "pydantic" not in filename:
+            return level
+        frame = frame.f_back
+        level += 1
+    return 2
+
 
 class SiftWarning(UserWarning):
     """Base warning for Sift generated warnings."""
diff --git a/python/lib/sift_client/sift_types/_base.py b/python/lib/sift_client/sift_types/_base.py
index 7af93087e6..a06e99039e 100644
--- a/python/lib/sift_client/sift_types/_base.py
+++ b/python/lib/sift_client/sift_types/_base.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import difflib
+import warnings
 from abc import ABC, abstractmethod
 from datetime import datetime
 from enum import Enum
@@ -16,6 +18,8 @@
 from google.protobuf import field_mask_pb2, message
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
 
+from sift_client.errors import SiftWarning, caller_stacklevel
+
 if TYPE_CHECKING:
     from sift_client.client import SiftClient
 
@@ -113,6 +117,32 @@ class ModelCreateUpdateBase(BaseModel, ABC):
     def __init__(self, **data: Any):
         super().__init__(**data)
 
+    @model_validator(mode="before")
+    @classmethod
+    def _warn_on_unknown_keys(cls, data: Any) -> Any:
+        """Warn for input keys that are not fields of this model.
+
+        Pydantic's default ``extra="ignore"`` drops an unrecognized key in
+        silence. A create then omits that field, and an update carries an empty
+        field mask, which the API rejects.
+
+        Warn rather than raise, since a reporting mistake must not fail a test
+        run in progress. ``filterwarnings`` promotes it to an error. Models that
+        set ``extra="forbid"`` raise on their own, so they are left alone.
+        """
+        if not isinstance(data, dict) or cls.model_config.get("extra") == "forbid":
+            return data
+        known = sorted(cls.model_fields)
+        for key in [key for key in data if key not in cls.model_fields]:
+            close = difflib.get_close_matches(str(key), known, n=1, cutoff=0.6)
+            hint = f" (did you mean `{close[0]}`?)" if close else ""
+            warnings.warn(
+                f"Unknown field `{key}` for {cls.__name__}{hint}; ignored.",
+                SiftWarning,
+                stacklevel=caller_stacklevel(),
+            )
+        return data
+
     @model_validator(mode="after")
     def _check_mapping_helpers(self):
         if self._to_proto_helpers:

From 3f8cb1f941af0b6ce454f9935b1ffe7177cec222 Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Wed, 30 Sep 2026 15:38:52 -0700
Subject: [PATCH 2/3] python(fix): address review on empty update masks

Move caller_stacklevel out of the public errors module. mkdocs publishes
non-underscore names from there, and a stack-walking utility is not an error
type; it now sits beside the other generic helpers in _internal/util.

Add SiftIgnoredInputWarning so filterwarnings can target an unknown model
field and a no-op update without promoting every other Sift warning.

Stop fabricating an entity when a live no-op update has nothing to return.
_skip_empty_update now hands back a simulated response only on the log and
simulate paths, and raises otherwise. The fabricated entity carried
_simulated=True, which the pytest plugin reads to resolve the report link, so
a caller passing an ID string could not tell it from a real read.

Return from the unknown-key validator before sorting the field names, so the
path where every key is known does no extra work.
---
 python/CHANGELOG.md                           |  4 +-
 .../low_level_wrappers/test_results.py        | 53 ++++++++++++-------
 python/lib/sift_client/_internal/util/util.py | 26 +++++++++
 .../test_incremental_replay.py                |  4 +-
 .../_tests/sift_types/test_base.py            | 29 +++++++---
 python/lib/sift_client/errors.py              | 36 +++++--------
 python/lib/sift_client/sift_types/_base.py    | 16 ++++--
 7 files changed, 111 insertions(+), 57 deletions(-)

diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md
index 78b39a4947..9af11c02c6 100644
--- a/python/CHANGELOG.md
+++ b/python/CHANGELOG.md
@@ -3,10 +3,10 @@ All notable changes to this project will be documented in this file.
 
 This project adheres to [Semantic Versioning](http://semver.org/).
 
-## Unreleased
+## [Unreleased]
 
 - Fix a test-results log that could never finish uploading. An update whose fields were all `None`, or whose keys were not fields of the update model, produced an empty field mask. The client logged it without complaint, then the API rejected it at import. The replay cursor only advances past a line that succeeded, so every retry stopped at the same line and stranded the rest of the log. Such an update is now a no-op that is neither logged nor sent, and a logged one is skipped at import. Re-run `import-test-result-log` to finish a log written by an earlier version.
-- A key that is not a field of a create or update model now warns, instead of being dropped in silence. The warning suggests a near match when the key resembles a real field. Add `error::sift_client.errors.SiftWarning` to `filterwarnings` to make it fail the call.
+- A key that is not a field of a create or update model now warns, instead of being dropped in silence. The warning suggests a near match when the key resembles a real field. Both this and an update that names no fields raise the new `SiftIgnoredInputWarning`. Add `error::sift_client.errors.SiftIgnoredInputWarning` to `filterwarnings` to make either one fail the call.
 
 ## [v0.22.1] - September 29, 2026
 
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 8571754c27..23810a152b 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
@@ -4,7 +4,7 @@
 import uuid
 import warnings
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable, NamedTuple, TypeVar, cast
+from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar, cast
 
 from google.protobuf import json_format
 from grpc import RpcError, StatusCode
@@ -52,7 +52,8 @@
 )
 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.errors import SiftWarning, caller_stacklevel
+from sift_client._internal.util.util import caller_stacklevel
+from sift_client.errors import SiftIgnoredInputWarning
 from sift_client.sift_types.test_report import (
     TestMeasurement,
     TestMeasurementCreate,
@@ -125,11 +126,11 @@ def _mark_simulated(instance: _EntityT) -> _EntityT:
         return instance
 
     @classmethod
-    def _no_op_update(
+    def _skip_empty_update(
         cls,
         entity_name: str,
         existing: _EntityT | None,
-        simulated: Callable[[], _EntityT],
+        simulated: _EntityT | None,
     ) -> _EntityT:
         """Short-circuit an update whose field mask is empty.
 
@@ -138,18 +139,31 @@ def _no_op_update(
         replay at its line, and no retry gets past it.
 
         Warn, since either every field was ``None`` or the keys were dropped as
-        unknown (see ``ModelCreateUpdateBase._warn_on_unknown_keys``). Return
-        ``existing`` when the caller has it, otherwise a response built from the
-        request and stamped simulated, since it never reached Sift.
+        unknown (see ``ModelCreateUpdateBase._warn_on_unknown_keys``), then return
+        the entity unchanged. ``simulated`` is the response to hand back on the
+        log and simulate paths, where a synthesized entity is the correct answer,
+        and None on a live call, where it would not be.
+
+        Raises:
+            ValueError: On a live call with no ``existing`` entity. Nothing
+                changed server-side and there is no entity to return, so the
+                alternative is a fabricated one, which callers cannot tell from a
+                real read.
         """
         warnings.warn(
             f"Update to {entity_name} requested no field changes; ignored.",
-            SiftWarning,
+            SiftIgnoredInputWarning,
             stacklevel=caller_stacklevel(),
         )
         if existing is not None:
             return existing
-        return cls._mark_simulated(simulated())
+        if simulated is not None:
+            return cls._mark_simulated(simulated)
+        raise ValueError(
+            f"Update to {entity_name} named no fields to change. Pass the "
+            f"{entity_name} rather than its ID to get it back unchanged, or name "
+            "at least one field to update."
+        )
 
     @staticmethod
     def simulate_create_test_report_response(
@@ -564,7 +578,7 @@ async def update_test_report(
 
         Returns:
             The updated TestReport, or the report unchanged when the update names
-            no fields (see ``_no_op_update``).
+            no fields (see ``_skip_empty_update``).
         """
         if request is None:
             if update is None:
@@ -572,11 +586,12 @@ async def update_test_report(
             test_report_proto, field_mask = update.to_proto_with_mask()
             request = UpdateTestReportRequest(test_report=test_report_proto, update_mask=field_mask)
 
+        simulating = log_file is not None or simulate
         if not request.update_mask.paths:
-            return self._no_op_update(
+            return self._skip_empty_update(
                 "TestReport",
                 existing,
-                lambda: self.simulate_update_test_report_response(request),
+                self.simulate_update_test_report_response(request) if simulating else None,
             )
 
         if log_file is not None or simulate:
@@ -727,7 +742,7 @@ async def update_test_step(
 
         Returns:
             The updated TestStep, or the step unchanged when the update names no
-            fields (see ``_no_op_update``).
+            fields (see ``_skip_empty_update``).
         """
         if request is None:
             if update is None:
@@ -738,11 +753,12 @@ async def update_test_step(
                 field_mask.paths.append("error_info")
             request = UpdateTestStepRequest(test_step=test_step_proto, update_mask=field_mask)
 
+        simulating = log_file is not None or simulate
         if not request.update_mask.paths:
-            return self._no_op_update(
+            return self._skip_empty_update(
                 "TestStep",
                 existing,
-                lambda: self.simulate_update_test_step_response(request),
+                self.simulate_update_test_step_response(request) if simulating else None,
             )
 
         if log_file is not None or simulate:
@@ -938,7 +954,7 @@ async def update_test_measurement(
 
         Returns:
             The updated TestMeasurement, or the measurement unchanged when the
-            update names no fields (see ``_no_op_update``).
+            update names no fields (see ``_skip_empty_update``).
         """
         if request is None:
             if update is None:
@@ -948,11 +964,12 @@ async def update_test_measurement(
                 test_measurement=test_measurement_proto, update_mask=field_mask
             )
 
+        simulating = log_file is not None or simulate
         if not request.update_mask.paths:
-            return self._no_op_update(
+            return self._skip_empty_update(
                 "TestMeasurement",
                 existing,
-                lambda: self.simulate_update_test_measurement_response(request),
+                self.simulate_update_test_measurement_response(request) if simulating else None,
             )
 
         if log_file is not None or simulate:
diff --git a/python/lib/sift_client/_internal/util/util.py b/python/lib/sift_client/_internal/util/util.py
index 28f69ef921..3d886556d5 100644
--- a/python/lib/sift_client/_internal/util/util.py
+++ b/python/lib/sift_client/_internal/util/util.py
@@ -1,16 +1,42 @@
 from __future__ import annotations
 
+import inspect
+import os
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
 
+_PACKAGE_ROOT = str(Path(__file__).resolve().parent.parent.parent)
+_PYDANTIC_PATH = f"{os.sep}pydantic{os.sep}"
+
 
 def count_non_none(*args: Any) -> int:
     """Count the number of non-none arguments."""
     return sum(1 for arg in args if arg is not None)
 
 
+def caller_stacklevel() -> int:
+    """Return the ``warnings.warn`` stacklevel of the first frame outside the SDK.
+
+    A fixed stacklevel points at SDK or pydantic internals, which tells the
+    caller nothing about which of their lines caused the warning.
+
+    Call this from the function that calls ``warnings.warn``.
+    """
+    frame = inspect.currentframe()
+    frame = frame.f_back if frame is not None else None  # the function calling warn()
+    level = 1
+    while frame is not None:
+        filename = frame.f_code.co_filename
+        if not filename.startswith(_PACKAGE_ROOT) and _PYDANTIC_PATH not in filename:
+            return level
+        frame = frame.f_back
+        level += 1
+    return 2
+
+
 def chunked(items: list[Any], size: int) -> Iterator[list[Any]]:
     """Yield successive chunks of at most ``size`` items."""
     for i in range(0, len(items), size):
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 6a6d1f7281..02156bf74c 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
@@ -24,7 +24,7 @@
     # Aliased so pytest doesn't try to collect the `Test`-prefixed client as a suite.
     TestResultsLowLevelClient as ResultsLowLevelClient,
 )
-from sift_client.errors import SiftWarning
+from sift_client.errors import SiftIgnoredInputWarning
 from sift_client.sift_types.test_report import (
     TestMeasurement,
     TestMeasurementCreate,
@@ -776,7 +776,7 @@ async def test_update_with_no_fields_is_not_logged(tmp_path):
 
     update = TestReportUpdate(run_id=None)
     update.resource_id = report.id_
-    with pytest.warns(SiftWarning, match="requested no field changes"):
+    with pytest.warns(SiftIgnoredInputWarning, match="requested no field changes"):
         returned = await client.update_test_report(
             update=update, log_file=log_file, existing=report
         )
diff --git a/python/lib/sift_client/_tests/sift_types/test_base.py b/python/lib/sift_client/_tests/sift_types/test_base.py
index 2264c79761..1562e70d60 100644
--- a/python/lib/sift_client/_tests/sift_types/test_base.py
+++ b/python/lib/sift_client/_tests/sift_types/test_base.py
@@ -16,13 +16,14 @@
     CreateCalculatedChannelRequest,
 )
 
-from sift_client.errors import SiftWarning
+from sift_client.errors import SiftIgnoredInputWarning
 from sift_client.sift_types._base import (
     BaseType,
     MappingHelper,
     ModelCreate,
     ModelUpdate,
 )
+from sift_client.sift_types.annotation import PhaseCreate
 
 
 class SimpleCreateModel(ModelCreate[CreateCalculatedChannelRequest]):
@@ -234,7 +235,9 @@ class TestUnknownKeys:
 
     def test_unknown_key_warns_and_is_ignored(self):
         """An unrecognized key contributes nothing to the field mask."""
-        with pytest.warns(SiftWarning, match="Unknown field `tags` for SimpleUpdateModel"):
+        with pytest.warns(
+            SiftIgnoredInputWarning, match="Unknown field `tags` for SimpleUpdateModel"
+        ):
             model = SimpleUpdateModel.model_validate({"tags": ["a"], "name": "new_name"})
 
         model.resource_id = "test_id"
@@ -243,18 +246,20 @@ def test_unknown_key_warns_and_is_ignored(self):
 
     def test_unknown_key_suggests_a_near_match(self):
         """A misspelled or camelCase key is usually a typo for a real field."""
-        with pytest.warns(SiftWarning, match=r"did you mean `description`\?"):
+        with pytest.warns(SiftIgnoredInputWarning, match=r"did you mean `description`\?"):
             SimpleUpdateModel.model_validate({"descriptionn": "x"})
 
     def test_unknown_key_on_a_create_model_warns(self):
         """A create drops the key too, which loses the value instead of no-opping."""
-        with pytest.warns(SiftWarning, match="Unknown field `unit` for SimpleCreateModel"):
+        with pytest.warns(
+            SiftIgnoredInputWarning, match="Unknown field `unit` for SimpleCreateModel"
+        ):
             SimpleCreateModel.model_validate({"name": "n", "unit": "volts"})
 
     def test_known_keys_do_not_warn(self):
         """The common case stays quiet, including a field explicitly set to None."""
         with warnings.catch_warnings():
-            warnings.simplefilter("error", SiftWarning)
+            warnings.simplefilter("error", SiftIgnoredInputWarning)
             SimpleUpdateModel.model_validate({"name": "new_name", "description": None})
 
     def test_extra_forbid_model_is_left_to_raise(self):
@@ -264,10 +269,22 @@ class StrictUpdateModel(SimpleUpdateModel):
             model_config = ConfigDict(extra="forbid")
 
         with warnings.catch_warnings():
-            warnings.simplefilter("error", SiftWarning)
+            warnings.simplefilter("error", SiftIgnoredInputWarning)
             with pytest.raises(ValidationError):
                 StrictUpdateModel.model_validate({"tags": ["a"]})
 
+    def test_extra_forbid_is_detected_through_inheritance(self):
+        """The real forbid model sets the config on a parent, not on itself.
+
+        ``PhaseCreate`` inherits from ``AnnotationCreateBase``, so the skip relies
+        on pydantic merging parent config into the subclass. Asserting it against
+        a locally defined model would not exercise that merge.
+        """
+        with warnings.catch_warnings():
+            warnings.simplefilter("error", SiftIgnoredInputWarning)
+            with pytest.raises(ValidationError):
+                PhaseCreate.model_validate({"name": "p", "state": "open"})
+
 
 class TestMappingHelper:
     """Tests for MappingHelper functionality."""
diff --git a/python/lib/sift_client/errors.py b/python/lib/sift_client/errors.py
index c87f1350a8..9e853f37a0 100644
--- a/python/lib/sift_client/errors.py
+++ b/python/lib/sift_client/errors.py
@@ -1,31 +1,7 @@
 from __future__ import annotations
 
-import inspect
-from pathlib import Path
 from typing import NoReturn
 
-_PACKAGE_ROOT = str(Path(__file__).resolve().parent)
-
-
-def caller_stacklevel() -> int:
-    """Return the ``warnings.warn`` stacklevel of the first frame outside the SDK.
-
-    A fixed stacklevel points at SDK or pydantic internals, which tells the
-    caller nothing about which of their lines caused the warning.
-
-    Call this from the function that calls ``warnings.warn``.
-    """
-    frame = inspect.currentframe()
-    frame = frame.f_back if frame is not None else None  # the function calling warn()
-    level = 1
-    while frame is not None:
-        filename = frame.f_code.co_filename
-        if not filename.startswith(_PACKAGE_ROOT) and "pydantic" not in filename:
-            return level
-        frame = frame.f_back
-        level += 1
-    return 2
-
 
 class SiftWarning(UserWarning):
     """Base warning for Sift generated warnings."""
@@ -35,6 +11,18 @@ class SiftExperimentalWarning(SiftWarning):
     """Warning for experimental features."""
 
 
+class SiftIgnoredInputWarning(SiftWarning):
+    """Input the SDK accepted but could not act on.
+
+    Raised for a key that is not a field of a create or update model, and for an
+    update that names no fields to change. Both are caller mistakes that the SDK
+    tolerates rather than fails, so this class exists to let a caller promote
+    exactly these to errors without also promoting unrelated Sift warnings::
+
+        filterwarnings = error::sift_client.errors.SiftIgnoredInputWarning
+    """
+
+
 class SiftCredentialsError(ValueError):
     """Raised when Sift credentials cannot be resolved.
 
diff --git a/python/lib/sift_client/sift_types/_base.py b/python/lib/sift_client/sift_types/_base.py
index a06e99039e..7acec28abd 100644
--- a/python/lib/sift_client/sift_types/_base.py
+++ b/python/lib/sift_client/sift_types/_base.py
@@ -18,7 +18,8 @@
 from google.protobuf import field_mask_pb2, message
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
 
-from sift_client.errors import SiftWarning, caller_stacklevel
+from sift_client._internal.util.util import caller_stacklevel
+from sift_client.errors import SiftIgnoredInputWarning
 
 if TYPE_CHECKING:
     from sift_client.client import SiftClient
@@ -127,18 +128,23 @@ def _warn_on_unknown_keys(cls, data: Any) -> Any:
         field mask, which the API rejects.
 
         Warn rather than raise, since a reporting mistake must not fail a test
-        run in progress. ``filterwarnings`` promotes it to an error. Models that
-        set ``extra="forbid"`` raise on their own, so they are left alone.
+        run in progress. ``filterwarnings`` promotes it to an error. A model that
+        sets ``extra="forbid"`` raises on its own and is left alone; that is what
+        ``PhaseCreate`` does, and forbidding here instead would abort a live test
+        run over a dropped field.
         """
         if not isinstance(data, dict) or cls.model_config.get("extra") == "forbid":
             return data
+        unknown = [key for key in data if key not in cls.model_fields]
+        if not unknown:
+            return data
         known = sorted(cls.model_fields)
-        for key in [key for key in data if key not in cls.model_fields]:
+        for key in unknown:
             close = difflib.get_close_matches(str(key), known, n=1, cutoff=0.6)
             hint = f" (did you mean `{close[0]}`?)" if close else ""
             warnings.warn(
                 f"Unknown field `{key}` for {cls.__name__}{hint}; ignored.",
-                SiftWarning,
+                SiftIgnoredInputWarning,
                 stacklevel=caller_stacklevel(),
             )
         return data

From 658e6a395a37967c1792e51e7b8cccffa59dc29e Mon Sep 17 00:00:00 2001
From: Alex Luck <luck@siftstack.com>
Date: Wed, 30 Sep 2026 15:40:53 -0700
Subject: [PATCH 3/3] 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.
---
 python/CHANGELOG.md                           |   1 +
 .../low_level_wrappers/_test_results_log.py   |  14 +-
 .../low_level_wrappers/test_results.py        | 248 ++++++++++++---
 .../test_incremental_replay.py                | 299 +++++++++++++++++-
 4 files changed, 500 insertions(+), 62 deletions(-)

diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md
index 9af11c02c6..6eb7b3f686 100644
--- a/python/CHANGELOG.md
+++ b/python/CHANGELOG.md
@@ -7,6 +7,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 - Fix a test-results log that could never finish uploading. An update whose fields were all `None`, or whose keys were not fields of the update model, produced an empty field mask. The client logged it without complaint, then the API rejected it at import. The replay cursor only advances past a line that succeeded, so every retry stopped at the same line and stranded the rest of the log. Such an update is now a no-op that is neither logged nor sent, and a logged one is skipped at import. Re-run `import-test-result-log` to finish a log written by an earlier version.
 - A key that is not a field of a create or update model now warns, instead of being dropped in silence. The warning suggests a near match when the key resembles a real field. Both this and an update that names no fields raise the new `SiftIgnoredInputWarning`. Add `error::sift_client.errors.SiftIgnoredInputWarning` to `filterwarnings` to make either one fail the call.
+- 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.
 
 ## [v0.22.1] - September 29, 2026
 
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 23810a152b..cd1aae0e65 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.
@@ -870,6 +878,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,
         *,
@@ -1230,6 +1283,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,
@@ -1244,8 +1339,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)
@@ -1256,26 +1351,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
@@ -1403,19 +1485,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
@@ -1436,18 +1522,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
@@ -1475,10 +1570,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
@@ -1502,9 +1597,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(
@@ -1547,6 +1714,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 02156bf74c..90047d4dc0 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