Handle VOICE CALL: BEGIN arriving during answer without double ACTIVE transition

[Justinabox]

Summary

CallService.answer() can raise InvalidStateTransition when a modem emits VOICE CALL: BEGIN during the ATA response window. The executor correctly treats VOICE CALL: BEGIN as a URC and dispatches a CallStateEvent(ACTIVE), but answer() then unconditionally transitions the FSM to ACTIVE again.

This is a real modem-integration hazard because SIMCOM-style voice URCs can arrive interleaved with command responses.

Evidence

Affected code:

• callstack/protocol/executor.py:274-286 dispatches URCs that arrive while a command is in flight.
• callstack/protocol/urc.py:95-99 maps VOICE CALL: BEGIN to CallStateEvent(state=CallState.ACTIVE).
• callstack/voice/service.py:160-168 handles that event by transitioning RINGING/DIALING -> ACTIVE and enabling audio.
• callstack/voice/service.py:95-102 then returns from ATA and unconditionally calls await self._fsm.transition(CallState.ACTIVE).
• callstack/voice/state.py:27-34 does not allow ACTIVE -> ACTIVE, so the second transition raises.

Minimal reproduction run during scouting:

PYTHONPATH=. uv run --no-project --with pyserial-asyncio python - <<'PY'
import asyncio
from callstack.events.bus import EventBus
from callstack.events.types import RingEvent, CallStateEvent, CallState
from callstack.protocol.executor import ATResponse
from callstack.voice.service import CallService

class FakeAudio:
    running = False
    async def start(self):
        self.running = True
    async def stop(self):
        self.running = False

class FakeAT:
    def __init__(self, bus):
        self.bus = bus
        self.calls = []
    async def execute(self, cmd, **kwargs):
        self.calls.append(cmd)
        if cmd == 'ATA':
            await self.bus.emit(CallStateEvent(state=CallState.ACTIVE))
            await asyncio.sleep(0.01)
        return ATResponse(success=True, lines=['OK'])

async def main():
    bus = EventBus()
    at = FakeAT(bus)
    svc = CallService(at, FakeAudio(), bus)
    await bus.emit(RingEvent())
    await asyncio.sleep(0.01)
    try:
        await svc.answer()
    except Exception as exc:
        print(type(exc).__name__, str(exc))
        print('state', svc.state)
        print('calls', at.calls)

asyncio.run(main())
PY

Observed output:

InvalidStateTransition Invalid transition: CallState.ACTIVE -> CallState.ACTIVE
state CallState.ACTIVE
calls ['ATA', 'AT+CPCMREG=1']

The existing tests only queue OK for ATA (tests/test_call_service.py:106-125) and do not cover an interleaved VOICE CALL: BEGIN URC.

Repository health checks from this scout run:

git diff --check
# exit 0

PYTHONPATH=. uv run --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/ -q
# 288 passed in 3.94s

Duplicate check:

gh search issues --repo Justinabox/Callstack "VOICE CALL BEGIN answer InvalidStateTransition" --state open
# []

Expected behavior

Answering a ringing call should be idempotent with respect to the modem's connected-call URC ordering:

• If VOICE CALL: BEGIN arrives before the ATA command has returned, answer() should return a valid inbound CallSession and leave the service in ACTIVE.
• Audio enabling should happen once.
• No InvalidStateTransition should leak to the caller for a successfully answered call.

Actual behavior

When the URC transitions the FSM to ACTIVE first, answer() attempts a second ACTIVE transition and raises InvalidStateTransition, even though the call is already active.

Suggested fix direction

• Add a regression test where answering receives/emits CallStateEvent(ACTIVE) before ATA completes.
• In answer(), check self._fsm.state before transitioning, or make the connected-call handler set a pending flag/session so answer() can treat the URC-first path as success.
• Ensure _enable_audio() is not called twice across the URC path and the explicit answer path.

Acceptance criteria

• New test covers VOICE CALL: BEGIN during ATA and asserts answer() returns CallSession(direction="inbound") with state ACTIVE.
• Existing answer success/failure and remote hangup tests still pass.
• Audio transport setup happens exactly once in the URC-first answer path.

Verification gates

git diff --check
PYTHONPATH=. uv run --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/ -q