How do I detect that a telephony call has been picked up? I get a participant join event for both the user and agent when the call is just ringing and not picked up

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

I need the total duration of the call. rn, I have subscribed to participant joined and left webhooks. Are these webhooks reliable and accurate, or are there better ways to handle this?

Thanks — you’re asking whether participant joined/left webhooks are reliable for measuring total call duration, and if there’s a better approach. Short answer: they work for basic use, but they are not the most accurate source for precise call timing. I’ll explain why and give recommended alternatives plus practical next steps. Why joined/left webhooks can be unreliable • Webhooks are delivered over the network and can be delayed. That means the time you receive the webhook may not match the exact server-side connect/disconnect time. • Webhooks are delivered at-least-once, so you may receive duplicates and must deduplicate by webhook id. • Reconnects, transient network issues, or participant re-joins can make start/end events ambiguous if you treat join/leave alone as the authoritative timing source. Better options (recommended) • Use server-side session timing (SessionTimer): this tracks session duration inside the LiveKit server process using monotonic clocks and avoids HTTP delivery delays. It gives more accurate duration measurements for each participant/session. • Use call summaries / call-level timestamps: enable the call summary that includes

started_at

and

ended_at

(ISO timestamps) in the summary payload. Those server-side timestamps are reliable for calculating total call duration for reporting and billing. • Combine methods: use the server-side timer or call summary as the authoritative duration, and use webhooks as notifications to trigger workflows (e.g., start billing process, update UI) rather than for calculating exact duration. Practical best practices if you keep using webhooks • De-duplicate events using the webhook

id

. • Correlate participant webhooks with the server-side session/call id (don’t rely only on display names). • Allow tolerance for network jitter when calculating durations from webhook timestamps. • If a participant reconnects during a call (transient network), merge the segments server-side to compute a single total duration. Recommended approach summary • For accurate duration: use SessionTimer or call summary timestamps as the source of truth. • For event-driven workflows: keep participant joined/left webhooks to trigger logic, but pull the authoritative duration from the call summary or session timer when you need exact numbers. If you want, I can: • Provide code examples for your SDK (JS, Python, Go, or server-side) showing how to read session timer values or extract `started_at`/`ended_at` from call summaries. • Help design a pattern for merging reconnect segments into one total duration if your environment has flaky network conditions. Which SDK or language are you using, and do you need the duration for billing/reporting or just UI display? Sources: Server-side | livekit/livekit_composite | DeepWiki | examples/survey | livekit/livekit_composite | DeepWiki | server-sdk-go | livekit/livekit_composite | DeepWiki | Agent Builder | LiveKit Documentation | Best practices for managing webhook event streams | LiveKit

so my usecase is i have tools for mid call handoff to human agent from ai agent and i need separate duaration for both these participants.

Thanks — that use case makes the need for accurate, participant-level timing clear. You want separate durations for the AI agent and the human agent across a mid-call handoff. Here’s a concise, reliable approach and practical steps you can implement. Short summary • Don’t rely only on participant joined/left webhooks for precise durations. Use server-side timing as the source of truth and use webhooks/events to trigger start/stop recording of those timers. • Track each participant’s active segment(s) and then sum segments per participant to get total duration for AI and human agents. Recommended strategy 1. Use server-side session timing as authoritative ◦ Use a server-side SessionTimer or the call summary timestamps (

started_at

/

ended_at

) to get accurate, monotonic timing. These avoid webhook delivery delays. 1. Record participant-level segments on state transitions ◦ Start a participant segment when they begin handling the call (for AI: when AI begins speaking / taking media; for human: when human answer is confirmed). ◦ End the participant segment when they stop handling the call (AI hands off OR human disconnects). ◦ Save each segment with start and end timestamps (server time). 1. Detect handoff boundaries reliably ◦ For SIP/human answer: wait for

sip.callStatus == "active"

or participant attribute change that indicates the human is on the line. ◦ For AI: mark AI active when your AI agent starts sending audio/media or explicitly when your app switches control to AI. ◦ Use

participant_attributes_changed

and

participant_connected

/

participant_disconnected

events to detect joins/attribute changes, but use the server timer timestamps for the actual times. 1. Handle reconnects and duplicates ◦ De-duplicate webhook events using the webhook

id

