Serialize or reject concurrent USSD sends to avoid response cross-correlation

[Justinabox]

Summary

Concurrent USSDService.send() calls can both resolve with the first USSDResponseEvent, so a caller asking for *200# can receive the response for *100#. The service subscribes each in-flight call to the same global USSDResponseEvent stream without a lock, request id, or single-session guard.

Evidence

Affected code:

• callstack/ussd.py:51-57 creates a per-call future and subscribes a capture handler to every USSDResponseEvent.
• callstack/ussd.py:59-65 sends the command and waits for the first event that arrives.
• callstack/ussd.py:68-69 unsubscribes only after the future resolves/times out.
• There is no lock/rejection path around concurrent USSD sessions, even though USSD is a single interactive modem/carrier session.

No-hardware reproduction from main (6926b369cdba):

PYTHONPATH=. uv run --no-project --with pyserial-asyncio python - <<'PY'
import asyncio
from callstack.events.bus import EventBus
from callstack.events.types import USSDResponseEvent
from callstack.protocol.executor import ATResponse
from callstack.ussd import USSDService

class FakeExecutor:
    def __init__(self, bus):
        self.bus = bus
        self.calls = []
    async def execute(self, command, expect=("OK",), timeout=5.0):
        self.calls.append(command)
        if command.startswith('AT+CUSD'):
            seq = len(self.calls)
            async def later():
                await asyncio.sleep(0.01 if seq == 1 else 0.02)
                await self.bus.emit(USSDResponseEvent(status=0, message=f'response-for-{command}', encoding=15))
            asyncio.create_task(later())
        return ATResponse(success=True, lines=['OK'])

async def main():
    bus = EventBus()
    service = USSDService(FakeExecutor(bus), bus)
    results = await asyncio.gather(
        service.send('*100#', timeout=0.2),
        service.send('*200#', timeout=0.2),
    )
    print('result_messages:', [r.message for r in results])
    print('distinct_results:', len({r.message for r in results}))

asyncio.run(main())
PY

Actual output:

result_messages: ['response-for-AT+CUSD=1,"*100#",15', 'response-for-AT+CUSD=1,"*100#",15']
distinct_results: 1

Both tasks returned the first response.

Expected behavior

Callstack should not cross-correlate concurrent USSD responses. Because USSD sessions are effectively single-session on the modem/carrier side, either:

1. serialize USSDService.send() with an async lock so the second request is sent only after the first response/timeout/cancel path completes, or
2. fail fast with a clear RuntimeError/domain error when a USSD request is already in flight.

Suggested fix direction

• Add an in-flight guard or asyncio.Lock around USSDService.send().
• Ensure timeout/error paths release the guard and unsubscribe the capture handler.
• Add a regression test that starts two sends concurrently and asserts they are serialized or that the second call fails explicitly rather than returning the first call's response.
• Keep timeout errors redacted; do not include raw USSD codes in exception text.

Acceptance criteria

• Two concurrent USSDService.send() calls cannot both resolve from the same USSDResponseEvent.
• If serialization is chosen, the second command is not written until the first response or timeout completes.
• If fail-fast is chosen, the second call receives a clear, redacted error and no second AT+CUSD write occurs.
• Existing /ussd/send HTTP timeout validation and parser tests remain green.

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/test_ussd.py tests/test_http_api_validation.py -q
PYTHONPATH=. uv run --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/ -q

Duplicate check

Searched existing issues/PRs for concurrent USSD send, USSDService response correlation, and same response terms. I did not find an existing issue or PR covering concurrent USSD cross-correlation.