Stage 2: thread ownership at the SIP boundary, leg-state wiring, task hygiene #2

r · 2026-07-09T23:54:00Z

r commented

2026-07-09 23:54:00 +00:00

Second stage of the architecture-review roadmap. Stacked on #1 (feature/stage1-agent-surface) — merge that first; this diff shows only Stage 2.

The problem

Three thread domains were touching shared state with no synchronization:

Sippy's ED thread mutated _legs / _registered_devices directly from SIP handlers.
The asyncio loop mutated the same dicts in make_call / hangup.
run_in_executor(None, ...) put default-pool threads in charge of sippy UA objects — a third domain.

Plus: AudioTap.feed() called asyncio.Queue.put_nowait from the PJSUA2 thread (not thread-safe), the classifier ran FFTs and a per-sample Python Goertzel loop inline on the event loop, on_leg_state_change was never wired (outbound call state never reached CallManager), and hold-slayer/receptionist/recording-timeout tasks were bare create_task calls — unretained and never cancelled.

The ownership rule (now structural, no locks)

The asyncio loop owns all application-visible state. The only mutator is the new _on_engine_event funnel; Sippy handlers extract plain strings on the ED thread and post {leg_id, kind, ...} payloads via run_coroutine_threadsafe.
The ED thread owns every sippy object plus new _ed_ua_to_leg / _ed_leg_to_ua maps (replacing the UA scans over _legs). Loop-side commands hop over via ED2.callFromThread. UA references no longer live on SipCallLeg.
leg_id strings are the only tokens that cross.

Found while implementing: the installed sippy 2.x exposes ED2, not ED — the old from sippy.Core.EventDispatcher import ED raised ImportError inside the thread runner, so the Sippy event loop could never start ("Sippy event loop crashed" swallowed in logs). Fixed as part of the funnel work. Note the UA(event_cb=controller) usage still has 2.x API drift; full sippy bring-up is Stage 4 territory (explicit engine mode).

Also in this stage

