Session Event Batch Encoder (omnigent/session_event_batch.py)

Serializes and validates byte-bounded session event batches for HTTP transmission. Useful for efficient telemetry and event dispatching.

# Source: https://github.com/omnigent-ai/omnigent | License: Apache-2.0 | Copyright You must retain, in the Source form of any Derivative Works
# Provided via justclaw.it.com; keep this notice with the code.
"""Wire contract for byte-bounded session event batches."""

from __future__ import annotations

import json
from collections.abc import Sequence
from typing import Any

# Limit the encoded JSON request body itself, not just the sum of event text.
# The count bound also keeps validation and dispatch work predictable.
MAX_SESSION_EVENT_REQUEST_BYTES = 10 * 1024 * 1024
MAX_SESSION_EVENT_BATCH_EVENTS = 100


def encode_session_event_batch(events: Sequence[dict[str, Any]]) -> bytes:
    """Encode an event array exactly as it will be sent over HTTP."""
    return json.dumps(
        list(ev

... (truncated -- full source via MCP)