. ◦ If a participant reconnects, treat that as a new segment and merge segments for the same logical participant when calculating totals. 1. Compute totals ◦ Sum all segments for each participant (AI, human) to get total active duration. ◦ If you want “exclusive” durations (only one active at a time), ensure segments do not overlap or trim overlaps based on policy (e.g., give overlap to the recipient of the active media). Practical implementation outline 1. Maintain a server-side state store for the call: ◦ call_id ◦ participant records keyed by participant_id with an array of segments [{start_ts, end_ts}] ◦ current_active_party (AI or human) 1. On relevant events: ◦ participant_attributes_changed: ▪︎ If

sip.callStatus

changes to

active

for human, create human segment start (server timestamp) and close AI segment (if open). ▪︎ If

sip.callStatus

changes to `ringing`/`dialing`, update state but do not start human duration until

active

. ◦ AI handoff trigger (your app/AI decides to hand off): ▪︎ End AI segment (server ts), create human segment start once human is confirmed

active

. ▪︎ If human answer is delayed, you can mark AI paused until human

active

. ◦ participant_disconnected / left: ▪︎ Close the current participant segment with server timestamp. 1. At call end: ◦ Use session timer or call summary timestamps to validate totals and reconcile any timing drift. Edge cases & recommendations • If AI and human can be simultaneously active, define rules for attributing overlapping time (split, priority to human, or count both). • Log events and timestamps for auditing. Agent Observability / traces can help validate your computed durations. • For billing/reporting, prefer call summary or session-timer derived timestamps as the final authoritative source, and reconcile webhooks/events against those. Would you like: • A concrete code example in JS, Python, or Go showing how to implement the segment tracking and reconcile with session timestamps? • Guidance on overlap-handling policy (how to treat simultaneous AI + human activity)? Sources: SIP Lifecycle Management Agent | LiveKit Documentation | Agents telephony integration | LiveKit Documentation | Streamline troubleshooting with Agent Observability

I am using pyton and its for billing so i need very accurate data

Thanks — this is clear and important since it’s for billing. Below I’ll lay out a precise, production-ready approach in Python to get highly accurate, participant-level durations for AI and human agents during a mid-call handoff. I’ll include best practices, edge-case handling, and a concrete example you can adapt. High-level strategy (authoritative timing) • Treat server-side timing as the source of truth: use LiveKit server session timing or the call summary

started_at

/

ended_at

timestamps for authoritative start/end times. • Use events (webhooks or realtime SDK events) to mark state transitions (AI start/stop, handoff trigger, human answer) and record server timestamps when those transitions occur. • Store participant-level "segments" (start_ts, end_ts) on your server and sum segments for billing. Reconcile those summed durations against the call summary/session timer at call end to correct drift or missed events. Key decisions for billing accuracy • Use monotonic server clock timestamps when creating segment boundaries (avoid client local time). • Start human billing only when

sip.callStatus == "active"

(or the equivalent event that confirms the human is actually on the line). • Start AI billing when AI begins sending audio/media or when your app explicitly marks AI as active. • On handoff, close the current participant’s open segment using a server timestamp, then open the other participant’s segment when that participant is confirmed active. • Handle reconnects and duplicates: use webhook

id

to deduplicate, treat reconnect as a new segment, and merge segments for the same billing identity. • Reconcile: at call end, compare your summed participant durations to the call summary/session timer and resolve discrepancies (log and apply reconciliation rules). Data model (server-side) • Call record: ◦ call_id ◦ session_start (server ts from server/session timer) ◦ session_end (server ts from call summary) ◦ participants: dict keyed by participant_id containing: ▪︎ role (e.g., "ai", "human") ▪︎ segments: list of {start_ts, end_ts (nullable until closed)} ▪︎ total_seconds (computed) ◦ current_active: participant_id or None Python implementation outline 1. Central server that receives events and records segments (use Flask/FastAPI). 2. Use monotonic server timestamps (e.g., time.time() for wall clock; note this is sufficient for billing when using server authoritative timestamps — if you need monotonic for elapsed time, use time.monotonic() for internal deltas and convert to wall clock anchored at session_start). 3. Deduplicate webhooks using webhook id. 4. Reconcile at call end using call summary timestamps. Concrete Python example (conceptual, adapt to your environment) • This example uses Flask for incoming webhooks/events and an in-memory store (replace with Redis/Postgres in prod).

from flask import Flask, request, jsonify
import time
import threading

app = Flask(__name__)

# Replace with persistent DB in production
CALLS = {}
WEBHOOK_IDS = set()
LOCK = threading.Lock()

def server_ts():
    # server wall-clock timestamp in seconds (float)
    return time.time()

