fix(OMN-13118): pin terminal-consumer read offset synchronously (close lazy seek_to_end publish race)

src/omnibase_infra/runtime/service_pattern_b_broker.py

@@ -260,12 +260,67 @@ async def _assign_direct_terminal_partitions(
         await _refresh_terminal_topic_metadata(consumer)
 
 
+async def _pin_direct_terminal_end_offsets(
+    consumer: AIOKafkaConsumer,
+    *,
+    timeout_seconds: float,
+) -> None:
+    """Resolve the current end offsets SYNCHRONOUSLY and pin them via ``seek``.
+
+    Why not ``seek_to_end`` (OMN-13118 strike-three)
+    ------------------------------------------------
+    ``AIOKafkaConsumer.seek_to_end`` is LAZY: it calls
+    ``Fetcher.request_offset_reset(partitions, LATEST)``, which only marks the
+    partition "awaiting reset" and registers a position future. The actual LATEST
+    offset is resolved by a ``ListOffsets`` round-trip that happens on the FIRST
+    ``getone`` — which is AFTER the caller has published its command. With
+    generation completing in ~1s (dispatch loop freed, OMN-13010), the correlated
+    terminal lands in the gap between this open() and the first poll, so the
+    lazily-resolved LATEST position is the high-water-mark AFTER the record and
+    the poll starts reading PAST it. The terminal is never read, the trial never
+    correlates, and the experiment matrix re-fires the same cell forever — the
+    consume-leg wedge that survived PR #1969 (set_topics no-op) and PR #1970
+    (partition-less assign-cap) because neither addressed the seek timing.
+
+    ``end_offsets`` is a SYNCHRONOUS broker round-trip that returns the HWM
+    observed NOW (without changing the consumer position); ``seek`` then pins each
+    partition's read offset to that value deterministically. The resume position
+    is therefore fixed at open() time, BEFORE the publish, so any terminal emitted
+    at-or-after this offset is delivered to the subsequent poll. This is the real
+    subscribe-before-publish guarantee the two-phase protocol promised.
+
+    An empty assignment (partition-less reply topic, OMN-13118) has no offsets to
+    resolve, so this is a no-op for that leg.
+
+    ``end_offsets`` is a broker ``ListOffsets`` round-trip that aiokafka documents
+    as able to block indefinitely. It is bounded by ``timeout_seconds`` (the same
+    cap as ``start`` and partition assignment) so a stalled broker fails fast here
+    instead of hanging the pre-publish positioning and reintroducing a wait wedge
+    in a new spot.
+    """
+    assignment = consumer.assignment()
+    if not assignment:
+        return
+    end_offsets = await asyncio.wait_for(
+        consumer.end_offsets(list(assignment)),
+        timeout=timeout_seconds,
+    )
+    for topic_partition, offset in end_offsets.items():
+        consumer.seek(topic_partition, offset)
+
+
 async def open_direct_terminal_consumer(
     *,
     event_bus: object,
     terminal_topic: str,
 ) -> DirectTerminalConsumer:
-    """Start, assign, and seek an ephemeral terminal consumer to the current end."""
+    """Start, assign, and pin an ephemeral terminal consumer to the current end.
+
+    The read position is captured SYNCHRONOUSLY via ``end_offsets`` + ``seek`` (not
+    the lazy ``seek_to_end``) so it is fixed at open() time, before the caller
+    publishes — the subscribe-before-publish guarantee that closes the
+    seek-past-the-terminal race (OMN-13118).
+    """
     consumer = _build_direct_terminal_consumer(event_bus)
     try:
         await asyncio.wait_for(
@@ -277,7 +332,10 @@ async def open_direct_terminal_consumer(
             terminal_topic,
             _DIRECT_TERMINAL_ASSIGN_TIMEOUT_CAP_SECONDS,
         )
-        await consumer.seek_to_end(*consumer.assignment())
+        await _pin_direct_terminal_end_offsets(
+            consumer,
+            timeout_seconds=_DIRECT_TERMINAL_ASSIGN_TIMEOUT_CAP_SECONDS,
+        )
     except BaseException:
         await close_direct_terminal_consumer(
             DirectTerminalConsumer(consumer),
@@ -558,7 +616,14 @@ async def _dispatch_and_wait_with_direct_kafka_consumer(
                 terminal_topics,
                 command.timeout_seconds,
             )
-            await consumer.seek_to_end(*consumer.assignment())
+            # Pin the read position SYNCHRONOUSLY at the current end (OMN-13118):
+            # ``seek_to_end`` is a lazy LATEST reset resolved on the first poll —
+            # AFTER this publish — so a terminal landing in the publish→poll gap
+            # is skipped. ``end_offsets`` + ``seek`` fixes the position now.
+            await _pin_direct_terminal_end_offsets(
+                consumer,
+                timeout_seconds=min(30, command.timeout_seconds),
+            )
             await self._publish_worker_command(command, route)
 
             deadline = asyncio.get_running_loop().time() + command.timeout_seconds

tests/integration/test_pattern_b_broker_terminal_waiter.py

@@ -46,6 +46,9 @@ def __init__(self, *topics: object, **kwargs: object) -> None:
         self.topics_calls = 0
         self.set_topics_calls: list[tuple[str, ...]] = []
         self.seeked_to_end = False
+        # True once the read position is pinned to the current end (synchronous
+        # end_offsets + seek, OMN-13118).
+        self.positioned = False
         self.started = False
         self.stopped = False
         type(self).created.append(self)
@@ -72,6 +75,16 @@ def assignment(self) -> set[object]:
 
     async def seek_to_end(self, *_assignment: object) -> None:
         self.seeked_to_end = True
+        self.positioned = True
+
+    async def end_offsets(self, partitions: object) -> dict[object, int]:
+        # Synchronous HWM round-trip (OMN-13118). Queue-backed fake delivers
+        # regardless of offset, so 0 is faithful for the next-message offset.
+        return dict.fromkeys(partitions, 0)
+
+    def seek(self, _partition: object, _offset: int) -> None:
+        # Synchronous position pin (OMN-13118).
+        self.positioned = True
 
     async def getone(self) -> SimpleNamespace:
         return await self.messages.get()
@@ -117,7 +130,7 @@ async def publish(
         assert consumer.kwargs["group_id"] is None
         assert consumer.topics == ()
         assert consumer.assigned_partitions
-        assert consumer.seeked_to_end is True
+        assert consumer.positioned is True
 
         command_envelope = ModelEventEnvelope[object].model_validate_json(value)
         terminal_envelope = ModelEventEnvelope[object](

tests/integration/test_terminal_consumer_battery_load_wedge.py

@@ -225,6 +225,17 @@ async def seek_to_end(self, *assignment: object) -> None:
         assert self._topic is not None
         self._cursor = self._cluster.logs[self._topic].hwm
 
+    async def end_offsets(self, partitions: Any) -> dict[Any, int]:
+        # Synchronous HWM round-trip (OMN-13118): the offset of the next message,
+        # captured NOW without changing the consumer position.
+        assert self._topic is not None
+        hwm = self._cluster.logs[self._topic].hwm
+        return dict.fromkeys(partitions, hwm)
+
+    def seek(self, _partition: Any, offset: int) -> None:
+        # Synchronous position pin (OMN-13118) replacing the lazy seek_to_end.
+        self._cursor = offset
+
     async def getone(self) -> SimpleNamespace:
         assert self._topic is not None
         log = self._cluster.logs[self._topic]

tests/integration/test_terminal_consumer_concurrent_assign_race.py

@@ -155,6 +155,15 @@ def assignment(self) -> set[Any]:
     async def seek_to_end(self, *_assignment: object) -> None:
         return None
 
+    async def end_offsets(self, partitions: Any) -> dict[Any, int]:
+        # Synchronous HWM round-trip (OMN-13118 pinning). This fake delivers by
+        # pending-payload dict rather than offset, so the value is incidental.
+        return dict.fromkeys(partitions, 0)
+
+    def seek(self, _partition: Any, _offset: int) -> None:
+        # Synchronous position pin (OMN-13118): no-op for the payload-dict fake.
+        return None
+
     async def getone(self) -> SimpleNamespace:
         # The COMPLETED-topic consumer delivers each pending correlated terminal
         # exactly once; otherwise (FAILED topic, or no pending payload) block