perf(sessions): batch native-forwarder event posts

A native harness's forwarder mirrors the session one HTTP POST at a
time — a request per transcript item, a request per streamed text
chunk. On loopback that is invisible. From another region it caps the
forwarder at roughly one event per round trip, so a turn that the pane
finished in seconds keeps trickling into the web UI for tens of
seconds.

Add `POST /v1/sessions/{id}/events/batch`, which runs a run of events
through the same handler as the single-post route, in order, and
reports each one's outcome so a caller can advance a durable cursor
over the delivered prefix instead of re-sending events that already
landed. Point the claude-native delta forwarder at it: a poll's chunks
now cost one round trip instead of one each, with a per-event fallback
for deployments older than the route.

At `--network-delay-ms 100`, mirroring one poll of streamed text drops
from ~2.52s to ~0.14s (24 requests to 1); the two new
`forward_native_deltas_*` benchmark journeys are the matched pair that
measures it and guards the regression.

Co-authored-by: Isaac <no-reply@databricks.com>
Signed-off-by: harry-yao_data <harry.yao@databricks.com>

dev/benchmarks/omnigent/README.md

@@ -62,6 +62,15 @@ see *Network* below).
 | `add_comment` | `POST /v1/sessions/{id}/comments` — create a review comment | write path |
 | `list_projects` | `GET /v1/sessions/projects` — sidebar project list (dual-read union) | project count |
 | `list_project_sessions` | `GET /v1/sessions?project=` — a project folder's sessions (dual-read filter) | sessions/project |
