@session.on("close")
def _on_close(ev):
    session_history = session.history.to_dict()
    <http://logger.info|logger.info>(f"Session history is this: {session_history}")
    # Send transcript to Kafka using the new method
    asyncio.create_task(send_session_transcript_to_kafka(
        session_history=session_history,
        chat_id=agent_instance.chat_id,
        audio_session_id=agent_instance.audio_session_id
    ))

how do I know if the order of messages received in this session_history is correct or not

Thanks for reaching out, Gaurav. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

this does not seem like it is in chronological order: {"message": "Session history is this: {'items': [{'id': 'item_ded68d0ba9e0', 'type': 'agent_handoff', 'new_agent_id': 'voice_assistant'}, {'id': 'GR_2bf01204c7fa', 'type': 'message', 'role': 'assistant', 'content': ['Hello! I am Aristo, your OneAssist assistant. How may I help you today?'], 'interrupted': False, 'extra': {}, 'metrics': {'started_speaking_at': 1766076535.6485136, 'stopped_speaking_at': 1766076539.9923906}}, {'id': 'item_b50772388b7f', 'type': 'function_call', 'call_id': 'function-call-11828173556835006742', 'arguments': '{\"detected_language\": \"English\", \"message\": \"Can you tell me what is my claim status?\"}', 'name': 'oneassist_chat_tool'}, {'id': 'GI_8da66bd39958', 'type': 'message', 'role': 'user', 'content': ['Can you tell me what is my claim status?'], 'interrupted': False, 'extra': {}, 'metrics': {}}, {'id': 'item_24fd4aeace12', 'type': 'function_call_output', 'name': 'oneassist_chat_tool', 'call_id': 'function-call-11828173556835006742', 'output': 'We currently do not have any active service request or claim associated with the mobile number you are logged in from. Please log in with the mobile number registered under your OneAssist plan to check your claim or service request status.', 'is_error': False}, {'id': 'GR_12840923490e', 'type': 'message', 'role': 'assistant', 'content': ['We currently do not have any active service request or claim associated with the mobile number you are logged in from. Please log in with the mobile number registered under your OneAssist plan to check your claim or service request status.'], 'interrupted': False, 'extra': {}, 'metrics': {'started_speaking_at': 1766076552.3906772, 'stopped_speaking_at': 1766076564.7274282}}, {'id': 'GR_aaea3bcfcba5', 'type': 'message', 'role': 'assistant', 'content': ['Maaf kijiye, main sirf OneAssist se sambandhit prashnon mein aapki sahayata kar sakti hoon. Mausam ke liye, aap kripya kisi mausam app ya website ka upayog karein.'], 'interrupted': False, 'extra': {}, 'metrics': {'started_speaking_at': 1766076572.8678753, 'stopped_speaking_at': 1766076581.7524781}}, {'id': 'GI_030e27dc9a65', 'type': 'message', 'role': 'user', 'content': ['आज का मौसम कैसा है'], 'interrupted': False, 'extra': {}, 'metrics': {}}, {'id': 'GR_9c0a42e69111', 'type': 'message', 'role': 'assistant', 'content': ['Estoy bien, gracias. ¿En qué puedo ayudarte con OneAssist?'], 'interrupted': False, 'extra': {}, 'metrics': {'started_speaking_at': 1766076592.3932989, 'stopped_speaking_at': 1766076596.1622906}}, {'id': 'GI_4ce1a85d15c4', 'type': 'message', 'role': 'user', 'content': ['¿Cómo estás?'], 'interrupted': False, 'extra': {}, 'metrics': {}}, {'id': 'item_aa6886ce8d87', 'type': 'function_call', 'call_id': 'function-call-17664241729123424253', 'arguments': '{\"message\": \"Do I get unlimited cleans when I buy any plants in one assist\", \"detected_language\": \"English\"}', 'name': 'oneassist_chat_tool'}, {'id': 'GI_d8f38fe5765c', 'type': 'message', 'role': 'user', 'content': ['Do I get unlimited cleans when I buy any plants in one assist'], 'interrupted': False, 'extra': {}, 'metrics': {}}, {'id': 'item_c190b9552cde', 'type': 'function_call_output', 'name': 'oneassist_chat_tool', 'call_id': 'function-call-17664241729123424253', 'output': 'OneAssist currently offers a range of plans for home appliances, personal electronics, and financial protection, but it does not provide any specific service related to \"unlimited cleans\" or cleaning services for plants. \\n\\nIf you have purchased a plan for a home appliance or another product, please log in with the registered mobile number and check the status of your service request. If you are referring to a different type of service or product, kindly clarify so that I can assist you better.', 'is_error': False}, {'id': 'GR_5e7190de0e49', 'type': 'message', 'role': 'assistant', 'content': ['OneAssist currently offers a range of plans for home appliances, personal electronics, and financial protection, but it does not provide any specific service related to \"unlimited cleans\" or cleaning services for plants. If you have purchased a plan for a home appliance or another product, please log in with the registered mobile number and check the status of your service request. If you are referring to a different type of service or product, kindly clarify so that I can assist you better.'], 'interrupted': False, 'extra': {}, 'metrics': {'started_speaking_at': 1766076703.6234858, 'stopped_speaking_at': 1766076729.700411}}]}", "level": "INFO", "name": "__mp_main__", "pid": 634, "job_id": "AJ_SbuGZRBBHsei", "room_id": "RM_numoDNxkqW7J", "timestamp": "2025-12-18T165215.626278+00:00"}

Thanks — I see the sample