def start_segment(call_id, participant_id):
    with LOCK:
        call = CALLS.setdefault(call_id, {
            "participants": {},
            "session_start": None,
            "session_end": None,
            "current_active": None
        })
        p = call["participants"].setdefault(participant_id, {"segments": [], "role": None})
        # If already active, ignore
        if p["segments"] and p["segments"][-1].get("end_ts") is None:
            return
        p["segments"].append({"start_ts": server_ts(), "end_ts": None})
        call["current_active"] = participant_id

def end_segment(call_id, participant_id):
    with LOCK:
        call = CALLS.get(call_id)
        if not call:
            return
        p = call["participants"].get(participant_id)
        if not p or not p["segments"]:
            return
        if p["segments"][-1].get("end_ts") is None:
            p["segments"][-1]["end_ts"] = server_ts()
        if call.get("current_active") == participant_id:
            call["current_active"] = None

def sum_segments(p):
    total = 0.0
    for seg in p.get("segments", []):
        s = seg.get("start_ts")
        e = seg.get("end_ts") or server_ts()
        if s and e and e > s:
            total += (e - s)
    return total

@app.route("/webhook", methods=["POST"])
def webhook():
    data = request.json
    webhook_id = data.get("webhook_id")
    # dedupe
    with LOCK:
        if webhook_id in WEBHOOK_IDS:
            return jsonify({"status": "duplicate"}), 200
        WEBHOOK_IDS.add(webhook_id)

    event_type = data.get("type")
    call_id = data.get("call_id")
    participant_id = data.get("participant_id")
    # Example attributes payload might contain sip.callStatus or role
    attrs = data.get("attributes", {})

    # Example handling logic
    if event_type == "call_started":
        with LOCK:
            CALLS.setdefault(call_id, {})["session_start"] = server_ts()

    if event_type == "participant_attributes_changed":
        # If sip.callStatus becomes active => human answered
        sip_status = attrs.get("sip.callStatus")
        role = attrs.get("role")  # ensure you tag participants as 'ai' or 'human'
        with LOCK:
            call = CALLS.setdefault(call_id, {"participants": {}, "current_active": None})
            p = call["participants"].setdefault(participant_id, {"segments": [], "role": role})
            p["role"] = role

        if sip_status == "active" and role == "human":
            # Human started handling call: end AI, start human
            # find AI participant(s) and close their segments
            with LOCK:
                for pid, rec in CALLS[call_id]["participants"].items():
                    if rec.get("role") == "ai":
                        end_segment(call_id, pid)
            start_segment(call_id, participant_id)

    if event_type == "ai_active":
        # Your app signals AI starts handling (e.g., when audio out begins)
        # Close human segments and start AI
        with LOCK:
            # close any human active segments
            for pid, rec in CALLS.setdefault(call_id, {"participants": {}})["participants"].items():
                if rec.get("role") == "human":
                    end_segment(call_id, pid)
        start_segment(call_id, participant_id)

    if event_type == "participant_left":
        end_segment(call_id, participant_id)

    if event_type == "call_ended":
        with LOCK:
            CALLS.setdefault(call_id, {})["session_end"] = server_ts()
            # close all open segments
            for pid in CALLS[call_id].get("participants", {}):
                end_segment(call_id, pid)

    return jsonify({"status": "ok"}), 200

@app.route("/billing/<call_id>", methods=["GET"])
def billing(call_id):
    call = CALLS.get(call_id)
    if not call:
        return jsonify({"error": "not found"}), 404
    result = {}
    for pid, p in call.get("participants", {}).items():
        total_seconds = sum_segments(p)
        result[pid] = {
            "role": p.get("role"),
            "total_seconds": total_seconds,
            "segments": p.get("segments", [])
        }
    # Optionally reconcile with session timestamps if available
    return jsonify({"call_id": call_id, "session_start": call.get("session_start"),
                    "session_end": call.get("session_end"), "participants": result})

if __name__ == "__main__":
    app.run(port=5000)