+| `forward_native_deltas_per_event` | `POST /v1/sessions/{id}/events` × 24 — a native forwarder mirroring one poll's streamed text one request at a time | round-trip count |
+| `forward_native_deltas_batched` | `POST /v1/sessions/{id}/events/batch` — the same poll's text in one request | round-trip count |
+
+The two `forward_native_deltas_*` journeys are a matched pair: same work, 24
+round trips versus 1. Run them with `--network-delay-ms` to price the wire,
+which is what a user in another region actually pays — at `--network-delay-ms
+100` the per-event journey takes ~2.5s per poll against ~0.14s batched, and a
+turn is many polls. They are also the regression guard: if the forwarder ever
+fans back out into per-event posts, the batched journey's `HTTP/op` climbs off 1.
 
 Read journeys target a **pre-seeded** session when the DB has a corpus; against
 an empty DB they self-seed a small fallback session over HTTP (the

dev/benchmarks/omnigent/journeys.py

@@ -51,6 +51,7 @@ import subprocess
 import sys
 import tempfile
 import time
+import uuid
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from typing import Literal, cast
@@ -789,6 +790,68 @@ async def _measure_policy_evaluate(env: BenchEnvironment, ctx: JourneyContext) -
     resp.raise_for_status()
 
 
+# ── native-forwarder event mirroring (per-event vs batched) ──────────────────
+
+# Chunks in one simulated poll's worth of streamed assistant text. A native
+# harness produces this many in well under a second, and the forwarder used to
+# spend one round trip on each — which is why a far-from-server client's live
+# view fell behind. Pair these two journeys with ``--network-delay-ms`` to see
+# the round-trip cost directly: the per-event journey pays it 24 times, the
+# batched one once.
+_FORWARD_DELTA_COUNT = 24
+
+# One op is 24 requests for the per-event journey, so the default 100 iterations
+# would let this pair dominate the CI leg (and add tail noise to the journeys
+# after it). The pair exists to price round trips, which a handful of samples
+# establishes.
+_FORWARD_DELTA_MAX_ITERATIONS = 20
+
+
+def _forward_delta_events(message_id: str) -> list[dict[str, object]]:
+    """
+    Build one poll's worth of streamed-text events.
+
+    :param message_id: Assistant message the chunks belong to, so the
+        server scopes them to one in-flight buffer.
+    :returns: ``external_output_text_delta`` event bodies, in order.
+    """
+    return [
+        {
+            "type": "external_output_text_delta",
+            "data": {
+                "delta": f"chunk-{index} ",
+                "message_id": message_id,
+                "index": index,
+                "final": index == _FORWARD_DELTA_COUNT - 1,
+            },
+        }
+        for index in range(_FORWARD_DELTA_COUNT)
+    ]
+
+
+async def _measure_forward_deltas_per_event(env: BenchEnvironment, ctx: JourneyContext) -> None:
+    """Mirror a poll's chunks one request at a time (the pre-batching path)."""
+    session_id = cast(str, ctx)  # _setup_target_session
+    assert env.client is not None
+    for event in _forward_delta_events(f"bench-{uuid.uuid4().hex}"):
+        resp = await env.client.post(f"/v1/sessions/{session_id}/events", json=event)
+        resp.raise_for_status()
+
+
+async def _measure_forward_deltas_batched(env: BenchEnvironment, ctx: JourneyContext) -> None:
+    """Mirror the same chunks in one request."""
+    session_id = cast(str, ctx)  # _setup_target_session
+    assert env.client is not None
+    resp = await env.client.post(
+        f"/v1/sessions/{session_id}/events/batch",
+        json={
+            "events": _forward_delta_events(f"bench-{uuid.uuid4().hex}"),
+            "on_error": "continue",
+        },
+    )
+    resp.raise_for_status()
+
+
 # ── CLI startup (omnigent polly against the local bench server) ──────────────
 
 # Signal that the REPL is ready — the last spinner message before the prompt.
@@ -1015,6 +1078,31 @@ ALL_JOURNEYS: dict[str, Journey] = {
             description="POST /v1/sessions/{id}/policies/evaluate — PreToolUse hook "
             "(single tree scan, preloaded conversation row, caches warm).",
         ),
+        Journey(
+            name="forward_native_deltas_per_event",
+            kind="latency",
+            measure=_measure_forward_deltas_per_event,
+            setup=_setup_target_session,
+            concurrency_safe=True,
+            max_iterations=_FORWARD_DELTA_MAX_ITERATIONS,
+            description=(
+                f"POST /v1/sessions/{{id}}/events × {_FORWARD_DELTA_COUNT} — a native "
+                "forwarder mirroring one poll's streamed text one request at a time."
+            ),
+        ),
+        Journey(
+            name="forward_native_deltas_batched",
+            kind="latency",
+            measure=_measure_forward_deltas_batched,
+            setup=_setup_target_session,
+            concurrency_safe=True,
+            max_iterations=_FORWARD_DELTA_MAX_ITERATIONS,
+            description=(
+                "POST /v1/sessions/{id}/events/batch — the same poll's streamed text "
+                "in one request. Compare against forward_native_deltas_per_event "
+                "under --network-delay-ms."
+            ),
+        ),
         # Runner (full-turn) journeys — with_runner=True, openai-agents, mock LLM.
         Journey(
             name="session_cold_start",

omnigent/_native_post_delivery.py

@@ -23,6 +23,7 @@ import contextlib
 import json
 import logging
 import time
+import weakref
 from collections.abc import Callable, Coroutine
 from dataclasses import dataclass
 from pathlib import Path
@@ -218,6 +219,199 @@ async def post_external_session_status(
     resp.raise_for_status()
 
 
+# Batch event ingestion (``POST /v1/sessions/{id}/events/batch``). A forwarder
+# far from the server is otherwise capped at one event per round trip, so a
+# turn's live text trickles in for tens of seconds after the harness produced
+# it. Batching collapses a poll's worth of events into one request.
+#
+# Kept at half the server's ``MAX_SESSION_EVENTS_PER_BATCH`` (256) so a client
+# never trips the server's envelope validation, and so one request stays small
+# enough to retry cheaply. Longer runs are chunked.
+MAX_EVENTS_PER_BATCH = 128
+
+# Clients whose server has no batch route. Latched on the first route-miss 404,
+# so talking to an older deployment costs one wasted request per client, not one
+# per event. Keyed by client (weakly) rather than by base URL: a forwarder holds
+# one client for its whole life, and nothing leaks into an unrelated client that
+# happens to share a server.
+_EVENTS_BATCH_UNSUPPORTED: weakref.WeakSet[httpx.AsyncClient] = weakref.WeakSet()
+
+
+def events_batch_supported(client: httpx.AsyncClient) -> bool:
+    """
+    Report whether this server is still believed to serve the batch route.
+
+    :param client: Omnigent HTTP client.
+    :returns: ``False`` once a route-miss 404 latched the fallback.
+    """
+    return client not in _EVENTS_BATCH_UNSUPPORTED
+
+
+def reset_events_batch_support(client: httpx.AsyncClient | None = None) -> None:
+    """
+    Clear the batch-unsupported latch (tests; and after a server upgrade).
+
+    :param client: Client to clear, or ``None`` to clear every latch.
+    :returns: None.
+    """
+    if client is None:
+        _EVENTS_BATCH_UNSUPPORTED.clear()
+        return
+    _EVENTS_BATCH_UNSUPPORTED.discard(client)
+
+
+def _is_route_miss(response: httpx.Response) -> bool:
+    """
+    Distinguish "this server has no such route" from "no such session".
+
+    Omnigent's error handler shapes application 404s as
+    ``{"error": {...}}``; Starlette's route miss answers
+    ``{"detail": "Not Found"}``. Only the latter means the deployment
+    predates the batch endpoint.
+
+    :param response: The 404 response to classify.
+    :returns: ``True`` when the 404 came from routing, not the handler.
+    """
+    try:
+        body = response.json()
+    except ValueError:
+        return True
+    return not (isinstance(body, dict) and "error" in body)
+
+
+@dataclass(frozen=True)
+class BatchedEventOutcome:
+    """
+    Per-event outcome of one batched session-event POST.
+
+    :param index: Position of the event in the submitted run.
+    :param delivered: ``True`` when the server accepted the event (the
+        status the equivalent single post would have returned was 2xx).
+    :param status: That status, or ``None`` when the event was never
+        attempted because an earlier failure stopped the batch.
+    :param error: Failure reason reported by the server, when present.
+    """
+
+    index: int
+    delivered: bool
+    status: int | None = None
+    error: str | None = None
+
+
+def _parse_batch_results(
+    payload: object,
+    *,
+    total: int,
+    offset: int,
+) -> list[BatchedEventOutcome]:
+    """
+    Convert one batch response body into per-event outcomes.
+
+    Events the server never attempted (it stopped at a failure) are
+    reported as not-delivered with a ``None`` status so callers keep them
+    for the next attempt instead of treating silence as success.
+
+    :param payload: Decoded JSON body of the batch response.
+    :param total: Number of events submitted in this request.
+    :param offset: Index of this request's first event within the caller's
+        full run, so outcomes are numbered in the caller's terms.
+    :returns: One outcome per submitted event, in order.
+    """
+    raw_results = payload.get("results") if isinstance(payload, dict) else None
+    by_index: dict[int, BatchedEventOutcome] = {}
+    if isinstance(raw_results, list):
+        for position, entry in enumerate(raw_results):
+            if not isinstance(entry, dict):
+                continue
+            raw_index = entry.get("index")
+            index = raw_index if isinstance(raw_index, int) else position
+            raw_status = entry.get("status")
+            status = raw_status if isinstance(raw_status, int) else None
+            raw_error = entry.get("error")
+            by_index[index] = BatchedEventOutcome(
+                index=offset + index,
+                delivered=status is not None and 200 <= status < 400,
+                status=status,
+                error=raw_error if isinstance(raw_error, str) else None,
+            )
+    return [
+        by_index.get(index, BatchedEventOutcome(index=offset + index, delivered=False))
+        for index in range(total)
+    ]
+
+
+async def post_session_events_batch(
+    client: httpx.AsyncClient,
+    *,
+    session_id: str,
+    events: list[dict[str, object]],
+    on_error: str = "stop",
+) -> list[BatchedEventOutcome] | None:
+    """
+    POST a run of session events in as few round trips as possible.
+
+    Each event is the same ``{"type": ..., "data": ...}`` body a single
+    ``POST /v1/sessions/{id}/events`` would carry, and the server
+    dispatches them in order through that same handler — so this changes
+    only how many times the wire is crossed. That is the whole point: on
+    a client far from the server, per-event posting caps the forwarder at
+    a few events per second and its live view falls behind the harness.
+
+    Runs longer than :data:`MAX_EVENTS_PER_BATCH` are chunked. With
+    ``on_error="stop"`` a chunk that ends in a failure stops the run
+    there, so a caller advancing a cursor over the delivered prefix
+    behaves exactly as it did when posting one event at a time.
+
+    :param client: Omnigent HTTP client.
+    :param session_id: Omnigent session/conversation id.
+    :param events: Event bodies to dispatch, in order. Empty is a no-op.
+    :param on_error: ``"stop"`` to leave the rest of a chunk unattempted
+        after a failure (the default, matching a sequential caller), or
+        ``"continue"`` to attempt every event — for best-effort streams
+        where dropping one chunk beats stalling the tail.
+    :returns: One outcome per event, in order; or ``None`` when this
+        server has no batch route and the caller should post singly.
+    :raises httpx.HTTPError: On a transport failure or an
+        envelope-level rejection. Delivery of the events in the failed
+        chunk is then unknown, exactly as for a single post whose
+        response was lost.
+    """
+    if not events:
+        return []
+    if not events_batch_supported(client):
+        return None
+    outcomes: list[BatchedEventOutcome] = []
+    for offset in range(0, len(events), MAX_EVENTS_PER_BATCH):
+        chunk = events[offset : offset + MAX_EVENTS_PER_BATCH]
+        response = await client.post(
+            f"/v1/sessions/{session_id}/events/batch",
+            json={"events": chunk, "on_error": on_error},
+        )
+        if response.status_code == 404 and _is_route_miss(response):
+            _EVENTS_BATCH_UNSUPPORTED.add(client)
+            _logger.debug(
+                "Server has no session-events batch route; falling back to per-event posts for %s",
+                client.base_url,
+            )
+            return None
+        response.raise_for_status()
+        note_native_post_success()
+        try:
+            payload = response.json()
+        except ValueError as exc:
+            raise httpx.HTTPError(f"malformed session-events batch response: {exc}") from exc
+        chunk_outcomes = _parse_batch_results(payload, total=len(chunk), offset=offset)
+        outcomes.extend(chunk_outcomes)
+        if on_error == "stop" and not all(outcome.delivered for outcome in chunk_outcomes):
+            # The server stopped at a failure; everything after it in this
+            # chunk was never attempted, and later chunks must not jump the
+            # queue ahead of it.
+            for index in range(offset + len(chunk), len(events)):
+                outcomes.append(BatchedEventOutcome(index=index, delivered=False))
+            break
+    return outcomes
+
+
 async def post_session_event_with_retry(
     *,
     client: httpx.AsyncClient,

omnigent/claude_native_forwarder.py

@@ -20,6 +20,7 @@ from omnigent._native_post_delivery import (
     append_dead_letter,
     post_external_session_status,
     post_may_have_been_delivered,
+    post_session_events_batch,
 )
 from omnigent.claude_native_bridge import (
     BRIDGE_ID_LABEL_KEY,
@@ -4021,19 +4022,129 @@ async def _post_external_output_text_delta(
     """
     resp = await client.post(
         f"/v1/sessions/{session_id}/events",
-        json={
-            "type": "external_output_text_delta",
-            "data": {
-                "delta": delta.delta,
-                "message_id": delta.message_id,
-                "index": delta.index,
-                "final": delta.final,
-            },
-        },
+        json=_output_text_delta_event(delta),
     )
     resp.raise_for_status()
 
 
+def _output_text_delta_event(delta: ClaudeMessageDelta) -> dict[str, object]:
+    """
+    Build the ``external_output_text_delta`` event body for one chunk.
+
+    :param delta: Parsed streamed chunk.
+    :returns: The event body a single or batched post carries.
+    """
+    return {
+        "type": "external_output_text_delta",
+        "data": {
+            "delta": delta.delta,
+            "message_id": delta.message_id,
+            "index": delta.index,
+            "final": delta.final,
+        },
+    }
+
+
+def _log_dropped_delta(
+    *,
+    session_id: str,
+    bridge_dir: Path,
+    delta: ClaudeMessageDelta,
+    http_status: int | None = None,
+    error_type: str | None = None,
+) -> None:
+    """
+    Note one streamed chunk that never reached the server.
+
+    Records the status and the exception's class, never a rendered
+    exception or server message: an httpx error's text carries the request
+    it was made with, and this log is not the place to spill anything that
+    travelled in a header.
+
+    :param session_id: Omnigent session/conversation id.
+    :param bridge_dir: Native Claude bridge directory.
+    :param delta: The chunk that was dropped.
+    :param http_status: Status the server reported, when it responded.
+    :param error_type: Exception class name, for a transport failure.
+    :returns: None.
+    """
+    _logger.debug(
+        "Dropping Claude streamed delta after HTTP failure; session=%s "
+        "bridge_dir=%s message_id=%s index=%s http_status=%s error_type=%s",
+        session_id,
+        bridge_dir,
+        delta.message_id,
+        delta.index,
+        http_status,
+        error_type,
+    )
+
+
+async def _forward_delta_run(
+    client: httpx.AsyncClient,
+    *,
+    session_id: str,
+    bridge_dir: Path,
+    deltas: list[ClaudeMessageDelta],
+) -> None:
+    """
+    Publish a poll's worth of streamed chunks in as few round trips as possible.
+
+    One request for the whole run when the server serves the batch route,
+    else one per chunk as before. Deltas are best-effort live preview, so
+    a failure is logged and dropped (the authoritative final text still
+    arrives via ``external_conversation_item``) — never retried, so a
+    transient blip can't wedge the tail. That is also why the batch runs
+    with ``on_error="continue"``: one rejected chunk must not shadow the
+    rest of the run.
+
+    :param client: Omnigent HTTP client.
+    :param session_id: Omnigent session/conversation id.
+    :param bridge_dir: Native Claude bridge directory (logging only).
+    :param deltas: The new chunks to publish, in order.
+    :returns: None.
+    """
+    try:
+        outcomes = await post_session_events_batch(
+            client,
+            session_id=session_id,
+            events=[_output_text_delta_event(delta) for delta in deltas],
+            on_error="continue",
+        )
+    except httpx.HTTPError as exc:
+        for delta in deltas:
+            _log_dropped_delta(
+                session_id=session_id,
+                bridge_dir=bridge_dir,
+                delta=delta,
+                http_status=_http_status_for_log(exc),
+                error_type=type(exc).__name__,
+            )
+        return
+    if outcomes is not None:
+        for delta, outcome in zip(deltas, outcomes, strict=False):
+            if not outcome.delivered:
+                _log_dropped_delta(
+                    session_id=session_id,
+                    bridge_dir=bridge_dir,
+                    delta=delta,
+                    http_status=outcome.status,
+                )
+        return
+    # Server predates the batch route: post one at a time.
+    for delta in deltas:
+        try:
+            await _post_external_output_text_delta(client, session_id=session_id, delta=delta)
+        except httpx.HTTPError as exc:
+            _log_dropped_delta(
+                session_id=session_id,
+                bridge_dir=bridge_dir,
+                delta=delta,
+                http_status=_http_status_for_log(exc),
+                error_type=type(exc).__name__,
+            )
+
+
 async def _forward_available_deltas(
     *,
     client: httpx.AsyncClient,
@@ -4076,6 +4187,7 @@ async def _forward_available_deltas(
     )
     if result.byte_offset == state.byte_offset and not result.deltas:
         return state
+    fresh: list[ClaudeMessageDelta] = []
     for delta in result.deltas:
         key = (delta.message_id, delta.index)
         if key in seen_keys:
@@ -4086,18 +4198,11 @@ async def _forward_available_deltas(
         # limit.
         while len(seen_keys) > _MAX_SEEN_DELTA_KEYS:
             del seen_keys[next(iter(seen_keys))]
-        try:
-            await _post_external_output_text_delta(client, session_id=session_id, delta=delta)
-        except httpx.HTTPError as exc:
-            _logger.debug(
-                "Dropping Claude streamed delta after HTTP failure; session=%s "
-                "bridge_dir=%s message_id=%s index=%s http_status=%s",
-                session_id,
-                bridge_dir,
-                delta.message_id,
-                delta.index,
-                _http_status_for_log(exc),
-            )
+        fresh.append(delta)
+    if fresh:
+        await _forward_delta_run(
+            client, session_id=session_id, bridge_dir=bridge_dir, deltas=fresh
+        )
     updated = DeltaForwardState(byte_offset=result.byte_offset)
     await _write_delta_forward_state_async(bridge_dir, updated)
     return updated

omnigent/server/API.md

@@ -1026,6 +1026,54 @@ mirror that status into the parent stream as `session.child_session.updated`
 when the child is registered for fan-out. New user-facing event types should
 default to the queue.
 
+### Post Events (batch)
+
+```
+POST /v1/sessions/{session_id}/events/batch
+Content-Type: application/json
+
+{
+  "events": [
+    {"type": "external_output_text_delta", "data": {"delta": "Hel"}},
+    {"type": "external_output_text_delta", "data": {"delta": "lo"}}
+  ],
+  "on_error": "stop"
+}
+```
+
+Dispatches a run of events through the same handler as `POST
+.../events`, in request order — one round trip instead of
+`len(events)`. Internal (hidden from the OpenAPI reference): the native
+harness forwarders use it to mirror a poll's transcript items and text
+deltas at once. Without it a client far from the server can push only
+about one event per round trip, so a long turn's live view falls tens of
+seconds behind the harness.
+
+`events` carries 1–256 `SessionEventInput` bodies. `on_error` selects
+what happens after a failure: `"stop"` (default) leaves the rest of the
+batch unattempted, matching a sequential caller that retries from its
+first failure; `"continue"` attempts every event, for best-effort
+streams (live deltas) where dropping one chunk beats stalling the tail.
+
+Always `200` with per-event outcomes; a rejected event does not fail the
+request, so a caller can advance a durable cursor over the delivered
+prefix instead of re-sending — and duplicating — events that landed:
+
+```
+{
+  "results": [
+    {"index": 0, "status": 202, "body": {"queued": false}},
+    {"index": 1, "status": 400, "error": "...", "code": "invalid_input"}
+  ],
+  "stopped_at": 1
+}
+```
+
+`results` is shorter than `events` when `stopped_at` is set — entries
+past it were never attempted. Clients tell "this deployment has no batch
+route" from "no such session" by the 404 body: a route miss has no
+`error` envelope, an application 404 does.
+
 ### Resolve Elicitation (URL-based)
 
 ```

omnigent/server/routes/sessions/routes_events.py

@@ -14,7 +14,9 @@ from fastapi import (
     Request,
 )
 from fastapi.responses import StreamingResponse
+from sqlalchemy.exc import StatementError
 
+from omnigent.db.db_models import InvalidUuidError
 from omnigent.entities import (
     ErrorData,
     NewConversationItem,
@@ -200,6 +202,7 @@ from omnigent.server.schemas import (
     ElicitationRequestParams,
     ErrorDetail,
     McpServerStartup,
+    SessionEventBatchInput,
     SessionEventInput,
 )
 from omnigent.session_lifecycle import (
@@ -1738,6 +1741,120 @@ def register_events_routes(
             response["pending_id"] = dispatch.pending_id
         return response
 
+    # ── POST /sessions/{session_id}/events/batch ─────────────
+
+    @router.post(
+        "/sessions/{session_id}/events/batch",
+        # Internal event ingestion — hidden from the public API reference.
+        include_in_schema=False,
+        # response_model=None: the body is a small per-event outcome
+        # envelope, not a domain model.
+        response_model=None,
+    )
+    async def post_events_batch(
+        request: Request,
+        session_id: str,
+        body: SessionEventBatchInput,
+    ) -> dict[str, Any]:
+        """
+        Dispatch a run of session events in one round trip.
+
+        Every event runs through the same handler as
+        ``POST /sessions/{session_id}/events``, in request order, so a
+        batch is observationally equivalent to the sequence of single
+        posts it replaces. What it saves is round trips: a native
+        harness's forwarder posts one transcript item or text delta per
+        request, so a client an ocean away from the server can only push
+        a handful of events per second and its live view falls far behind
+        a long turn.
+
+        Per-event failures land in ``results`` instead of failing the
+        request, so a caller can advance its cursor over the successful
+        prefix and retry from the first failure rather than re-sending
+        (and duplicating) events that already landed. ``on_error="stop"``
+        leaves everything after the failure unattempted, matching a
+        sequential caller; ``"continue"`` attempts all of them, for
+        best-effort delta streams. Envelope-level problems (unparseable
+        body, oversized batch) are rejected by validation as usual.
+
+        :param session_id: Session/conversation identifier.
+        :param body: Events to dispatch plus the on-error policy.
+        :returns: ``{"results": [...], "stopped_at": int | None}`` — see
+            :class:`SessionEventBatchResponse`.
+        """
+        results: list[dict[str, Any]] = []
+        stopped_at: int | None = None
+        for index, event in enumerate(body.events):
+            try:
+                outcome = await post_event(request, session_id, event)
+            except OmnigentError as exc:
+                # A wrong-replica landing is a fact about the *request*, not the
+                # event: on a multi-replica deployment the caller is expected to
+                # re-address and re-send. Reporting it per-event would hand the
+                # caller an opaque 400 it reads as a permanent rejection — the
+                # forwarder would dead-letter and drop a live item instead of
+                # retrying it at the right replica. Every event in a batch
+                # shares the session's replica binding, so this fires before
+                # anything is dispatched; propagate then, exactly as a single
+                # post does. If it somehow lands mid-batch (a tunnel that moved
+                # under us), keep the delivered prefix accounted for and report
+                # it with its code so the caller can still re-address.
+                if exc.code == ErrorCode.WRONG_REPLICA and not results:
+                    raise
+                results.append(
+                    {
+                        "index": index,
+                        "status": exc.http_status,
+                        "error": str(exc),
+                        "code": exc.code,
+                    }
+                )
+            except StatementError as exc:
+                # A malformed id can address no row. The app's handler maps
+                # this to 404 for a single post; mirror it so a batched event
+                # reports the same status instead of a bogus 500.
+                if not isinstance(exc.orig, InvalidUuidError):
+                    raise
+                results.append(
+                    {
+                        "index": index,
+                        "status": 404,
+                        "error": "Not found.",
+                        "code": ErrorCode.NOT_FOUND,
+                    }
+                )
+            except Exception:
+                # A single post would have failed only its own request;
+                # propagating here would fail the whole batch and make the
+                # caller re-send events that already landed. Report it
+                # per-event and keep the log loud so the bug stays visible.
+                #
+                # The body carries the same generic message the app's 500
+                # handler returns, never the exception's text: an internal
+                # failure's detail (paths, SQL, stack context) must not reach
+                # a client just because it happened inside a batch.
+                _logger.exception(
+                    "Batched session event failed; session=%s index=%s type=%s",
+                    session_id,
+                    index,
+                    event.type,
+                )
+                results.append(
+                    {
+                        "index": index,
+                        "status": 500,
+                        "error": "An internal error occurred.",
+                        "code": ErrorCode.INTERNAL_ERROR,
+                    }
+                )
+            else:
+                results.append({"index": index, "status": 202, "body": dict(outcome)})
+                continue
+            if body.on_error == "stop":
+                stopped_at = index
+                break
+        return {"results": results, "stopped_at": stopped_at}
+
     # ── GET /sessions/{session_id}/stream ────────────────────────
 
     # Live-tail only. Clients reconnect via GET /v1/sessions/{id}

omnigent/server/schemas.py

@@ -1197,6 +1197,74 @@ class SessionEventInput(BaseModel):
     created_by: str | None = None
 
 
+# Cap on how many events one batch may carry. Bounds the worst-case work a
+# single request can queue behind the event loop (each event runs the full
+# ``POST /events`` dispatch) while staying far above what a forwarder poll
+# produces in practice.
+MAX_SESSION_EVENTS_PER_BATCH = 256
+
+
+class SessionEventBatchInput(BaseModel):
+    """
+    Body of ``POST /v1/sessions/{id}/events/batch``.
+
+    Carries a run of events that would otherwise be posted one at a
+    time. Events are dispatched **in order** through the same code path
+    as ``POST /v1/sessions/{id}/events``, so a batch is observationally
+    equivalent to the sequence of single posts it replaces — it just
+    costs one round trip instead of ``len(events)``. This matters most
+    for clients far from the server, where a native harness's transcript
+    and delta forwarders are otherwise rate-limited to one event per RTT.
+
+    :param events: The events to dispatch, in order. At least one, at
+        most :data:`MAX_SESSION_EVENTS_PER_BATCH`.
+    :param on_error: What to do when an event fails. ``"stop"`` (the
+        default) leaves the remaining events unattempted, mirroring a
+        sequential caller that stops at its first failure and retries
+        from there. ``"continue"`` attempts every event regardless, for
+        best-effort streams (live text deltas) where a dropped chunk is
+        preferable to stalling the tail.
+    """
+
+    events: list[SessionEventInput] = Field(min_length=1, max_length=MAX_SESSION_EVENTS_PER_BATCH)
+    on_error: Literal["stop", "continue"] = "stop"
+
+
+class SessionEventBatchResult(BaseModel):
+    """
+    Outcome of one event inside a ``POST /events/batch`` request.
+
+    :param index: Position of the event in the request's ``events``.
+    :param status: HTTP status the equivalent single post would have
+        returned — ``202`` on success, otherwise the error's status.
+    :param body: The single-post response body, on success.
+    :param error: Human-readable failure reason, on failure.
+    :param code: Machine-readable Omnigent error code, on failure.
+    """
+
+    index: int
+    status: int
+    body: dict[str, Any] | None = None
+    error: str | None = None
+    code: str | None = None
+
+
+class SessionEventBatchResponse(BaseModel):
+    """
+    Response of ``POST /v1/sessions/{id}/events/batch``.
+
+    :param results: One entry per *attempted* event, in request order.
+        Shorter than the request's ``events`` when ``on_error="stop"``
+        tripped: entries after ``stopped_at`` were never attempted and
+        the caller should re-send them.
+    :param stopped_at: Index of the failure that ended the batch early,
+        or ``None`` when every event was attempted.
+    """
+
+    results: list[SessionEventBatchResult]
+    stopped_at: int | None = None
+
+
 class SessionGitOptions(BaseModel):
     """
     Git worktree options for ``POST /v1/sessions``.

tests/server/integration/test_sessions_endpoints.py

@@ -23,6 +23,7 @@ from unittest.mock import AsyncMock, patch
 import httpx
 import pytest
 
+from omnigent.errors import ErrorCode, OmnigentError
 from omnigent.llms.context_window import ModelPricing
 from omnigent.runtime.tool_output import MAX_TOOL_OUTPUT_BYTES
 from omnigent.server.background_session_titles import BackgroundTitleRequest
@@ -30,6 +31,7 @@ from omnigent.server.routes._sessions.helpers import (
     _NativeTerminalEnsureOutcome,
     _RunnerForwardResult,
 )
+from omnigent.server.schemas import MAX_SESSION_EVENTS_PER_BATCH
 from omnigent.spec.types import SkillSpec
 from omnigent.stores.conversation_store.sqlalchemy_store import (
     SqlAlchemyConversationStore,
@@ -3967,6 +3969,295 @@ async def test_post_external_output_text_delta_rejects_malformed_delta(
     assert published == []
 
 
+def _text_delta_event(text: str) -> dict[str, Any]:
+    """
+    Build one ``external_output_text_delta`` event body.
+
+    :param text: The chunk text, e.g. ``"hel"``.
+    :returns: An event body for ``/events`` or ``/events/batch``.
+    """
+    return {"type": "external_output_text_delta", "data": {"delta": text}}
+
+
+async def test_post_events_batch_dispatches_every_event_in_order(
+    client: httpx.AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    A batch is the sequence of single posts it replaces, in one round trip.
+
+    Native forwarders post one transcript item or text chunk per request,
+    which caps a far-away client at a few events per second and leaves its
+    live view far behind the harness. Batching only changes the number of
+    round trips, so ordering and per-event acknowledgement must match the
+    single-post path exactly. Fails if events are reordered, dropped, or
+    reported without their individual status.
+    """
+    published: list[tuple[str, dict[str, Any]]] = []
+    monkeypatch.setattr(
+        "omnigent.server.routes.sessions.session_stream.publish",
+        lambda sid, ev: published.append((sid, ev)),
+    )
+    agent = await create_test_agent(client)
+    session = await _create_session(client, agent["id"])
+
+    resp = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={"events": [_text_delta_event(text) for text in ("a", "b", "c")]},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["stopped_at"] is None
+    assert [entry["status"] for entry in body["results"]] == [202, 202, 202]
+    assert [entry["index"] for entry in body["results"]] == [0, 1, 2]
+    assert [entry["body"] for entry in body["results"]] == [{"queued": False}] * 3
+    assert [event["delta"] for _sid, event in published] == ["a", "b", "c"]
+
+
+async def test_post_events_batch_persists_items_like_single_posts(
+    client: httpx.AsyncClient,
+) -> None:
+    """
+    Batched ``external_conversation_item`` events land in history, in order.
+
+    This is the equivalence that lets a forwarder mirror a whole poll's
+    transcript in one request. Fails if the batch path skips persistence
+    or scrambles item order (resume would replay a garbled transcript).
+    """
+    agent = await create_test_agent(client)
+    session = await _create_session(client, agent["id"])
+
+    resp = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={
+            "events": [
+                {
+                    "type": "external_conversation_item",
+                    "data": {
+                        "item_type": "message",
+                        "item_data": {
+                            "role": "user",
+                            "content": [{"type": "input_text", "text": text}],
+                            "is_meta": True,
+                        },
+                        "response_id": "turn_1",
+                        "source_id": f"src-{index}",
+                    },
+                }
+                for index, text in enumerate(("first", "second", "third"))
+            ]
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    assert [entry["status"] for entry in resp.json()["results"]] == [202, 202, 202], resp.text
+
+    items = (await client.get(f"/v1/sessions/{session['id']}/items")).json()["data"]
+    texts = [item["content"][0]["text"] for item in items if item["type"] == "message"]
+    assert texts == ["first", "second", "third"]
+
+
+async def test_post_events_batch_stops_at_the_first_failure(
+    client: httpx.AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    By default a failed event leaves the rest of the batch unattempted.
+
+    A sequential caller stops at its first failure and retries from
+    there; the batch must behave the same so a forwarder can advance its
+    cursor over the delivered prefix without re-sending — and therefore
+    duplicating — events that already landed. Fails if later events are
+    dispatched anyway (the caller then can't tell what to retry) or if
+    one bad event fails the whole request.
+    """
+    published: list[tuple[str, dict[str, Any]]] = []
+    monkeypatch.setattr(
+        "omnigent.server.routes.sessions.session_stream.publish",
+        lambda sid, ev: published.append((sid, ev)),
+    )
+    agent = await create_test_agent(client)
+    session = await _create_session(client, agent["id"])
+
+    resp = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={
+            "events": [
+                _text_delta_event("ok"),
+                {"type": "external_output_text_delta", "data": {"delta": {"bad": 1}}},
+                _text_delta_event("never"),
+            ]
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["stopped_at"] == 1
+    assert [entry["status"] for entry in body["results"]] == [202, 400]
+    assert "requires string data.delta" in body["results"][1]["error"]
+    # The third event was never attempted.
+    assert [event["delta"] for _sid, event in published] == ["ok"]
+
+
+async def test_post_events_batch_continue_attempts_every_event(
+    client: httpx.AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    ``on_error="continue"`` keeps going past a rejected event.
+
+    Live text deltas are a preview the final message supersedes, so
+    dropping one chunk beats stalling the tail behind it. Fails if a
+    mid-run rejection swallows the chunks after it.
+    """
+    published: list[tuple[str, dict[str, Any]]] = []
+    monkeypatch.setattr(
+        "omnigent.server.routes.sessions.session_stream.publish",
+        lambda sid, ev: published.append((sid, ev)),
+    )
+    agent = await create_test_agent(client)
+    session = await _create_session(client, agent["id"])
+
+    resp = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={
+            "events": [
+                _text_delta_event("a"),
+                {"type": "external_output_text_delta", "data": {"delta": None}},
+                _text_delta_event("c"),
+            ],
+            "on_error": "continue",
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["stopped_at"] is None
+    assert [entry["status"] for entry in body["results"]] == [202, 400, 202]
+    assert [event["delta"] for _sid, event in published] == ["a", "c"]
+
+
+async def test_post_events_batch_rejects_an_oversized_or_empty_batch(
+    client: httpx.AsyncClient,
+) -> None:
+    """
+    The envelope caps how much work one request can queue.
+
+    Each event runs the full single-post dispatch, so an unbounded batch
+    is an unbounded amount of work behind one request. Fails if the cap
+    or the non-empty requirement stops being enforced.
+    """
+    agent = await create_test_agent(client)
+    session = await _create_session(client, agent["id"])
+
+    too_many = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={"events": [_text_delta_event("x")] * (MAX_SESSION_EVENTS_PER_BATCH + 1)},
+    )
+    assert too_many.status_code == 422, too_many.text
+
+    empty = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={"events": []},
+    )
+    assert empty.status_code == 422, empty.text
+
+
+async def test_post_events_batch_reports_a_missing_session_per_event(
+    client: httpx.AsyncClient,
+) -> None:
+    """
+    An application 404 lands in ``results``, not as a route miss.
+
+    Clients probe this endpoint to learn whether the deployment has it,
+    and tell "no such route" from "no such session" by the error
+    envelope. Fails if a missing session answers with a bare route-miss
+    shape, which would make a client permanently downgrade to per-event
+    posts against a server that supports batching.
+    """
+    resp = await client.post(
+        f"/v1/sessions/{'0' * 32}/events/batch",
+        json={"events": [_text_delta_event("a")]},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["stopped_at"] == 0
+    assert body["results"][0]["status"] == 404
+    assert body["results"][0]["code"]
+
+
+async def test_post_events_batch_propagates_a_wrong_replica_landing(
+    client: httpx.AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    A wrong-replica miss fails the request, not one event inside it.
+
+    On a multi-replica deployment a session-scoped request can land on a
+    replica that doesn't hold the session's tunnel; the contract is a 400
+    carrying ``wrong_replica`` so the caller re-addresses and re-sends.
+    Reporting it per-event instead hands the caller an opaque 400 it reads
+    as a permanent rejection — the forwarder would dead-letter and drop a
+    live transcript item rather than retry it at the right replica. Fails
+    if the code is swallowed into ``results``.
+    """
+    agent = await create_test_agent(client)
+    session = await _create_session(client, agent["id"])
+
+    def _wrong_replica(*_args: Any, **_kwargs: Any) -> None:
+        """Raise the routing miss the real handler raises."""
+        raise OmnigentError(
+            "session runner is on another replica; retry",
+            code=ErrorCode.WRONG_REPLICA,
+        )
+
+    monkeypatch.setattr(
+        "omnigent.server.routes.sessions.session_stream.publish",
+        _wrong_replica,
+    )
+    resp = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={"events": [_text_delta_event("a"), _text_delta_event("b")]},
+    )
+    assert resp.status_code == 400, resp.text
+    assert resp.json()["error"]["code"] == ErrorCode.WRONG_REPLICA
+
+
+async def test_post_events_batch_never_leaks_an_internal_error_detail(
+    client: httpx.AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    An unexpected server-side failure reports a generic message, not its text.
+
+    The batch route has to catch per-event failures rather than let them fail
+    the whole request — otherwise a caller re-sends events that already
+    landed — but that catch must not become a channel for internal detail
+    (paths, SQL, stack context) that the app's own 500 handler withholds.
+    Fails if the exception's text reaches the response body.
+    """
+    agent = await create_test_agent(client)
+    session = await _create_session(client, agent["id"])
+
+    _LEAK_MARKER = "internal-detail-that-must-not-escape"
+
+    def _boom(*_args: Any, **_kwargs: Any) -> None:
+        """Raise an error whose text must not reach the client."""
+        raise RuntimeError(f"{_LEAK_MARKER} at /srv/omnigent/state row 42")
+
+    monkeypatch.setattr(
+        "omnigent.server.routes.sessions.session_stream.publish",
+        _boom,
+    )
+    resp = await client.post(
+        f"/v1/sessions/{session['id']}/events/batch",
+        json={"events": [_text_delta_event("a")]},
+    )
+    assert resp.status_code == 200, resp.text
+    entry = resp.json()["results"][0]
+    assert entry["status"] == 500
+    assert entry["error"] == "An internal error occurred."
+    assert _LEAK_MARKER not in resp.text
+    assert "RuntimeError" not in resp.text
+
+
 async def test_post_external_tool_output_delta_publishes_transient_delta(
     client: httpx.AsyncClient,
     monkeypatch: pytest.MonkeyPatch,

tests/test_claude_native_forwarder.py

@@ -24,6 +24,7 @@ from opentelemetry.sdk.trace.export import SimpleSpanProcessor
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 
 import omnigent.claude_native_forwarder as forwarder
+from omnigent._native_post_delivery import reset_events_batch_support
 from omnigent.claude_native_bridge import (
     BRIDGE_ID_LABEL_KEY,
     ClaudeMessageDelta,
@@ -5898,34 +5899,60 @@ def _write_deltas_file(bridge_dir: Path, records: list[dict[str, Any]]) -> None:
 def _delta_capture_client(
     captured: list[_CapturedDeltaPost],
     status_code: int = 202,
+    *,
+    batch_supported: bool = True,
+    batch_event_statuses: list[int] | None = None,
 ) -> httpx.AsyncClient:
     """
-    Build an AsyncClient whose ``/events`` POSTs are captured.
+    Build an AsyncClient whose ``/events`` and ``/events/batch`` POSTs are captured.
 
     :param captured: List appended to with each observed POST body.
     :param status_code: HTTP status the stub returns, e.g. ``202`` for
         success or ``500`` to exercise the best-effort drop path.
+    :param batch_supported: ``False`` answers the batch route with the
+        route-miss 404 an older deployment returns, so the caller's
+        per-event fallback runs.
+    :param batch_event_statuses: Per-event statuses for the batch reply,
+        to exercise a batch that partially fails. Defaults to all-202.
     :returns: An ``httpx.AsyncClient`` bound to the capturing transport.
     """
 
     def handler(request: httpx.Request) -> httpx.Response:
-        captured.append(
-            _CapturedDeltaPost(url_path=request.url.path, body=json.loads(request.content))
+        body = json.loads(request.content)
+        captured.append(_CapturedDeltaPost(url_path=request.url.path, body=body))
+        if not request.url.path.endswith("/events/batch"):
+            return httpx.Response(status_code, json={"queued": False})
+        if not batch_supported:
+            # Starlette's route miss — no ``error`` envelope, which is how
+            # the client tells "no such route" from "no such session".
+            return httpx.Response(404, json={"detail": "Not Found"})
+        if status_code >= 400:
+            return httpx.Response(status_code, json={"detail": "boom"})
+        statuses = batch_event_statuses or [202] * len(body["events"])
+        return httpx.Response(
+            200,
+            json={
+                "results": [
+                    {"index": index, "status": status} for index, status in enumerate(statuses)
+                ],
+                "stopped_at": None,
+            },
         )
-        return httpx.Response(status_code, json={"queued": False})
 
     return httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="http://ap")
 
 
 async def test_forward_available_deltas_posts_each_and_advances_offset(tmp_path: Path) -> None:
     """
-    Each appended chunk is POSTed as an ``external_output_text_delta``.
+    A poll's chunks go out as one batch of ``external_output_text_delta`` events.
 
     Proves the forwarder turns deltas-file lines into the exact event
-    shape the Omnigent route expects (delta + message_id + index + final) and
-    advances+persists the byte offset so the next poll resumes after
-    them. Fails if a field is dropped (UI can't scope/order the buffer)
-    or the offset doesn't persist (chunks re-POST on restart).
+    shape the Omnigent route expects (delta + message_id + index + final),
+    ships the whole run in a single round trip, and advances+persists the
+    byte offset so the next poll resumes after them. Fails if a field is
+    dropped (UI can't scope/order the buffer), if the run costs a request
+    per chunk again (a far-from-server client then streams at one chunk
+    per RTT), or if the offset doesn't persist (chunks re-POST on restart).
     """
     bridge_dir = prepare_bridge_dir("conv_x", bridge_id="b1", workspace=tmp_path)
     _write_deltas_file(
@@ -5946,12 +5973,10 @@ async def test_forward_available_deltas_posts_each_and_advances_offset(tmp_path:
             seen_keys=seen,
         )
 
-    assert [c.url_path for c in captured] == [
-        "/v1/sessions/conv_x/events",
-        "/v1/sessions/conv_x/events",
-    ]
+    # One request, not one per chunk.
+    assert [c.url_path for c in captured] == ["/v1/sessions/conv_x/events/batch"]
     # Full event shape proves every field survived hook → file → POST.
-    assert [c.body for c in captured] == [
+    assert captured[0].body["events"] == [
         {
             "type": "external_output_text_delta",
             "data": {"delta": "Hello ", "message_id": "m1", "index": 0, "final": False},
@@ -5961,6 +5986,8 @@ async def test_forward_available_deltas_posts_each_and_advances_offset(tmp_path:
             "data": {"delta": "world", "message_id": "m1", "index": 1, "final": True},
         },
     ]
+    # Best-effort preview: a rejected chunk must not shadow the rest of the run.
+    assert captured[0].body["on_error"] == "continue"
     # Offset advanced to EOF and was persisted, so a reload resumes past
     # the two chunks instead of re-POSTing them.
     assert new_state.byte_offset == os.path.getsize(bridge_dir / "message_deltas.jsonl")
@@ -5996,11 +6023,12 @@ async def test_forward_available_deltas_dedupes_by_message_id_and_index(tmp_path
             seen_keys=seen,
         )
     # The duplicate (m1, 0) is collapsed: only the first (m1,0) and the
-    # distinct (m1,1) are POSTed — 2 requests, not 3.
-    assert [(c.body["data"]["message_id"], c.body["data"]["index"]) for c in captured] == [
-        ("m1", 0),
-        ("m1", 1),
-    ]
+    # distinct (m1,1) are sent — 2 events in the batch, not 3.
+    assert len(captured) == 1
+    assert [
+        (event["data"]["message_id"], event["data"]["index"])
+        for event in captured[0].body["events"]
+    ] == [("m1", 0), ("m1", 1)]
 
 
 async def test_forward_available_deltas_drops_on_http_error(tmp_path: Path) -> None:
@@ -6030,6 +6058,126 @@ async def test_forward_available_deltas_drops_on_http_error(tmp_path: Path) -> N
     # The POST was attempted (and 500'd) but no exception escaped, and
     # the offset moved past the chunk so it won't be retried endlessly.
     assert len(captured) == 1
+    assert captured[0].url_path == "/v1/sessions/conv_x/events/batch"
+    assert new_state.byte_offset == os.path.getsize(bridge_dir / "message_deltas.jsonl")
+
+
+async def test_forward_available_deltas_batches_a_long_run_in_one_round_trip(
+    tmp_path: Path,
+) -> None:
+    """
+    A whole turn's worth of chunks costs one request, not one per chunk.
+
+    This is the property that makes streaming usable from far away: an
+    assistant message is hundreds of chunks, and a per-chunk POST caps the
+    live preview at one chunk per round trip, so the web view falls tens of
+    seconds behind the pane. Fails the moment the run fans back out into
+    per-chunk requests.
+    """
+    bridge_dir = prepare_bridge_dir("conv_x", bridge_id="b1", workspace=tmp_path)
+    _write_deltas_file(
+        bridge_dir,
+        [
+            {"message_id": "m1", "index": index, "final": index == 79, "delta": f"c{index}"}
+            for index in range(80)
+        ],
+    )
+    captured: list[_CapturedDeltaPost] = []
+    seen: dict[tuple[str, int], None] = {}
+    async with _delta_capture_client(captured) as client:
+        await forwarder._forward_available_deltas(
+            client=client,
+            session_id="conv_x",
+            bridge_dir=bridge_dir,
+            state=forwarder.DeltaForwardState(),
+            seen_keys=seen,
+        )
+    assert len(captured) == 1
+    assert len(captured[0].body["events"]) == 80
+
+
+async def test_forward_available_deltas_falls_back_when_server_lacks_batch_route(
+    tmp_path: Path,
+) -> None:
+    """
+    An older server's route-miss 404 falls back to one POST per chunk.
+
+    A user's CLI can be newer than the deployment it talks to, and the
+    live preview must keep working there — just without the round-trip
+    saving. Fails if a missing batch route silently drops the run (no
+    preview at all) or raises out of the poll loop.
+    """
+    bridge_dir = prepare_bridge_dir("conv_x", bridge_id="b1", workspace=tmp_path)
+    _write_deltas_file(
+        bridge_dir,
+        [
+            {"message_id": "m1", "index": 0, "final": False, "delta": "Hello "},
+            {"message_id": "m1", "index": 1, "final": True, "delta": "world"},
+        ],
+    )
+    captured: list[_CapturedDeltaPost] = []
+    seen: dict[tuple[str, int], None] = {}
+    async with _delta_capture_client(captured, batch_supported=False) as client:
+        reset_events_batch_support(client)
+        await forwarder._forward_available_deltas(
+            client=client,
+            session_id="conv_x",
+            bridge_dir=bridge_dir,
+            state=forwarder.DeltaForwardState(),
+            seen_keys=seen,
+        )
+        # One probe, then both chunks the old way.
+        assert [c.url_path for c in captured] == [
+            "/v1/sessions/conv_x/events/batch",
+            "/v1/sessions/conv_x/events",
+            "/v1/sessions/conv_x/events",
+        ]
+        # The next poll must not re-probe: the miss is latched per client.
+        _write_deltas_file(
+            bridge_dir, [{"message_id": "m2", "index": 0, "final": True, "delta": "again"}]
+        )
+        captured.clear()
+        await forwarder._forward_available_deltas(
+            client=client,
+            session_id="conv_x",
+            bridge_dir=bridge_dir,
+            state=forwarder.DeltaForwardState(),
+            seen_keys=seen,
+        )
+    # The one new chunk went straight to the per-event route — no second probe.
+    assert [c.url_path for c in captured] == ["/v1/sessions/conv_x/events"]
+
+
+async def test_forward_available_deltas_keeps_going_when_one_batched_chunk_is_rejected(
+    tmp_path: Path,
+) -> None:
+    """
+    One rejected chunk inside a batch doesn't stop the rest of the run.
+
+    Deltas are a preview the final message supersedes, so the tail matters
+    more than any single chunk. Fails if a mid-run rejection stalls the
+    cursor (chunks re-POST forever) or raises into the poll loop.
+    """
+    bridge_dir = prepare_bridge_dir("conv_x", bridge_id="b1", workspace=tmp_path)
+    _write_deltas_file(
+        bridge_dir,
+        [
+            {"message_id": "m1", "index": 0, "final": False, "delta": "a"},
+            {"message_id": "m1", "index": 1, "final": False, "delta": "b"},
+            {"message_id": "m1", "index": 2, "final": True, "delta": "c"},
+        ],
+    )
+    captured: list[_CapturedDeltaPost] = []
+    seen: dict[tuple[str, int], None] = {}
+    async with _delta_capture_client(captured, batch_event_statuses=[202, 400, 202]) as client:
+        new_state = await forwarder._forward_available_deltas(
+            client=client,
+            session_id="conv_x",
+            bridge_dir=bridge_dir,
+            state=forwarder.DeltaForwardState(),
+            seen_keys=seen,
+        )
+    assert len(captured) == 1
     assert new_state.byte_offset == os.path.getsize(bridge_dir / "message_deltas.jsonl")
 
 

tests/test_native_post_delivery.py

@@ -12,12 +12,16 @@ from omnigent._native_post_delivery import (
     _DEAD_LETTER_BACKUP_FILE,
     _DEAD_LETTER_FILE,
     _DEAD_LETTER_MAX_BYTES,
+    MAX_EVENTS_PER_BATCH,
     RepostResult,
     _dead_letter_record_replayable,
     append_dead_letter,
+    events_batch_supported,
     post_external_session_status,
     post_may_have_been_delivered,
+    post_session_events_batch,
     replay_dead_letters,
+    reset_events_batch_support,
 )
 
 # Status codes a forwarder treats as transient/retryable, used by the replay
@@ -676,3 +680,215 @@ async def test_post_external_session_status_attaches_failure_reason() -> None:
         "output": "transcript item item-1 rejected",
     }
     assert captured[1]["data"] == {"status": "idle"}
+
+
+def _batch_client(
+    requests: list[dict[str, object]],
+    *,
+    responder: object = None,
+) -> httpx.AsyncClient:
+    """
+    Build an AsyncClient recording every batch request body.
+
+    :param requests: List appended to with each parsed request body.
+    :param responder: Optional ``(index, body) -> httpx.Response`` override,
+        for exercising route misses and partial failures. Defaults to
+        accepting every event.
+    :returns: An ``httpx.AsyncClient`` bound to the recording transport.
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        requests.append({"path": request.url.path, "body": body})
+        if responder is not None:
+            return responder(len(requests) - 1, body)  # type: ignore[operator]
+        return httpx.Response(
+            200,
+            json={
+                "results": [
+                    {"index": index, "status": 202} for index in range(len(body["events"]))
+                ],
+                "stopped_at": None,
+            },
+        )
+
+    return httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="http://ap")
+
+
+@pytest.mark.asyncio
+async def test_batch_post_sends_one_request_and_reports_each_event() -> None:
+    """
+    A run of events costs one request, with one outcome per event.
+
+    This is the round-trip saving the whole change exists for: a client
+    far from the server can otherwise only push one event per RTT. Fails
+    if the helper fans back out into per-event posts or loses the
+    per-event acknowledgement callers key their cursor off.
+    """
+    requests: list[dict[str, object]] = []
+    events = [{"type": "external_output_text_delta", "data": {"delta": str(i)}} for i in range(5)]
+    async with _batch_client(requests) as client:
+        reset_events_batch_support(client)
+        outcomes = await post_session_events_batch(client, session_id="conv_x", events=events)
+
+    assert len(requests) == 1
+    assert requests[0]["path"] == "/v1/sessions/conv_x/events/batch"
+    assert outcomes is not None
+    assert [(o.index, o.delivered) for o in outcomes] == [(i, True) for i in range(5)]
+
+
+@pytest.mark.asyncio
+async def test_batch_post_chunks_runs_over_the_cap() -> None:
+    """
+    A run longer than the cap is split, not truncated or rejected.
+
+    A resumed session can replay a large backlog in one poll; the helper
+    must ship all of it. Fails if events past the cap are silently
+    dropped (the live view would be missing a chunk of the turn).
+    """
+    requests: list[dict[str, object]] = []
+    total = MAX_EVENTS_PER_BATCH + 3
+    events = [
+        {"type": "external_output_text_delta", "data": {"delta": str(i)}} for i in range(total)
+    ]
+    async with _batch_client(requests) as client:
+        reset_events_batch_support(client)
+        outcomes = await post_session_events_batch(client, session_id="conv_x", events=events)
+
+    assert [len(r["body"]["events"]) for r in requests] == [MAX_EVENTS_PER_BATCH, 3]  # type: ignore[index]
+    assert outcomes is not None
+    assert [o.index for o in outcomes] == list(range(total))
+    assert all(o.delivered for o in outcomes)
+
+
+@pytest.mark.asyncio
+async def test_batch_post_stops_after_a_failed_chunk() -> None:
+    """
+    With ``on_error="stop"`` a failure ends the run there.
+
+    A caller advancing a durable cursor must not have later events land
+    ahead of a failed one — on retry it re-sends from the failure, and a
+    delivered successor would then be duplicated. Fails if the helper
+    keeps sending chunks past a failure, or reports unattempted events as
+    delivered.
+    """
+    requests: list[dict[str, object]] = []
+    total = MAX_EVENTS_PER_BATCH + 2
+    events = [{"type": "external_conversation_item", "data": {"i": i}} for i in range(total)]
+
+    def responder(index: int, body: dict[str, object]) -> httpx.Response:
+        """Fail the second event of the first chunk."""
+        del index
+        results = [
+            {"index": position, "status": 202}
+            for position in range(len(body["events"]))  # type: ignore[arg-type]
+        ]
+        results[1] = {"index": 1, "status": 400, "error": "bad item"}
+        return httpx.Response(200, json={"results": results[:2], "stopped_at": 1})
+
+    async with _batch_client(requests, responder=responder) as client:
+        reset_events_batch_support(client)
+        outcomes = await post_session_events_batch(client, session_id="conv_x", events=events)
+
+    # Only the first chunk was sent.
+    assert len(requests) == 1
+    assert outcomes is not None
+    assert outcomes[0].delivered is True
+    assert outcomes[1].delivered is False
+    assert outcomes[1].error == "bad item"
+    # Everything after the failure is reported unattempted, never delivered.
+    assert not any(o.delivered for o in outcomes[1:])
+    assert [o.index for o in outcomes] == list(range(total))
+
+
+@pytest.mark.asyncio
+async def test_batch_post_returns_none_on_route_miss_and_latches_it() -> None:
+    """
+    An older server's route miss tells the caller to post singly, once.
+
+    A user's CLI can be newer than the deployment it talks to. Fails if
+    the miss surfaces as an error (the forwarder would drop the run) or
+    if every later run re-probes (a wasted round trip per poll, on the
+    very link this change is trying to spare).
+    """
+    requests: list[dict[str, object]] = []
+    events = [{"type": "external_output_text_delta", "data": {"delta": "a"}}]
+
+    def responder(index: int, body: dict[str, object]) -> httpx.Response:
+        """Answer with Starlette's route-miss shape."""
+        del index, body
+        return httpx.Response(404, json={"detail": "Not Found"})
+
+    async with _batch_client(requests, responder=responder) as client:
+        reset_events_batch_support(client)
+        assert await post_session_events_batch(client, session_id="conv_x", events=events) is None
+        assert events_batch_supported(client) is False
+        assert await post_session_events_batch(client, session_id="conv_x", events=events) is None
+
+    assert len(requests) == 1
+
+
+@pytest.mark.asyncio
+async def test_batch_post_keeps_batching_when_the_session_is_missing() -> None:
+    """
+    An application 404 is not a route miss and must not latch the fallback.
+
+    ``POST /events/batch`` answers 404 both when the deployment lacks the
+    route and when the session is gone; only the first should downgrade
+    the client. Fails if a deleted session permanently costs a
+    far-away client its batching.
+    """
+    requests: list[dict[str, object]] = []
+    events = [{"type": "external_output_text_delta", "data": {"delta": "a"}}]
+
+    def responder(index: int, body: dict[str, object]) -> httpx.Response:
+        """Answer with Omnigent's application-error envelope."""
+        del index, body
+        return httpx.Response(
+            404, json={"error": {"code": "not_found", "message": "Session not found."}}
+        )
+
+    async with _batch_client(requests, responder=responder) as client:
+        reset_events_batch_support(client)
+        with pytest.raises(httpx.HTTPStatusError):
+            await post_session_events_batch(client, session_id="conv_x", events=events)
+        assert events_batch_supported(client) is True
+
+
+@pytest.mark.asyncio
+async def test_batch_post_treats_a_malformed_response_as_an_http_error() -> None:
+    """
+    A non-JSON batch reply is an error, not silent success.
+
+    Callers advance cursors on delivery; an unparseable reply proves
+    nothing about delivery, so it must reach the caller's failure path.
+    Fails if the helper swallows it and reports the run as delivered.
+    """
+    requests: list[dict[str, object]] = []
+    events = [{"type": "external_output_text_delta", "data": {"delta": "a"}}]
+
+    def responder(index: int, body: dict[str, object]) -> httpx.Response:
+        """Answer 200 with a body that is not JSON."""
+        del index, body
+        return httpx.Response(200, content=b"<html>gateway</html>")
+
+    async with _batch_client(requests, responder=responder) as client:
+        reset_events_batch_support(client)
+        with pytest.raises(httpx.HTTPError):
+            await post_session_events_batch(client, session_id="conv_x", events=events)
+
+
+@pytest.mark.asyncio
+async def test_batch_post_of_nothing_is_a_noop() -> None:
+    """
+    An empty run makes no request.
+
+    The forwarder calls this every poll; an idle poll must not cross the
+    wire at all. Fails if an empty run posts an empty envelope (which the
+    server rejects) on every poll.
+    """
+    requests: list[dict[str, object]] = []
+    async with _batch_client(requests) as client:
+        reset_events_batch_support(client)
+        assert await post_session_events_batch(client, session_id="conv_x", events=[]) == []
+    assert requests == []