Files
hold-slayer/docs/development.md
Robert Helewka ff7ea8623a Stage 6: learn_call_flow rebuilt on the learner, docs truth sweep
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>
2026-07-10 13:45:35 -04:00

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 runner
  • pytest-asyncio — async test support
  • pytest-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_FAILED events — the user is notified
  • HTTP errors in the API return structured error responses

Event-Driven Architecture

All components communicate through the EventBus:

  1. Publishers — SIP engine, Hold Slayer, classifier, services
  2. 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

  1. Create a feature branch
  2. Write tests for new functionality
  3. Ensure all tests pass: pytest tests/ -v
  4. Follow existing code conventions
  5. Update documentation in /docs if adding new features
  6. Submit a pull request