Notes to adapt and harden for production • Persist CALLS to a durable store (Postgres/Redis). In-memory store loses data on restart. • Use server monotonic time for elapsed deltas if you are computing intervals across possible clock jumps. Anchor monotonic deltas to an initial wall-clock timestamp if you need absolute times in logs. • Reconcile sums against the call summary `started_at`/`ended_at` at call end and treat call summary as authoritative for the overall call duration. If participant totals exceed the call summary total, log and flag for audit. • Ensure every participant has a stable billing identity (e.g., your internal agent_id for AI, or human agent user id) so segments are attributable, even if the underlying LiveKit participant_id changes. • Add robust tests for the following edge cases: ◦ Immediate handoff before human answers (AI should be billed until human

active

). ◦ Human rings but doesn’t answer (no human billing). ◦ Human reconnects mid-call (merge segments). ◦ Duplicate webhooks — dedupe by webhook id. • Secure your webhook endpoints (HMAC signature verification). Final reconciliation and audit • After call end, pull the call summary (if available) and session timer values to reconcile: ◦ Validate that sum(AI + human) <= total call duration (unless overlaps are allowed). ◦ If mismatch beyond tolerance, flag for manual audit and keep logs of all state transitions and timestamps. Would you like: • A version of the example that uses Redis/Postgres for durable storage? • A stricter overlap-handling policy example (split overlap vs. priority assignment)? • Code that fetches and reconciles with LiveKit call summary / session timer

should i caclulate timing on my server cant i get this from livekit

Great question — and important since this is for billing. Short answer: LiveKit provides authoritative server-side timestamps (call summary and session timers) that you should use as the source of truth, but for participant-level billing (separate AI vs human durations across handoffs) you still need to record and attribute participant segments on your server and reconcile them with LiveKit’s timestamps. Here’s a clear breakdown of why and how to combine both. What LiveKit gives you (authoritative) • Call summary / session timestamps: LiveKit can provide server-side

started_at

and

ended_at

timestamps (and telemetry session timers) that reflect the actual call/session start and end on the server. These timestamps are accurate and avoid webhook delivery delays. • Participant attribute changes: LiveKit emits events (webhooks or realtime events) like

participant_attributes_changed

including

sip.callStatus

, which tells you when a SIP participant goes

active

(answered) vs

ringing

. These are useful for detecting state transitions. Why you should calculate participant timing on your server • Billing needs participant-level attribution (AI vs human) and may require merging multiple segments per participant across handoffs or reconnects — LiveKit does not provide a pre-aggregated per-participant billing report tailored to your business rules. • Your server can: ◦ Record exact server timestamps when state transitions happen (use the event receipt time but prefer server-side monotonic timestamps when computing intervals). ◦ Attribute segments to your billing identities (AI agent id, human agent id). ◦ Handle deduplication, merging reconnect segments, and applying your overlap policy (e.g., split overlap, priority to human, or count both). ◦ Reconcile against LiveKit’s call-level session timer or call summary for final validation and auditing. • Relying only on webhook join/leave times is brittle due to network latency, duplicates, and reconnects. Recommended pattern (production-ready) 1. Use LiveKit server timestamps as authoritative for overall call duration. ◦ At call end, fetch or accept the call summary / session timer timestamps and treat these as the canonical call start/end. 1. Record participant segments on your server using events as triggers. ◦ Start a segment when your app determines a participant becomes active: ▪︎ AI: when AI starts sending audio or you mark AI active. ▪︎ Human: only when

sip.callStatus == "active"

or you receive a reliable confirmation the human is answered. ◦ End a segment when the participant stops handling the call (handoff, disconnect, or end of call). ◦ Store segments with server-side timestamps in durable storage (Postgres/Redis). 1. Deduplicate and handle reconnects. ◦ Deduplicate webhook events by webhook id. ◦ Treat reconnects as new segments and merge segments for the same billing identity. 1. Reconcile at call end. ◦ Compare sum of participant segments to LiveKit call summary/session duration. ◦ If there's drift beyond tolerance, log and flag for audit (and optionally adjust according to your business rules). 1. Apply billing policy for overlaps. ◦ Decide how to handle simultaneous activity (split time, give priority, etc.) and implement that logic server-side. Why this gives you the best accuracy • LiveKit timestamps ensure your total call duration is accurate. • Server-side segment tracking lets you implement business rules and reliably attribute time to AI vs human. • Reconciliation ensures discrepancies are caught and auditable. Practical next steps (I can help) • If you want, I can provide: ◦ A hardened Python example that persists segments to Redis/Postgres and reconciles with LiveKit call summary. ◦ A tested overlap-handling policy implementation (split vs priority). ◦ Webhook deduplication and HMAC verification snippets for security. ◦ Guidance on using monotonic timers vs wall-clock timestamps for robust elapsed-time computation. Which would you like next: the durable-storage Python example, overlap policy code, or the reconciliation flow that pulls the LiveKit call summary?

can you like tell me the flow. imma getting confused should i remove webhook or what. could you explain like a step by step guide what to setup

