The call-flow learner finally gets fed: exploration mode records its IVR discoveries on the call (ActiveCall.exploration_steps) instead of throwing them away, persistence stores them in the call record's metadata, and the rebuilt learn_call_flow MCP tool turns a completed exploration call into a stored flow via CallFlowLearner — correct constructor (llm_client from get_llm, heuristic labels when the LLM is unavailable), build for a new number, merge/refine when a flow already exists. save_learned_flow/update_flow_from_model keep the CallFlow↔row mapping in call_persistence. Test gaps closed: tests/test_learner.py (discoveries→linked steps, exploration persistence, learn-then-refine through the in-memory MCP client, no-data and unknown-call answers) and tests/test_websocket.py (4401 without token, trunk-status-then-replay on connect, per-call stream filtering). Docs aligned to code: README (15 tools incl. learn_call_flow, HTTP not SSE, Python 3.12+, PostgreSQL+Alembic — no SQLite fallback, media pipeline marked stub-mode until pjsua2 installed, Alembic and honest /health checked off); docs/mcp-server.md rewritten against the actual tool surface (hangup not end_call, real params, 3 real resources, /mcp/ streamable HTTP + bearer auth); architecture/development/ configuration drift fixed. pyproject: pruned never-imported deps (websockets, librosa, soundfile, python-multipart). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
5.4 KiB
Development
Setup
Prerequisites
- Python 3.12+
- Ollama (or any OpenAI-compatible LLM) — for IVR menu analysis
- Speaches or Whisper API — for speech-to-text (optional for dev)
- A SIP trunk account — for making real calls (optional for dev)
Install
git clone <repo-url>
cd hold-slayer
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
Dev Dependencies
The [dev] extras include:
pytest— test runnerpytest-asyncio— async test supportpytest-cov— coverage reporting
Testing
Run All Tests
pytest tests/ -v
Run Specific Test Files
pytest tests/test_audio_classifier.py -v # 18 tests — waveform analysis
pytest tests/test_call_flows.py -v # 10 tests — call flow models
pytest tests/test_hold_slayer.py -v # 20 tests — IVR nav, EventBus, CallManager
pytest tests/test_services.py -v # 27 tests — LLM, notifications, recording,
# analytics, learner, EventBus
Run with Coverage
pytest tests/ --cov=. --cov-report=term-missing
Test Architecture
Tests are organized by component:
| File | Tests | What's Covered |
|---|---|---|
test_audio_classifier.py |
18 | Silence, tone, DTMF, music, speech detection; feature extraction; classification history |
test_call_flows.py |
10 | CallFlowStep types, CallFlow navigation, serialization roundtrip, create/summary models |
test_hold_slayer.py |
20 | IVR menu navigation (6 intent scenarios), EventBus pub/sub, CallManager lifecycle, MockSIPEngine |
test_services.py |
27 | LLMClient init/stats/chat/JSON/errors/IVR analysis, NotificationService event mapping, RecordingService paths, CallAnalytics summaries, CallFlowLearner build/merge, EventBus integration |
Known Test Issues
test_complex_tone_as_music — A synthetic multi-harmonic tone is classified as LIVE_HUMAN instead of MUSIC. This is a known edge case. Real hold music has different spectral characteristics than synthetic test signals. This test documents the limitation rather than a bug.
Writing Tests
All tests use pytest-asyncio for async support. The test configuration in pyproject.toml:
[tool.pytest.ini_options]
asyncio_mode = "auto"
This means all async def test_* functions automatically run in an asyncio event loop.
Pattern for testing services:
import pytest
from services.llm_client import LLMClient
class TestLLMClient:
def test_init(self):
client = LLMClient(base_url="http://localhost:11434/v1", model="llama3")
assert client._model == "llama3"
@pytest.mark.asyncio
async def test_chat(self):
# Mock httpx for unit tests
...
Pattern for testing EventBus:
import asyncio
from core.event_bus import EventBus
from models.events import EventType, GatewayEvent
async def test_publish_receive():
bus = EventBus()
sub = bus.subscribe()
event = GatewayEvent(type=EventType.CALL_STARTED, call_id="test", data={})
await bus.publish(event)
received = await asyncio.wait_for(sub.get(), timeout=1.0)
assert received.type == EventType.CALL_STARTED
Project Conventions
Code Style
- Type hints everywhere — All function signatures have type annotations
- Pydantic models — All data structures are Pydantic BaseModel or dataclass
- Async by default — All I/O operations are async
- Logging — Every module uses
logging.getLogger(__name__) - Docstrings — Module-level docstrings explain purpose and usage
File Organization
module.py
├── Module docstring (purpose, usage examples)
├── Imports (stdlib → third-party → local)
├── Constants
├── Classes
│ ├── Class docstring
│ ├── __init__
│ ├── Public methods (async)
│ └── Private methods (_prefixed)
└── Module-level functions (if any)
Error Handling
- Services never crash the call — All service errors are caught, logged, and return sensible defaults
- LLM failures return empty string/dict — the Hold Slayer falls back to waiting
- SIP errors publish
CALL_FAILEDevents — the user is notified - HTTP errors in the API return structured error responses
Event-Driven Architecture
All components communicate through the EventBus:
- Publishers — SIP engine, Hold Slayer, classifier, services
- Subscribers — WebSocket handler, MCP server, notification service, analytics
This decouples components and makes the system extensible. Adding a new feature (e.g., Slack notifications) means subscribing to events — no changes to existing code.
Dependency Injection
The AIPSTNGateway owns all services and is injected into FastAPI routes via api/deps.py:
# api/deps.py
async def get_gateway() -> AIPSTNGateway:
return app.state.gateway
# api/calls.py
@router.post("/outbound")
async def make_call(request: CallRequest, gateway: AIPSTNGateway = Depends(get_gateway)):
...
This makes testing easy — swap the gateway for a mock in tests.
Contributing
- Create a feature branch
- Write tests for new functionality
- Ensure all tests pass:
pytest tests/ -v - Follow existing code conventions
- Update documentation in
/docsif adding new features - Submit a pull request