session_history

and your concern that items look out of sequence. Here’s a concise way to verify and defend ordering. 1) Confirm by timestamps • Each item should have a timestamp (e.g.,

started_speaking_at

,

stopped_speaking_at

, or a `created_at`/`timestamp` field). Compare those timestamps to determine chronological order. 2) Quick script to check ordering

entries = session_history.get("items", [])
# pick a timestamp field that exists on most items, fallback to None
def ts(e):
    return e.get("metrics", {}).get("started_speaking_at") or e.get("timestamp") or e.get("created_at")

timestamps = [ts(e) for e in entries]
is_sorted = all(timestamps[i] <= timestamps[i+1] for i in range(len(timestamps)-1))
out_of_order = [ (i, entries[i], timestamps[i]) for i in range(len(timestamps)-1) if timestamps[i] > timestamps[i+1] ]

<http://logger.info|logger.info>("history sorted: %s, out_of_order_count: %d", is_sorted, len(out_of_order))
if out_of_order:
    logger.warning("out of order samples: %s", out_of_order[:5])

3) Why it may appear unordered • Different message types use different timestamp fields (assistant messages include

metrics.started_speaking_at

, user messages may not), so if you inspect the raw list without normalizing timestamps it can look shuffled. • Function-call items may be inserted between user/assistant messages and can make the flow look interleaved even though timestamps are correct. 4) Recommended fix before sending to Kafka • Normalize each item to a single

timestamp

(choose

metrics.started_speaking_at

,

created_at

, or event time), sort by that timestamp, then send.

for e in entries:
    e["_ts"] = ts(e) or 0
entries_sorted = sorted(entries, key=lambda e: e["_ts"])

If you want, share a small slice of the actual entries (including the metrics/timestamp fields) and I’ll point out any out-of-order items and the appropriate field to use for sorting.

this is not chronological and neither is there any timestamp on user message. how to do that.

You’re right to be concerned — when user items have no timestamp you can’t rely on an explicit time field. Here’s a clear, reliable approach to guarantee chronological ordering before sending to Kafka. What to do (short) 1. Trust the session history order as the source of truth — items in

session.history.items

are appended in processing order. 2. Add or normalize timestamps as items are added, so every item has a single

timestamp

you can sort on downstream. 3. Sort by that normalized

timestamp

before sending to Kafka (or just use the list order if you trust it and don’t need explicit times). How to implement it • Use the conversation-item event to inject timestamps when items are committed:

import time

@session.on("conversation_item_added")
def on_item_added(ev):
    item = ev.item
    # set a normalized timestamp (epoch seconds)
    item.extra = item.extra or {}
    item.extra["created_at"] = item.extra.get("created_at") or time.time()

• When serializing and sending:

session_history = session.history.to_dict()
items = session_history.get("items", [])

# ensure every item has a timestamp fallback
def get_ts(e):
    # prefer explicit fields if present, then injected created_at
    return (
        e.get("metrics", {}).get("started_speaking_at")
        or e.get("created_at")
        or e.get("timestamp")
        or e.get("extra", {}).get("created_at")
        or 0
    )

for e in items:
    e["_ts"] = get_ts(e)

# sort by normalized timestamp; this also preserves original order for equal timestamps
items_sorted = sorted(items, key=lambda x: x["_ts"])

# send items_sorted to Kafka

Notes and recommendations • If you only need ordering (not absolute times), you can skip sorting and send

session.history.items

in its existing order — it reflects the processing sequence. • Injecting

created_at

via the event ensures user messages (and other types) always carry a timestamp for downstream consumers. • If you share a small sample of raw items (including `extra`/`metrics`), I can point to the best field to use as

timestamp

or provide a ready-to-run wrapper you can drop in. Would you like that? Sources: Observability data hooks | LiveKit docs | Events and error handling | LiveKit docs | Text and transcriptions | LiveKit docs

this is what I am using:

session = AgentSession(
    llm=google.realtime.RealtimeModel(
        model="gemini-2.5-flash-native-audio-preview-12-2025",
        voice="Aoede",
        temperature=0.8,
         )
    ),
    preemptive_generation=True,
    user_away_timeout=30.0,
)

Thanks — since you’re creating the session with

AgentSession(...)

and realtime LLM, the best approach is to inject timestamps as items are committed and then normalize/sort before sending to Kafka. Suggested minimal changes 1) Attach a

conversation_item_added

handler that stamps every item when it’s added:

import time

@session.on("conversation_item_added")
def on_item_added(ev):
    item = ev.item
    # ensure extra exists and store epoch seconds
    item.extra = item.extra or {}
    item.extra.setdefault("created_at", time.time())

2) When you serialize and send, normalize a single timestamp and sort (this preserves original order for ties):

session_history = session.history.to_dict()
items = session_history.get("items", [])

def get_ts(e):
    return (
        e.get("metrics", {}).get("started_speaking_at")
        or e.get("created_at")
        or e.get("timestamp")
        or e.get("extra", {}).get("created_at")
        or 0
    )

for e in items:
    e["_ts"] = get_ts(e)

items_sorted = sorted(items, key=lambda x: x["_ts"])
# send items_sorted to Kafka

Why this works • The session history list already reflects processing order, but user items often lack native timestamps. • Injecting

extra["created_at"]

at commit time gives every item a reliable epoch timestamp you can sort and/or expose downstream. If you want, I can provide a small wrapper that: • Adds the handler, • Produces the final Kafka payload, • Marks any items missing text/timestamps for inspection. Would you like that wrapper?