Thanks — I’ll make this simple and step‑by‑step so you can implement it without confusion. I’ll assume you’re using Python and need highly accurate billing-level durations split between AI and human agents during a mid‑call handoff. Overview (one sentence) • Keep webhooks/events as triggers, but calculate and store all participant-level timing on your server and reconcile with LiveKit’s call summary/session timer at the end. Don’t remove webhooks — use them correctly. Step‑by‑step setup and flow 1. Prepare infrastructure ◦ Endpoint: Create a secure webhook endpoint (FastAPI/Flask) to receive LiveKit events. ◦ Durable storage: Use Postgres or Redis to persist call records and participant segments (do not use in-memory storage). ◦ Security: Verify webhook signatures (HMAC) and deduplicate by webhook id. 1. Data model (what to store) ◦ Call table/record: ▪︎ call_id (LiveKit call or room id) ▪︎ session_start (server ts when call_started seen or call summary) ▪︎ session_end (server ts from call summary) ◦ Participant record (per call): ▪︎ participant_id (stable billing id for AI/human) ▪︎ role ("ai" or "human") ▪︎ segments: list of {start_ts, end_ts} ▪︎ total_seconds (computed) ◦ Webhook dedupe table: webhook_id -> processed_flag 1. Which events to keep and how to use them ◦ Keep these events: ▪︎ participant_attributes_changed (for

sip.callStatus

) — use to detect human answered (

sip.callStatus == "active"

). ▪︎ your app’s AI state events (e.g.,

ai_active

,

ai_inactive

) — instrument your AI service to call your webhook when AI starts/stops speaking. ▪︎ participant_joined / participant_left — keep for additional context, but don’t rely on join/leave timestamps for billing accuracy. ▪︎ call_started / call_ended or call summary webhook — use call summary’s

started_at

and

ended_at

as authoritative for total call duration. ◦ Remove? Don’t remove webhooks. You need them. Just don’t use join/leave times alone as final billing timestamps. 1. Event handling logic (server-side) ◦ Deduplicate incoming webhook (skip if webhook_id already processed). ◦ Convert event arrival to a server timestamp immediately (use server time). ◦ On

call_started

or first relevant event: ▪︎ Create call record and set session_start if first event; otherwise wait for call summary. ◦ On AI start (

ai_active

event from your AI system): ▪︎ Ensure AI participant record exists with role="ai". ▪︎ Close any open participant segment for human (end_segment). ▪︎ Start a new AI segment: append {start_ts = server_ts(), end_ts = null}. ◦ On AI stop / handoff request (your app triggers when handing off): ▪︎ Close AI current open segment with end_ts = server_ts(). ▪︎ Wait for human to actually answer; do not start human billing until human confirmed active. ◦ On

participant_attributes_changed

with

sip.callStatus == "active"

for human: ▪︎ Ensure human participant record exists with role="human". ▪︎ Close open AI segment if any (end_ts = server_ts()). ▪︎ Start human segment: append {start_ts = server_ts(), end_ts = null}. ◦ On

participant_left

or human hangup: ▪︎ Close current human segment with end_ts = server_ts(). ◦ On

call_ended

or call summary webhook: ▪︎ Close any remaining open segments (set end_ts = server_ts()). ▪︎ Set session_end from call summary

ended_at

if provided (treat as authoritative for total call duration). 1. Reconciliation and final billing calculation ◦ After call end, compute participant totals: ▪︎ For each participant, sum (end_ts - start_ts) across segments. ◦ Reconcile totals with LiveKit call summary: ▪︎ If sum(AI + human) > total call duration (from session timer/call summary) beyond a small tolerance, flag for audit. ▪︎ Adjust according to your policy (e.g., trim overlap, log discrepancy). ◦ Persist final billed durations and keep full event/segment logs for audits. 1. Edge cases & rules you must decide ◦ Overlap policy: Should overlapping AI/human time be split, assigned to human, or billed to both? Implement in reconciliation. ◦ Handoff before human answers: AI should be billed until human

sip.callStatus == "active"

. If human never answers, AI gets full time. ◦ Reconnects: Treat reconnect as new segment; merge by billing identity. ◦ Duplicate webhooks: Deduplicate by webhook id before processing. 1. Implementation tips for accuracy ◦ Use server wall-clock timestamps (time.time()) when storing start/end; for elapsed monotonic deltas use time.monotonic() to compute durations if you need to avoid clock jumps — but anchor to wall-clock for auditing. ◦ Always prefer call summary/session timer values for overall call start/end timestamps. ◦ Keep detailed logs of every event and segment for post-mortem/billing disputes. Minimal example flow timeline (AI → handoff → human) • t0: AI answers and begins speaking → your AI system sends