Leg state finally drives call state: ringing/connected map onto call status (never stomping service states like ON_HOLD); a call ends when its last leg terminates, so transfers survive the trunk leg dropping. CallManager.unmap_leg() / legs_for_call() added.
Classifier off the loop: AudioClassifier.classify() runs the pure classify_chunk in asyncio.to_thread and updates history on the loop. All four hold_slayer call sites route through it — fixing the loop-blocking and the 2-of-4 history-update gap in one move. The DTMF Goertzel per-sample loop is replaced with the mathematically-equal DFT-bin power (vectorized); all 18 classifier tests unchanged and passing.
Task hygiene: gateway.spawn() tracks per-call tasks, stop() cancels and gathers them; the recording safety-timeout is retained and cancelled on stop_recording (with a guard so the fired timeout doesn't cancel itself); the engine tracks incoming-call dispatch tasks.
AudioTap captures its loop at construction; feed() hops via call_soon_threadsafe with drop-oldest on overflow.

Tests & verification

10 new tests in tests/test_concurrency.py: funnel events posted from a real foreign thread (register/re-register/deregister, INVITE auto-answer, BYE, DTMF, trunk), AudioTap cross-thread feed, classifier history recording, leg-state → call-status propagation incl. the ON_HOLD no-stomp case, and stop() cancelling spawned tasks.
Suite: 114 passed, same 2 pre-existing failures as main.
Live boot on loopback: endpoints healthy, clean SIGTERM shutdown, no "Task was destroyed" warnings.

🤖 Generated with Claude Code

Second stage of the architecture-review roadmap. **Stacked on #1** (`feature/stage1-agent-surface`) — merge that first; this diff shows only Stage 2. ## The problem Three thread domains were touching shared state with no synchronization: 1. Sippy's ED thread mutated `_legs` / `_registered_devices` directly from SIP handlers. 2. The asyncio loop mutated the same dicts in `make_call` / `hangup`. 3. `run_in_executor(None, ...)` put default-pool threads in charge of sippy UA objects — a third domain. Plus: `AudioTap.feed()` called `asyncio.Queue.put_nowait` from the PJSUA2 thread (not thread-safe), the classifier ran FFTs and a per-sample Python Goertzel loop inline on the event loop, `on_leg_state_change` was never wired (outbound call state never reached CallManager), and hold-slayer/receptionist/recording-timeout tasks were bare `create_task` calls — unretained and never cancelled. ## The ownership rule (now structural, no locks) - **The asyncio loop owns all application-visible state.** The only mutator is the new `_on_engine_event` funnel; Sippy handlers extract plain strings on the ED thread and post `{leg_id, kind, ...}` payloads via `run_coroutine_threadsafe`. - **The ED thread owns every sippy object** plus new `_ed_ua_to_leg` / `_ed_leg_to_ua` maps (replacing the UA scans over `_legs`). Loop-side commands hop over via `ED2.callFromThread`. UA references no longer live on `SipCallLeg`. - **`leg_id` strings are the only tokens that cross.** Found while implementing: the installed sippy 2.x exposes `ED2`, not `ED` — the old `from sippy.Core.EventDispatcher import ED` raised ImportError inside the thread runner, so the Sippy event loop could **never** start ("Sippy event loop crashed" swallowed in logs). Fixed as part of the funnel work. Note the `UA(event_cb=controller)` usage still has 2.x API drift; full sippy bring-up is Stage 4 territory (explicit engine mode). ## Also in this stage - **Leg state finally drives call state**: ringing/connected map onto call status (never stomping service states like ON_HOLD); a call ends when its *last* leg terminates, so transfers survive the trunk leg dropping. `CallManager.unmap_leg()` / `legs_for_call()` added. - **Classifier off the loop**: `AudioClassifier.classify()` runs the pure `classify_chunk` in `asyncio.to_thread` and updates history on the loop. All four hold_slayer call sites route through it — fixing the loop-blocking and the 2-of-4 history-update gap in one move. The DTMF Goertzel per-sample loop is replaced with the mathematically-equal DFT-bin power (vectorized); all 18 classifier tests unchanged and passing. - **Task hygiene**: `gateway.spawn()` tracks per-call tasks, `stop()` cancels and gathers them; the recording safety-timeout is retained and cancelled on `stop_recording` (with a guard so the fired timeout doesn't cancel itself); the engine tracks incoming-call dispatch tasks. - `AudioTap` captures its loop at construction; `feed()` hops via `call_soon_threadsafe` with drop-oldest on overflow. ## Tests & verification - 10 new tests in `tests/test_concurrency.py`: funnel events posted from a real foreign thread (register/re-register/deregister, INVITE auto-answer, BYE, DTMF, trunk), AudioTap cross-thread feed, classifier history recording, leg-state → call-status propagation incl. the ON_HOLD no-stomp case, and stop() cancelling spawned tasks. - Suite: **114 passed**, same 2 pre-existing failures as `main`. - Live boot on loopback: endpoints healthy, clean SIGTERM shutdown, no "Task was destroyed" warnings. 🤖 Generated with [Claude Code](https://claude.com/claude-code)

r added 1 commit 2026-07-09 23:54:01 +00:00

fix: enforce thread ownership at the Sippy/PJSUA2 boundary 5880b59872

Three thread domains were mutating shared dicts with no locks: Sippy's
ED thread wrote _legs/_registered_devices directly from SIP handlers,
the asyncio loop wrote them from make_call/hangup, and
run_in_executor(None, ...) had default-pool threads driving sippy UA
objects. AudioTap.feed() pushed into an asyncio.Queue (not
thread-safe) from the PJSUA2 thread.

New ownership rule, enforced structurally:
- The asyncio loop owns all app-visible state; the only mutator is the
  new _on_engine_event funnel. Sippy handlers extract plain strings on
  the ED thread and post via run_coroutine_threadsafe.
- The ED thread owns sippy objects plus _ed_ua_to_leg/_ed_leg_to_ua;
  loop-side commands (INVITE/BYE/DTMF/trunk register) hop over via
  ED2.callFromThread. UA references no longer live on SipCallLeg.
- AudioTap captures its loop and feed() hops via call_soon_threadsafe.
- Fix ED import: installed sippy 2.x exposes ED2, not ED — the old
  import could never start the event loop.

Also:
- Wire the never-connected on_leg_state_change callback: outbound
  ringing/connected/terminated now reaches CallManager; a call ends
  when its last leg terminates (transfers keep it alive). Adds
  CallManager.unmap_leg/legs_for_call.
- AudioClassifier.classify(): async entry that runs the FFT work in
  asyncio.to_thread and updates history on the loop — all four
  hold_slayer call sites now route through it, fixing both the
  loop-blocking and the 2-of-4 history gap. DTMF Goertzel loop
  replaced by the equivalent vectorized DFT-bin power.
- Task hygiene: gateway.spawn() tracks hold-slayer/receptionist tasks
  and stop() cancels them; recording safety-timeout task is retained
  and cancelled on stop_recording; engine tracks incoming-call
  dispatch tasks.

10 new tests: funnel events from a foreign thread, auto-answer
fallback, AudioTap cross-thread feed, classifier history, leg-state →
call status (including no stomping of ON_HOLD), stop() cancellation.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

r referenced this pull request

2026-07-10 00:29:56 +00:00

Stage 3: composition root, core↔services cycle break, shared REST+MCP data layer #3

r merged commit 92d02f998f into feature/stage1-agent-surface

2026-07-10 11:10:50 +00:00

r referenced this issue from a commit

2026-07-10 11:10:51 +00:00

Merge pull request 'Stage 2: thread ownership at the SIP boundary, leg-state wiring, task hygiene' (#2) from feature/stage2-concurrency into feature/stage1-agent-surface

r referenced this pull request

2026-07-10 17:45:58 +00:00

Stage 6: learn_call_flow rebuilt on the learner, docs truth sweep #6

Sign in to join this conversation.