Call handler

from standin import CallServer, CallSession


class MyHandler:
    async def on_start(self, session: CallSession) -> None: ...
    async def on_caller_audio(self, pcm: bytes) -> None: ...
    async def on_context(self, text: str) -> None: ...
    async def on_goodbye(self, text: str) -> None: ...
    async def aclose(self, reason: str) -> None: ...


server = CallServer(handler_factory=MyHandler)

class SessionStart:
    call_id: str
    thread_id: str
    caller: Caller
    direction: str = "inbound"          # "inbound" or "outbound"
    recording_status: str | None = None
    tenant_id: str | None = None


class Caller:
    aad_id: str | None = None
    display_name: str | None = None
    tenant_id: str | None = None

async def on_agent_track_published(self) -> None:
    self._call.mark_answered()   # the agent's OWN track, not a participant joining

async def on_start(self, session: CallSession) -> None:
    self._call = session
    self._client = await open_realtime_socket()

async def on_start(self, session: CallSession) -> None:
    caller = session.start.caller
    if caller.aad_id not in self.allowlist:
        await session.end("caller-not-allowed")
        return
    self._call = session

async def on_caller_audio(self, pcm: bytes) -> None:
    await self._client.append_input(pcm)

async def on_context(self, text: str) -> None:
    await self._agent.add_system_note(text)

async def on_goodbye(self, text: str) -> None:
    await self._client.interrupt_and_say(text)

async def aclose(self, reason: str) -> None:
    await self._client.close()

async def on_speech_started(self) -> None:
    await self._call.cancel_playback()   # first: silence what is already queued
    await self._client.cancel_response()  # then: stop the model generating more
    self._aligner.reset()                 # and drop the partial frame of the dead turn

import asyncio

from standin import (
    REALTIME_SAMPLE_RATE_HZ,
    SAMPLE_RATE_HZ,
    CallServer,
    CallSession,
    FrameAligner,
    resample_pcm16,
)


class RealtimeHandler:
    """One Microsoft Teams call, bridged to one realtime model session."""

    def __init__(self, api_key: str, instructions: str) -> None:
        # One handler per call, so per-call state is safe on self.
        self._api_key = api_key
        self._instructions = instructions
        self._call: CallSession | None = None
        self._client: RealtimeClient | None = None
        # The model speaks 24 kHz and chunks where it likes; the wire wants
        # whole 640-byte frames at 16 kHz.
        self._aligner = FrameAligner()

    async def on_start(self, session: CallSession) -> None:
        caller = session.start.caller
        # No leading truthiness test on aad_id. Microsoft Teams reports no AAD object id
        # for guest and anonymous callers, so `if caller.aad_id and ...` would
        # fail OPEN for exactly the callers you least want to admit.
        if caller.aad_id not in ALLOWED:
            # Safe from inside on_start: it asks for the close and returns.
            await session.end("caller-not-allowed")
            return

        self._call = session
        self._client = await RealtimeClient.connect(
            api_key=self._api_key,
            instructions=(
                f"{self._instructions} You are on a Microsoft Teams call "
                f"with {caller.display_name or 'a caller'}."
            ),
            on_audio=self._on_model_audio,
            on_speech_started=self._on_barge_in,
            on_turn_done=self._on_turn_done,
        )

    async def on_caller_audio(self, pcm: bytes) -> None:
        # Receive path: hand it over and return. Never block here.
        client = self._client
        if client is not None:
            await client.append_input(
                resample_pcm16(pcm, SAMPLE_RATE_HZ, REALTIME_SAMPLE_RATE_HZ)
            )

    async def on_context(self, text: str) -> None:
        client = self._client
        if client is not None:
            # Non-interrupting: it informs the next turn, it does not start one.
            await client.add_context(text)

    async def on_goodbye(self, text: str) -> None:
        client = self._client
        if client is not None:
            # Cancel first, then speak: a plain say is swallowed by a provider
            # that guards on "is a response already active", and this line
            # almost always arrives mid-answer.
            await client.interrupt_and_say(text)

    async def aclose(self, reason: str) -> None:
        client, self._client = self._client, None
        if client is not None:
            await client.close()

    # ---- provider callbacks ------------------------------------------------

    async def _on_model_audio(self, pcm_24k: bytes) -> None:
        call = self._call
        if call is None:
            return
        pcm = resample_pcm16(pcm_24k, REALTIME_SAMPLE_RATE_HZ, SAMPLE_RATE_HZ)
        # Whole frames only; the remainder is carried, not dropped. Dropping it
        # clips a few milliseconds off every chunk seam, which is audible.
        for frame in self._aligner.push(pcm):
            await call.send_audio(frame)

    async def _on_turn_done(self) -> None:
        call = self._call
        tail = self._aligner.flush()  # zero-padded residual
        if call is not None and tail is not None:
            await call.send_audio(tail)

    async def _on_barge_in(self) -> None:
        call = self._call
        if call is None:
            return
        await call.cancel_playback()    # un-send what StandIn still holds
        if self._client is not None:
            await self._client.cancel_response()
        self._aligner.reset()           # the residual belongs to a dead turn


async def serve() -> None:
    server = CallServer(handler_factory=lambda: RealtimeHandler(API_KEY, PROMPT))
    await server.start()
    try:
        await asyncio.Event().wait()
    finally:
        await server.aclose()