ai_active

-> server starts AI segment at server_ts t0. • t1: AI triggers handoff -> your AI sends

ai_inactive

or

handoff_requested

-> server closes AI segment at t1. • t2: Human phone rings (LiveKit emits

participant_attributes_changed

with

sip.callStatus == "ringing

) -> no billing change. • t3: Human answers (LiveKit emits

participant_attributes_changed

with

sip.callStatus == "active

) -> server starts human segment at t3. • t4: Human hangs up -> `participant_left`/`sip.callStatus == "hangup"` -> server closes human segment at t4. • call ended: server reconciles with call summary `started_at`/`ended_at` and computes final totals: AI = t1 - t0, Human = t4 - t3. Do you want a concrete Python template now? • I can provide a hardened FastAPI + Redis/Postgres example that: ◦ Verifies webhooks (HMAC) ◦ Deduplicates webhooks ◦ Stores segments durably ◦ Reconciles with call summary ◦ Implements a chosen overlap policy (split or human-priority) Which one would you prefer: Redis or Postgres for storage, and do you want overlap splitting or human-priority?

i dont knnow why you think human and ai spit is an issue. that i am doing based on participant joined and left. i have metadata attached to participant to detect type. my issue is call state. i only want real call duration not ringing and all

Thanks — got it. You already distinguish AI vs human by participant metadata and record join/leave. Your remaining problem is excluding ringing time and only counting the real, answered call time. Below is a short, clear step‑by‑step flow you can implement in Python. Keep webhooks — but don’t use join/leave alone for timing. Use

sip.callStatus

(or equivalent server-side answer confirmation) as the authoritative signal that the call is active. Step‑by‑step flow to get real call duration (exclude ringing) 1. Keep these webhooks/events ◦

participant_attributes_changed

(to read

sip.callStatus

) ◦

participant_joined

/

participant_left

(keep for context/identities) ◦

call_ended

/ call summary (use

started_at

/

ended_at

as authoritative overall session timestamps) 1. What marks the real call start? ◦ Do NOT start billing at

participant_joined

. The SIP participant is created during dialing/ringing. ◦ Start the participant’s active segment only when you observe

sip.callStatus == "active"

for that participant. That indicates the far end answered. 1. What marks the real call end? ◦ End the participant’s active segment when you see: ▪︎

sip.callStatus == "hangup"

OR ▪︎

participant_left

/

participant_disconnected

for that SIP participant. ◦ Also close any open segments on

call_ended

(use call summary

ended_at

as authoritative). 1. Server timing (authoritative) ◦ When an event arrives, record a server timestamp immediately (time.time()) and use that for segment start/end. ◦ Don’t use the webhook's delivery time as proof of state — use the server timestamp when processing the event. 1. Reconciliation ◦ At call end, reconcile participant segments with the call summary/session timer: ▪︎ Use LiveKit call summary `started_at`/`ended_at` as the authoritative overall start/end. ▪︎ If sum(participant active durations) and call summary differ beyond tolerance, log and flag for audit. Minimal Python handling logic (conceptual) • On `participant_attributes_changed`: ◦ If

sip.callStatus

changed to `"active"`: ▪︎ Start active segment for that participant (server_ts). ◦ If

sip.callStatus

changed to

"ringing"

or `"dialing"`: ▪︎ Do not start billing; optionally note state for UI. ◦ If

sip.callStatus

changed to `"hangup"`: ▪︎ End any open active segment for that participant (server_ts). • On `participant_left`: ◦ End open segment for that participant if exists (server_ts). • On

call_ended

/ call summary: ◦ Close all open segments and reconcile with

ended_at

. Example event timeline (AI stays until human answers) • t0: AI starts (your AI system marks AI active) → start AI active segment at server_ts t0. • t1: AI triggers handoff and stops → end AI segment at t1. • t2: LiveKit emits

sip.callStatus == "ringing"

for human → no billing change. • t3: LiveKit emits

sip.callStatus == "active"

for human → start human segment at t3. • t4: Human hangs up → end human segment at t4. • Final billed durations: AI = t1 − t0, Human = t4 − t3. Edge cases to handle (brief) • Human never answers:

sip.callStatus

may stay

ringing

→ AI gets billed until call end. • Duplicate webhooks: dedupe by webhook id. • Reconnects: treat reconnect as new segment; merge segments per billing identity. • Overlap policy: you already handle AI/human split; ensure you only count

active

periods. Quick checklist to implement now • [ ] Ensure

participant_attributes_changed

webhook is enabled and includes

sip.callStatus

. • [ ] On that webhook, use server timestamp to start/stop segments when status becomes `active`/`hangup`. • [ ] Persist segments (Postgres/Redis) keyed by call_id + participant billing id. • [ ] On

call_ended

, reconcile segments with call summary `started_at`/`ended_at`. • [ ] Deduplicate webhooks and log everything for audits. If you want, I’ll: • Provide a compact, production-ready FastAPI + Postgres snippet that shows the exact webhook handlers for

participant_attributes_changed

and

call_ended

and how to store segments. • Or just give the concrete code for handling

sip.callStatus

events in your existing webhook handler. Which would you prefer?

• Provide a compact, production-ready FastAPI + Postgres snippet that shows the exact webhook handlers for

participant_attributes_changed

and

call_ended

and how to store segments. •

Thanks — I’ll give you a compact, production-ready FastAPI + PostgreSQL example that focuses only on what you asked for: accurate billing-grade durations that exclude ringing. This implementation: • Uses

participant_attributes_changed

to watch

sip.callStatus

. • Starts timing when

sip.callStatus == "active"

. • Stops timing on

sip.callStatus == "hangup"

or participant/room end. • Persists segments in Postgres and supports idempotent webhook processing. • Uses server-side timestamps and deduplicates by webhook id. Notes before you paste into production • Replace DATABASE_URL, webhook signature verification, and any lookup stubs with your production values. • Use a queue/worker (Celery/RQ) in production for processing; here we show synchronous processing for clarity. • Use migrations (Alembic) for schema changes; this snippet creates tables for simplicity. • Secure the webhook endpoint (HMAC) and rate-limit it. Code (FastAPI + SQLAlchemy + Postgres)

# filename: app.py
from fastapi import FastAPI, Request, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Float, Boolean, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from datetime import datetime, timezone
import os
import uuid
import hmac
import hashlib

DATABASE_URL = os.getenv("DATABASE_URL", "<postgresql://user:pass@localhost/livekit_billing>")
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "replace_with_your_secret")

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()

# --- DB models --- #
class ProcessedWebhook(Base):
    __tablename__ = "processed_webhooks"
    id = Column(String, primary_key=True, index=True)  # webhook id (UUID string)
    event_type = Column(String, nullable=False)
    received_at = Column(DateTime(timezone=True), nullable=False)
    raw = Column(String, nullable=True)

class CallSegment(Base):
    __tablename__ = "call_segments"
    id = Column(Integer, primary_key=True, index=True)
    room_id = Column(String, index=True, nullable=False)
    participant_sid = Column(String, index=True, nullable=False)
    participant_identity = Column(String, nullable=True)  # optional
    role = Column(String, nullable=True)  # 'ai' or 'human' as you annotate
    started_at = Column(DateTime(timezone=True), nullable=False)
    ended_at = Column(DateTime(timezone=True), nullable=True)  # null if open
    duration_seconds = Column(Float, nullable=True)  # null if open
    finalized = Column(Boolean, default=False, index=True)

# helpful index: quickly find latest open segment by room+participant
Index("ix_call_segments_room_participant_open", CallSegment.room_id, CallSegment.participant_sid, CallSegment.finalized)

Base.metadata.create_all(bind=engine)

# --- FastAPI app --- #
app = FastAPI()

# Utility / DI
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def server_ts() -> datetime:
    return datetime.now(timezone.utc)

def verify_signature(secret: str, body: bytes, header_signature: str) -> bool:
    # Example: header_signature expected as hex HMAC-SHA256
    mac = hmac.new(secret.encode(), msg=body, digestmod=hashlib.sha256)
    expected = mac.hexdigest()
    return hmac.compare_digest(expected, header_signature)

# --- Incoming webhook payload model (simplified) --- #
class WebhookPayload(BaseModel):
    id: str
    event: str
    room: dict | None = None
    participant: dict | None = None
    changed_attributes: dict | None = None
    # other fields may exist; ignore extras

# --- Helper functions for segments --- #
def start_segment(db: Session, room_id: str, participant_sid: str, participant_identity: str | None, role: str | None):
    # ensure there's no open segment for this participant in this room
    open_seg = db.query(CallSegment).filter_by(
        room_id=room_id, participant_sid=participant_sid, finalized=False
    ).order_by(CallSegment.started_at.desc()).first()
    if open_seg:
        return open_seg  # already started

    seg = CallSegment(
        room_id=room_id,
        participant_sid=participant_sid,
        participant_identity=participant_identity,
        role=role,
        started_at=server_ts(),
        finalized=False
    )
    db.add(seg)
    db.commit()
    db.refresh(seg)
    return seg

def end_segment(db: Session, room_id: str, participant_sid: str):
    open_seg = db.query(CallSegment).filter_by(
        room_id=room_id, participant_sid=participant_sid, finalized=False
    ).order_by(CallSegment.started_at.desc()).first()
    if not open_seg:
        return None

    now = server_ts()
    open_seg.ended_at = now
    open_seg.duration_seconds = (now - open_seg.started_at).total_seconds()
    open_seg.finalized = True
    db.commit()
    db.refresh(open_seg)
    return open_seg

def finalize_all_for_room(db: Session, room_id: str):
    now = server_ts()
    open_segs = db.query(CallSegment).filter_by(room_id=room_id, finalized=False).all()
    for seg in open_segs:
        seg.ended_at = now
        seg.duration_seconds = (now - seg.started_at).total_seconds()
        seg.finalized = True
    db.commit()
    return open_segs

# --- Webhook endpoint --- #
@app.post("/webhook")
async def webhook(request: Request, db: Session = Depends(get_db)):
    body = await request.body()
    signature = request.headers.get("x-livekit-signature", "")  # adapt header name to your config

    # verify signature (implement exact header & method your LiveKit config uses)
    if not verify_signature(WEBHOOK_SECRET, body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    try:
        payload = WebhookPayload.parse_raw(body)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid payload")

    webhook_id = payload.id
    # Idempotency: skip duplicates
    exists = db.query(ProcessedWebhook).get(webhook_id)
    if exists:
        return {"status": "duplicate"}

    # persist processed webhook id
    processed = ProcessedWebhook(
        id=webhook_id,
        event_type=payload.event,
        received_at=server_ts(),
        raw=body.decode("utf-8")
    )
    db.add(processed)
    db.commit()

    # Process important events synchronously here (for simplicity).
    # In production push payload to task queue and process reliably in background worker.

    room_id = None
    if payload.room:
        # room may give sid or name depending on your LiveKit config
        room_id = payload.room.get("sid") or payload.room.get("id") or payload.room.get("name")

    # Extract participant identifiers
    participant = payload.participant or {}
    participant_sid = participant.get("sid")
    participant_identity = participant.get("identity")
    # Your system sets metadata to detect role (ai/human). We will attempt to read it:
    role = None
    attrs = payload.changed_attributes or {}
    # If you attach role as a participant attribute, check there as fallback:
    role = attrs.get("role") or participant.get("metadata") or participant.get("role")

    # 1) participant_attributes_changed -> watch sip.callStatus
    if payload.event == "participant_attributes_changed":
        sip_status = attrs.get("sip.callStatus")
        # ignore if no sip status change
        if sip_status:
            # If status becomes active -> start segment
            if sip_status == "active":
                # Only start when LiveKit says active (exclude ringing/dialing)
                # Ensure we have participant_sid; if not, you may need to map identity -> sid from your join handling
                if participant_sid and room_id:
                    start_segment(db, room_id, participant_sid, participant_identity, role)
            # If status becomes hangup or anything other than active -> end segment if open
            elif sip_status in ("hangup", "failed", "busy", "cancel"):
                if participant_sid and room_id:
                    end_segment(db, room_id, participant_sid)
            else:
                # For "ringing", "dialing", etc. do not start timing
                pass

    # 2) treat participant_left / room_finished / egress_ended / call_ended as finalizers
    if payload.event in ("participant_left", "egress_ended", "room_finished", "call_ended"):
        # If participant_left and we have a participant_sid, finalize that participant only
        if payload.event == "participant_left" and participant_sid and room_id:
            end_segment(db, room_id, participant_sid)
        else:
            # finalize all open segments for the room
            if room_id:
                finalize_all_for_room(db, room_id)

    # respond quickly (webhook accepted)
    return {"status": "accepted"}

What this gives you • Accurate participant active time: starts only at

sip.callStatus == "active"

, so ringing/dialing time is excluded. • Durable segments saved in Postgres for billing Sources: Best practices for managing webhook event streams | LiveKit | Agents telephony integration | LiveKit Documentation