r/hold-slayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: r:c00cf0267649ad27a1c0b4fa2e2dea19d54591f8
Branches Tags
r:main r:feat/pjsua2-media-tap
...
compare: r:main
Branches Tags
r:main r:feat/pjsua2-media-tap

10 Commits
c00cf02676 ... main

Author SHA1 Message Date
Robert Helewka
5c178bb7bd feat(auth): rate-limit the unauthenticated /auth/* edge
All checks were successful
CVE Scan & Docker Build / security-scan (push) Successful in 45s
Details
CVE Scan & Docker Build / build-and-push (push) Successful in 1m53s
Details 
Blanket per-endpoint limits would have been the wrong shape here. Every
REST/WS/MCP surface is owner-only — an unauthenticated request is rejected by
resolve_bearer/is_owner before any handler runs — so limiting them would
mostly throttle the single legitimate operator, and real spend control for
outbound calls is already max_concurrent_calls in gateway.make_call.

What is genuinely exposed is the handful of /auth/* routes that must answer
before an identity exists. /auth/callback and /auth/refresh-callback each make
an outbound token exchange with Casdoor on every request; /auth/me opens a DB
session and runs a token lookup. All are free to trigger and none are cheap to
serve. /auth/logout is left unlimited — it builds a redirect URL and does no
I/O.

Not a defence against credential guessing: PATs are secrets.token_urlsafe(32)
(256 bits) compared by SHA-256 digest, so brute force was never the threat.
This is about unauthenticated work an attacker controls.

Fixed-window, in-process, no new dependency — one operator and one process
make a shared counter store infrastructure without a purpose. The bucket store
is bounded and evicts oldest-first, since an unbounded map keyed by source
address would itself be the exhaustion vector.

The limiter keys on the socket peer and deliberately ignores X-Forwarded-For.
That header is attacker-controlled unless a trusted proxy overwrites it, and
this app establishes no such trust; keying on it would let one client present
as thousands and make the limiter worse than useless. Behind the estate's
reverse proxy the limit is therefore per-proxy, not per-caller — correct for
exhaustion and honest about what it can enforce. Per-caller limits need an
explicit trusted-proxy config, noted in CLAUDE.md so it isn't added silently.

Verified against a real server: exactly 30 requests pass, then 429 with
Retry-After: 60, while an owner-gated route serves 40/40. The 429s appear in
the JSON access log with queryable status_code and client_addr, so an attack
is visible in Loki. The wiring test identifies the dependency by qualname
rather than string search, and was mutation-checked by removing the limit from
/auth/me.

Also documents 401/403/429 in the API reference — 401 and 403 have existed
since auth landed but were never in the status-code table. Phase 4 is now
complete.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-30 18:46:54 -04:00
Robert Helewka
59f0136370 test: pin graceful degradation of STT, LLM and TTS failures 
The Phase 4 checkbox was stale: the behaviour is already implemented across
the services, but almost nothing tested it, so a refactor could have quietly
removed it. The failure mode being guarded against is silent — an un-caught
exception in any of these paths aborts a live phone call.

What was verified rather than assumed:
- The classifier is purely spectral. classify_chunk takes only audio_data;
  there is no transcript parameter, so STT cannot be a hard dependency of hold
  detection. The README's "classifier works without STT" parenthetical
  described an aspiration, not a coupling.
- All three transcribe() callers catch, publish an ERROR event naming the
  service, and return "". TranscriptionService.transcribe raises deliberately
  so callers own the fallback — swallowing it made a down Speaches look like
  "the AI is deciding badly".
- An LLM failure in the receptionist still returns a usable decision, and in
  hold_slayer falls through to "press 0 for agent".
- _service_error is itself wrapped, so a dead event bus cannot turn
  degradation into a second failure.
- A failed transcription sets available=False, which is what /health reads —
  a failure that doesn't record itself makes the probe lie.

Each test was checked by mutation: removing the try/except in
HoldSlayerService._transcribe fails three of them. Without that check these
would assert behaviour they don't actually constrain.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-30 18:46:39 -04:00
Robert Helewka
e2051f7486 docs: document SIP_ENGINE and correct config tables against the models
All checks were successful
CVE Scan & Docker Build / security-scan (push) Successful in 44s
Details
CVE Scan & Docker Build / build-and-push (push) Successful in 1m53s
Details 
Started as the SIP_ENGINE row flagged in the last commit. Cross-checking the
tables against config.py mechanically (rather than by eye) turned up more,
including two entries that were actively wrong.

Corrections:
- GATEWAY_RTP_PORT_MIN/MAX and GATEWAY_HOST are documented in
  configuration.md but do not exist — no code reads them and they are absent
  from .env.example. Setting them today does nothing. Replaced with the real
  GATEWAY_SIP_ fields (host/port/domain).
- GATEWAY_SIP_PORT was documented as 5080 in two places; the code and
  .env.example both say 5060.
- DATABASE_URL was documented with a SQLite default. There is none, and
  startup exits if it is unset.

Additions — every env var the models accept is now documented somewhere
(verified bidirectionally: nothing in the models undocumented, nothing
documented that the models reject):
- Server section: HOST, PORT, DEBUG, LOG_LEVEL, LOG_FORMAT
- Safety section: MAX_CONCURRENT_CALLS, USE_MOCK_SIP, SIP_ENGINE
- Receptionist section (configuration.md had none, though seven vars exist)

Structural staleness, from the PR #8 media-plane work:
- core/pjsua_engine.py was absent from the component list and file tree; so
  were dial_plan.py (the emergency guard) and sip_engine.py.
- architecture.md's banner still read "media plane in transition". The engine
  landed; it is now two selectable engines with the audio consequence stated.
- Tech Stack described "single-process async architecture" — the
  simplification CLAUDE.md explicitly calls out. Now points at the threading
  model, since there are three execution contexts.
- The Asterisk lab shipped in PR #8 with its own README but nothing linked to
  it. Linked from the test section and both doc indexes.
- CLAUDE.md's "no structured JSON logging" gap is closed; test count was 146
  across 16 files, now 189 across 19. The other listed gaps (no /metrics, no
  rate limiting, no health-probe log filter) were re-verified and still hold.

Deliberately not hardcoding a test count in the README — that is the same
staleness this commit is clearing up. All internal links and anchors verified
to resolve; 189 tests pass; lint unchanged at its 216 baseline.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-30 17:57:31 -04:00
Robert Helewka
1644999bcb feat(logging): structured JSON logs, uvicorn access log included 
Hold Slayer's logs are shipped to Loki by the host's Alloy agent, which
reads container stdout. Text lines arrive there as an opaque blob:
filtering on a status code meant regex over a formatted string. This adds
LOG_FORMAT=json (default "text", so local dev stays readable) rendering
one JSON object per line.

Two parts were less obvious than a format= argument would suggest, and
both are why this is a module rather than a basicConfig tweak:

Uvicorn attaches its own handlers to `uvicorn` and `uvicorn.access` with
propagate=False, so configuring only the root logger would have left the
access log — the highest-volume, most useful stream — as colourised text
next to our JSON. configure_logging clears those handlers and re-enables
propagation, and is called both at import (for startup config checks) and
in lifespan (uvicorn configures itself after importing the app). The
__main__ path passes log_config=None so uvicorn never applies its own.

The access record's payload lives in record.args as a 5-tuple, not in the
message. Formatting it would throw the structure away and force Loki to
parse it back out, so the tuple is unpacked into real fields and
status_code is emitted as a number for range filtering.

Also drops uvicorn's `color_message` extra, an ANSI-coloured duplicate of
the message that generic extra-promotion would otherwise copy into every
startup line — the same unreadable-in-Grafana problem recently fixed for
the lab's Asterisk logs.

Verified against a real uvicorn server: 39/39 lines valid JSON, zero ANSI
escapes, no duplicates, access lines structured with correct status codes;
text mode unchanged. Thread name is included off the main thread, since
"which execution context logged this" is the first question when debugging
across the asyncio/Sippy/PJSUA2 boundary. SecretStr extras stay masked.

README Phase 4 item ticked; LOG_FORMAT and the previously-undocumented
LOG_LEVEL added to the config table and .env.example.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-30 06:18:06 -04:00
Robert Helewka
98c80e3a56 Merge pull request 'PJSUA2 media plane — audio finally reaches the classifier' (#8) from feat/pjsua2-media-tap into main
All checks were successful
CVE Scan & Docker Build / security-scan (push) Successful in 42s
Details
CVE Scan & Docker Build / build-and-push (push) Successful in 1m39s
Details 
Reviewed-on: #8
 2026-07-29 21:34:09 +00:00
Robert Helewka
3150f78552 fix(lab): make Asterisk logs usable — drop ANSI codes and healthcheck noise 
Logging was configured correctly and shipping to Loki, but the stream was
useless: measured at 100% healthcheck chatter, with every line wrapped in ANSI
escape codes. The real SIP events were there and completely buried.

Three causes, each masking the next:

- asterisk.conf was written in the first lab commit with `nocolor = yes` and
  never mounted, so the setting had no effect. Now mounted. It was also
  overriding [directories] and runuser/rungroup, which the image sets up
  correctly itself — removed, since overriding them risks breaking the
  container for no gain.

- The image's command is `-vvvdddf`: verbosity 3 and debug 3 forced on the
  command line, which overrides both asterisk.conf and logger.conf. Overridden
  in compose to drop -v and -d; warnings and errors still log, and verbosity
  is raisable at runtime when tracing a call.

- The actual source: the image's healthcheck makes ~7 separate `asterisk -rx`
  connections every 30s, and Asterisk logs a connect/disconnect pair for each.
  Replaced with a single check on a 60s interval, and the check now runs
  `pjsip show transports` rather than `core show version` — that fails when
  Asterisk is up but unconfigured, which is exactly the state that produced a
  "healthy" container with no SIP stack on first deploy.

logger.conf drops both `notice` and `verbose`, which is where those pairs
arrive.

Verified on galatea: noise down from ~48 to 8 lines per two minutes (-83%),
zero ANSI codes in Loki, 90% of the stream now signal, container still
healthy, transport and dialplan intact.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-29 17:26:03 -04:00
Robert Helewka
c516f659cc feat(lab): add softphone endpoint so device registration can be tested 
Asterisk is the registrar for devices, not Hold Slayer. A softphone REGISTERs
to the lab and the gateway transfers a live call to it by dialling extension
2001. This is deliberate: Hold Slayer's own SIP listener answers 200 OK to any
REGISTER with no digest challenge, so anything on the network could register
as a device and receive transferred calls. Keeping registration in Asterisk
means the lab does not exercise or depend on that path, and the device is
authenticated.

The pjsua CLI built alongside the Python bindings is the test device — same
library stack as the gateway, so no new dependency. Verified end to end: a
gateway call to 2001 produces two channels Up under one bridge id.

Three things that cost time and are now written down:

- `--realm=asterisk`, not `--realm='*'`: the wildcard fails against Asterisk's
  digest challenge with PJSIP_EFAILEDCREDENTIAL.

- pjsua is an interactive console app and exits ~8s after start if stdin is
  closed or /dev/null. `script -qfc` and `setsid </dev/null` both appear to
  work — registration succeeds — and then the process dies, leaving a stale
  contact in Asterisk that routes INVITEs to a port nobody is listening on.
  Hold a fifo open on stdin instead, and verify the port is actually bound
  rather than trusting `pjsip show contacts`.

- Qualify is off for this AOR: the pjsua console does not answer OPTIONS, so
  polling marks a working softphone Unavail and the dialplan refuses to ring
  it. The 2001 guard therefore tests PJSIP_AOR(softphone,contact) rather than
  DEVICE_STATE. A real hardphone answers OPTIONS and can have it re-enabled.

The identify block now matches source address *and port*. A host-only match
claims every packet from that address, so a co-located softphone's REGISTER
was attributed to the gateway endpoint and checked against the gateway's
password — surfacing as "Failed to authenticate" on a correct password.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-29 07:57:17 -04:00
Robert Helewka
92c45e9c4d fix(lab): make the speech fixture actually classify as speech 
The speech fixture classified as LIVE_HUMAN on its first 3s window and drifted
to MUSIC for every window after. I had validated only the first window and
reported the fixture as verified, which overstated it: any lab result resting
on that fixture — the hold-slayer scenarios above all — was proving less than
it appeared to.

The cause was one modelling error, not a tuning problem. `_detect_tonality`
looks for an autocorrelation peak above 0.5 in the 50-1000 Hz lag range, and
each syllable used a *constant* f0, which is perfectly periodic there. That
scored is_tonal=True, handing the music score a free 0.3 that speech could not
outrun — and the decision requires speech_score to strictly exceed
music_score, so ties went to music.

Real voices glide and jitter, so the periodicity never locks. The fundamental
now follows a per-syllable pitch contour (rise or fall, plus ~2% cycle-to-cycle
jitter), with the frequency integrated to phase rather than multiplied by t —
`2*pi*f*t` is only a chirp when f is the instantaneous rate, which it is not
once f0 itself moves. is_tonal is now False in every window.

Two smaller fixes fell out of that:

- Aspiration noise is high-passed rather than broadband. Flat noise puts
  energy in every Goertzel bin, so the strongest DTMF row and column both
  clear the detector's `total_power * 0.1` threshold and each syllable reads
  as a keypress. A first-difference filter leaves the 697-1633 Hz bands
  comparatively empty. The level is set for margin — spectral flatness lands
  at ~0.46, mid-way through the 0.1-0.5 band, not on an edge.

- The music fixture gained two more harmonics and a recording-style noise
  floor. Windows straddling a chord change had a momentarily sparse spectrum
  and fell *below* the music score's 0.05 flatness floor, scoring as speech.

All three fixtures now classify correctly in 100% of windows (music 27/27,
speech 5/5, silence 2/2), and remain correct when the window is stepped by
half a window — a fixture that only works on aligned boundaries would still
be a trap in a live call, where the analysis window has no relationship to
where the audio began.

Confirmed on a real call through the lab: scenario 1003 now shows the whole
hold-slayer arc, speech -> sustained music -> speech, matching the dialplan.

tests/test_lab_fixtures.py guards this: it sweeps every window rather than
sampling the first, which is exactly what the original validation missed, and
checks the generator is byte-for-byte deterministic. It skips when the
fixtures have not been generated, since they are gitignored.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-29 07:56:58 -04:00
Robert Helewka
2a05be27bf feat(sip): add PJSUA2 engine — audio finally reaches the classifier 
The gateway can now hear. Verified end to end against the Asterisk lab:
speech classifies as live_human, hold music as music, the speech→music
transition tracks the dialplan, and DTMF reaches a real IVR (Asterisk logged
"caller pressed 1 -> accounts" and branched). RTP stats show 0% packet loss.

PJSUAEngine implements the existing SIPEngine interface, so the gateway,
call manager and hold-slayer service are unchanged. It is selected with
SIP_ENGINE=pjsua2; the default stays "sippy" while this is proven, and the
mock remains opt-in as before.

Why a new engine rather than fixing the old path: PJSUA2 exposes no
standalone RTP media object, so it will not surface media for a dialog it
does not own. Owning the dialog is the price of owning the media. Sippy keeps
the SBC roles it is good at — device registration, routing, leg bridging —
and SippyEngine remains fully functional for signalling; it simply cannot
carry media, which its media branch now says plainly instead of calling a
method that could never work.

The safety invariants are untouched. This engine is reachable only through
gateway.make_call, which refuses emergency numbers and enforces the
concurrency cap before any SIP action. No new dial path was introduced.

MediaPipeline.add_remote_stream(host, port) is replaced by
attach_call_media(stream_id, audio_media), called from onCallMediaState —
the one place PJSUA2 hands out RTP-backed media. Taps requested before media
comes up are attached when it does, so the classifier never misses the start
of a call.

Three crash/lifetime bugs found by running against the real bindings, none of
which any unit test would have caught:

- pj.Call and pj.Account objects finalised after libDestroy() abort the
  process on a native assertion, exactly as media ports do. Both are now
  dropped and collected before the pipeline destroys the endpoint.
- PJSUA2 keeps delivering callbacks during interpreter teardown, when module
  globals may already be cleared. The callbacks alias what they need locally
  and swallow everything: a raise there escapes into C++ and takes the worker
  thread with it.
- hangup() only queues the BYE, so shutdown deleted the account with a call
  still active and left the far end on an unclosed dialog. stop() now waits
  briefly for the teardown to complete.

Threading follows the established rule: PJSUA2 worker threads reach the loop
only through _post_from_pj → run_coroutine_threadsafe, and any thread PJSUA2
did not create registers itself before touching a PJSUA2 object.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-29 07:06:56 -04:00
Robert Helewka
7979e70705 feat(media): implement PJSUA2 audio capture port; document the media-plane refactor 
Implements the tap half of the media path and records why the other half
requires moving call placement into PJSUA2.

MediaPipeline.create_tap was a stub: it logged "🎤 Audio tap created" and
returned a tap that nothing ever fed, so the classifier received no audio on
a live call. It now builds a real pj.AudioMediaPort subclass whose
onFrameReceived converts the SWIG ByteVector to PCM bytes and fans it out to
every tap on the stream.

One capture port per stream, shared by all taps: a second port on the same
stream would be mixed back into the conference bridge and the call would echo.

Thread safety is the constraint here. onFrameReceived runs on a PJSUA2 worker
thread — a third execution context beside the asyncio loop and the Sippy ED
thread — and touches nothing but AudioTap.feed, which hops to the owning loop
via call_soon_threadsafe. An exception escaping into PJSUA2's C++ callback
would tear down the worker thread and silently kill media for every call, so
the handler catches and logs once per port rather than on every 20ms frame.

Also fixes a hard crash found while testing this against real PJSUA2: a media
port finalised after Endpoint.libDestroy() calls pjmedia_conf_remove_port
against a freed conference bridge and aborts the process on a native
assertion. Ports are now released in remove_stream while the bridge still
exists, and stop() forces a collection before libDestroy — dropping the last
Python reference is not sufficient on its own.

Verified against the real bindings: frames fan out to multiple taps, cross the
thread boundary intact, and shutdown is clean.

add_remote_stream remains a stub, and deliberately so. PJSUA2 exposes no
standalone RTP media object — every AudioMedia subclass in the Python
bindings is a file player, recorder, tone generator or capture port, and RTP
is reachable only via pj.Call.getAudioMedia() on a dialog PJSUA2 itself owns.
A design where Sippy owns the dialog can never obtain media from PJSUA2, so
that function cannot be written against this API. docs/architecture.md now
explains this and records the resolution: PJSUA2 places the trunk call while
Sippy keeps the SBC roles (device registration, routing, leg bridging), with
the emergency guard and concurrency cap staying first in gateway.make_call
regardless of which library dials.

The architecture doc also had drift unrelated to media: it described the
thread boundary as asyncio.run_in_executor() when the real mechanism is
run_coroutine_threadsafe / ED2.callFromThread, claimed two execution contexts
where there are three, and cited MediaPipeline.add_stream() and
SippyEngine.bridge() — neither of which exists. Corrected, with the data flow
now showing the emergency guard and concurrency cap in their real positions.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
 2026-07-29 06:56:50 -04:00
28 changed files with 2161 additions and 166 deletions
Expand all files Collapse all files

3
.env.example
View File

@@ -100,6 +100,9 @@ HOST=0.0.0.0
PORT=8000
DEBUG=false
LOG_LEVEL=info
# Log rendering: "text" (human-readable) or "json" (one object per line, for
# Loki/Alloy). The Docker image sets json; text is the default for local dev.
LOG_FORMAT=text
# --- Safety ---
# Max simultaneous calls the gateway will place (REST + MCP)

21
CLAUDE.md
View File

@@ -68,7 +68,7 @@ REST / WS / MCP / Dashboard (asyncio, FastAPI)
4. **Consistent Patterns** — config via pydantic-settings sub-configs; MCP tools
return formatted strings; REST returns Pydantic models; DB access via
`session_scope()`. Match the neighbours.
5. **Actually Works** — `pytest tests/ -v` (146 tests across 16 files). A change
5. **Actually Works** — `pytest tests/ -v` (189 tests across 19 files). A change
to call placement, routing, the classifier, or auth needs a test. The suite
runs against SQLite (`aiosqlite`) and the mock SIP engine — no trunk, no
Postgres required to test.
@@ -203,15 +203,24 @@ Surfaced honestly so you don't rediscover them as surprises. The README's
Phase 4/5/6 checklists track most of these; **don't fold fixes into unrelated
work** — raise them.
- **No structured JSON logging.** Logging is plain `logging.basicConfig` in
[main.py](main.py); there's no `LOG_FORMAT`/JSON path (README Phase 4 has this
unchecked). If Heluca observability wants JSON logs shipped to a collector,
that's a deliberate piece of work, not a drive-by.
- ~~**No structured JSON logging.**~~ Done: `LOG_FORMAT=json` in
[core/logging_config.py](core/logging_config.py), applied at import and again
in `lifespan` because uvicorn installs its own handlers (`propagate=False`)
after importing the app. The access log is included, with `status_code` as a
number so Loki can range-filter it. Text remains the default; the Docker image
sets json.
- **No `/metrics` endpoint and no Prometheus.** Unlike the metrics-bearing
estate services, there's no exposition endpoint here yet.
- **No health-probe access-log filter.** Every `/health` poll hits the access
log. Other estate services suppress probe noise; this one doesn't.
- **No rate limiting** on API endpoints (README Phase 4, unchecked).
- ~~**No rate limiting** on API endpoints.~~ Done, but *narrowly*: only the
unauthenticated `/auth/*` routes are limited
([core/rate_limit.py](core/rate_limit.py)), because every other surface is
already owner-gated and a limit there would throttle the sole operator. The
limiter keys on the **socket peer, not `X-Forwarded-For`** — behind the
estate's reverse proxy that means per-proxy, not per-caller. Per-caller limits
need an explicit trusted-proxy config; don't silently start trusting the
header.
- **Docker: single-image `Dockerfile` + `docker-compose.yaml`** (app +
`postgres:17`) ship in-repo; the Gitea CI (`cve-scan-docker-build.yml`) builds
the image on push to `main`. The compose stack requires SSO enabled

5
Dockerfile
View File

@@ -45,6 +45,11 @@ RUN pip install --no-cache-dir -e . \
EXPOSE 21081
# Structured logs by default in the container: the host's Alloy agent reads
# stdout and ships it to Loki, where text lines arrive as an unqueryable blob.
# Overridable (LOG_FORMAT=text) for interactive `docker run` debugging.
ENV LOG_FORMAT=json
# Migrations run in the app's own init_db() on boot (db/database.py), so no
# separate `alembic upgrade` here. Bind host/port from the same env vars
# pydantic-settings reads (HOST/PORT) so configured values and the actual bind

48
README.md
View File

@@ -52,8 +52,9 @@ You give it a phone number and an intent ("dispute a charge on my December state
## What's Implemented
### Core Engine
- **Sippy B2BUA Engine** (`core/sippy_engine.py`) — SIP call control, DTMF, bridging, conference, trunk registration
- **PJSUA2 Media Pipeline** (`core/media_pipeline.py`) — Audio routing, recording ports, conference bridge, WAV playback (stub mode until the `pjsua2` bindings are installed — see note below)
- **Sippy B2BUA Engine** (`core/sippy_engine.py`) — SIP call control, DTMF, bridging, conference, trunk registration. Signalling only: no audio reaches the classifier on this path
- **PJSUA2 SIP Engine** (`core/pjsua_engine.py`) — Places the call itself so it owns the dialog, which is the only way PJSUA2 will surface RTP. Select with `SIP_ENGINE=pjsua2`; see [docs/architecture.md](docs/architecture.md#media-plane-why-pjsua2-places-the-call)
- **PJSUA2 Media Pipeline** (`core/media_pipeline.py`) — Audio routing, capture ports, conference bridge, WAV playback (stub mode until the `pjsua2` bindings are installed — see note below)
- **Call Manager** (`core/call_manager.py`) — Active call state tracking, lifecycle management
- **Event Bus** (`core/event_bus.py`) — Async pub/sub with per-subscriber queues, type filtering, history
@@ -99,8 +100,13 @@ hold-slayer/
├── config.py # Pydantic settings from .env
├── core/
│ ├── gateway.py # Top-level gateway orchestrator
│ ├── sippy_engine.py # Sippy B2BUA SIP engine
│ ├── dial_plan.py # Emergency-number guard + number normalisation
│ ├── sip_engine.py # SIPEngine ABC + MockSIPEngine
│ ├── sippy_engine.py # Sippy B2BUA SIP engine (signalling only)
│ ├── pjsua_engine.py # PJSUA2 SIP engine (call control + media)
│ ├── media_pipeline.py # PJSUA2 audio routing
│ ├── logging_config.py # Text/JSON log formatting
│ ├── rate_limit.py # Fixed-window limiter for the /auth/* edge
│ ├── call_manager.py # Active call state management
│ └── event_bus.py # Async pub/sub event bus
├── services/
@@ -213,6 +219,15 @@ uvicorn main:app --host 0.0.0.0 --port 8000
pytest tests/ -v
```
No external services required — the suite runs against SQLite (`aiosqlite`) and
the mock SIP engine, so it needs neither a trunk nor PostgreSQL.
For the parts a unit test cannot reach — real SIP signalling, RTP, IVR
navigation against a live switch — there is an **Asterisk lab**: a fake PSTN
that answers calls, plays hold music, and runs scripted IVR menus, so call
paths can be exercised without dialling a real number or incurring telephony
charges. See [tests/lab/README.md](tests/lab/README.md).
## Docker
A single image bundles the FastAPI process and the built dashboard (the node
@@ -373,11 +388,19 @@ All configuration is via environment variables (see `.env.example`):
| `OWNER_NAME` | Casdoor username of the single operator (owner) | — (required if SSO on) |
| `PUBLIC_BASE_URL` | Public base URL for OAuth discovery (else derived from headers) | — |
| `MAX_CONCURRENT_CALLS` | Cap on simultaneous outbound calls | `4` |
| `HOST` | Bind address (off-loopback requires SSO enabled) | `0.0.0.0` |
| `PORT` | Bind port | `8000` |
| `DEBUG` | SQLAlchemy echo + uvicorn reload | `false` |
| `LOG_LEVEL` | Root log level (`debug`/`info`/`warning`/`error`) | `info` |
| `LOG_FORMAT` | `text` (human-readable) or `json` (structured, for Loki) | `text` |
| `USE_MOCK_SIP` | Run the mock SIP engine — no real calls. Must be asked for | `false` |
| `SIP_ENGINE` | `sippy` (signalling only) or `pjsua2` (call control + media) | `sippy` |
| `NOTIFY_SMS_NUMBER` | SMS notification number (optional) | — |
| `SIP_TRUNK_HOST` | Your SIP provider hostname | — |
| `SIP_TRUNK_USERNAME` | SIP auth username | — |
| `SIP_TRUNK_PASSWORD` | SIP auth password | — |
| `SIP_TRUNK_DID` | Your phone number (E.164) | — |
| `GATEWAY_SIP_PORT` | Port for device registration | `5080` |
| `GATEWAY_SIP_PORT` | Port for device registration | `5060` |
| `SPEACHES_URL` | Speaches/Whisper STT endpoint | `http://localhost:22070` |
| `LLM_BASE_URL` | OpenAI-compatible LLM endpoint | `http://localhost:11434/v1` |
| `LLM_MODEL` | Model name for IVR analysis | `llama3` |
@@ -391,11 +414,11 @@ All configuration is via environment variables (see `.env.example`):
## Tech Stack
- **Python 3.12+** + **asyncio** — Single-process async architecture
- **Python 3.12+** + **asyncio** — Single process. Not single-threaded: the SIP stacks run their own event loops on separate OS threads, crossed only through defined funnels ([docs/architecture.md](docs/architecture.md#threading-model))
- **FastAPI** — REST API + WebSocket server
- **SvelteKit** — Dashboard UI (built static, served by FastAPI at `/`)
- **Sippy B2BUA** — SIP call control and DTMF
- **PJSUA2** — Media pipeline, conference bridge, recording, WAV playback
- **Sippy B2BUA** — SIP call control and DTMF (`SIP_ENGINE=sippy`, signalling only)
- **PJSUA2** — Call control + media: conference bridge, capture ports, recording, WAV playback (`SIP_ENGINE=pjsua2`)
- **Speaches** (Whisper) — Speech-to-text
- **Rhema** (Kokoro) — Text-to-speech (OpenAI-compatible `/v1/audio/speech`)
- **Ollama / vLLM / OpenAI** — LLM for IVR menu analysis and receptionist intent capture
@@ -408,6 +431,9 @@ Full documentation is in [`/docs`](docs/README.md):
- [Architecture](docs/architecture.md) — System design, data flow, threading model
- [Core Engine](docs/core-engine.md) — SIP engine, media pipeline, call manager, event bus
- [Dial Plan](docs/dial-plan.md) — Number normalisation and the emergency-number guard
- [PJSUA2 Build](docs/pjsua2-build.md) — Building the bindings (not pip-installable)
- [Asterisk Lab](tests/lab/README.md) — The fake PSTN used for media validation
- [Hold Slayer Service](docs/hold-slayer-service.md) — IVR navigation, hold detection, human detection
- [Audio Classifier](docs/audio-classifier.md) — Waveform analysis, feature extraction, classification
- [Services](docs/services.md) — LLM client, transcription, recording, analytics, notifications
@@ -443,15 +469,15 @@ Full documentation is in [`/docs`](docs/README.md):
- [x] Notification service (WebSocket + SMS)
- [x] Service wiring in main.py lifespan
### Phase 4: Production Hardening 🚧
### Phase 4: Production Hardening ✅
- [x] Alembic database migrations (baseline + upgrade-on-boot)
- [x] API authentication — Casdoor SSO (browser JWT) + owner-minted PATs, owner-only across REST/WS/MCP
- [x] Emergency-number guard + concurrent-call cap on outbound calls
- [ ] Rate limiting on API endpoints
- [ ] Structured JSON logging
- [x] Rate limiting on the unauthenticated `/auth/*` edge (everything else is owner-gated; see [core/rate_limit.py](core/rate_limit.py))
- [x] Structured JSON logging (`LOG_FORMAT=json`, uvicorn access log included)
- [x] Honest /health — engine mode, DB ping, trunk registration, STT/TTS availability
- [ ] Graceful degradation (classifier works without STT, etc.)
- [x] Graceful degradation — a down STT/LLM/TTS degrades the call and publishes an `ERROR` event naming the service, rather than aborting it
- [x] Docker Compose (Hold Slayer + PostgreSQL)
### Phase 5: Additional Services 🚧

22
api/auth.py
View File

@@ -15,15 +15,25 @@ non-owner sees an "access denied" screen instead of a bare 401.
import secrets
from urllib.parse import urlencode
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from auth import get_sdk, is_owner, resolve_from_header_or_query
from config import get_settings
from core.rate_limit import rate_limit
from db.database import session_scope
router = APIRouter(prefix="/auth", tags=["auth"])
# These are the only routes that must answer before an identity exists, so they
# are the only ones worth limiting — everything else is already owner-gated.
# `/callback` and `/refresh-callback` each trigger an outbound token exchange
# with Casdoor, and `/me` opens a DB session per call; all three are
# unauthenticated work an attacker controls. See core/rate_limit.py.
_limit_callback = [Depends(rate_limit("auth:callback", limit=10))]
_limit_me = [Depends(rate_limit("auth:me"))]
_limit_redirect = [Depends(rate_limit("auth:redirect"))]
def _build_casdoor_auth_url(
callback: str,
@@ -50,7 +60,7 @@ def _build_casdoor_auth_url(
return f"{c.endpoint.rstrip('/')}/login/oauth/authorize?{urlencode(params)}"
@router.get("/login")
@router.get("/login", dependencies=_limit_redirect)
async def login(request: Request, redirect_uri: str = Query(None)):
"""Redirect the browser to the Casdoor authorization page.
@@ -64,7 +74,7 @@ async def login(request: Request, redirect_uri: str = Query(None)):
return RedirectResponse(url=_build_casdoor_auth_url(callback))
@router.get("/callback")
@router.get("/callback", dependencies=_limit_callback)
async def callback(
code: str = Query(...),
state: str = Query(None),
@@ -89,7 +99,7 @@ async def callback(
return RedirectResponse(url=f"/#token={access_token}")
@router.get("/silent-refresh")
@router.get("/silent-refresh", dependencies=_limit_redirect)
async def silent_refresh(request: Request):
"""Start a silent token refresh via hidden iframe (``prompt=none``).
@@ -104,7 +114,7 @@ async def silent_refresh(request: Request):
return RedirectResponse(url=_build_casdoor_auth_url(callback, prompt="none"))
@router.get("/refresh-callback")
@router.get("/refresh-callback", dependencies=_limit_callback)
async def refresh_callback(
code: str = Query(None),
error: str = Query(None),
@@ -140,7 +150,7 @@ async def refresh_callback(
)
@router.get("/me")
@router.get("/me", dependencies=_limit_me)
async def me(request: Request):
"""Return the current authenticated user's profile + ``is_owner``.

11
config.py
View File

@@ -145,6 +145,11 @@ class Settings(BaseSettings):
debug: bool = False
log_level: str = "info"
# Log rendering: "text" (human-readable, for a terminal) or "json" (one
# object per line, for Loki). Text is the default so local dev is readable;
# the container sets LOG_FORMAT=json. See core/logging_config.py.
log_format: str = "text"
# Auth — Casdoor SSO for the browser + owner-minted PATs for MCP/CLI,
# gated to a single owner. `owner_name` is the Casdoor username that owns
# this gateway (everyone else gets 403). `public_base_url` seeds the OAuth
@@ -162,6 +167,12 @@ class Settings(BaseSettings):
# silently degrading to a gateway that can't place real calls.
use_mock_sip: bool = False
# SIP stack: "sippy" (signalling only — the classifier gets no audio) or
# "pjsua2" (call control + media, the only path where audio reaches the
# classifier). Opt-in while the PJSUA2 engine is proven against the lab;
# see docs/architecture.md → "Media plane: why PJSUA2 places the call".
sip_engine: str = "sippy"
# Notifications
notify_sms_number: str = ""

20
core/gateway.py
View File

@@ -55,6 +55,26 @@ def build_sip_engine(
"for development without a trunk."
)
if settings.sip_engine.lower() == "pjsua2":
from core.pjsua_engine import PJSUAEngine
logger.info("📞 SIP engine: PJSUA2 (call control + media)")
return PJSUAEngine(
sip_address=gw_sip.host,
sip_port=gw_sip.port,
trunk_host=trunk.host,
trunk_port=trunk.port,
trunk_username=trunk.username,
trunk_password=trunk.password.get_secret_value(),
trunk_transport=trunk.transport,
domain=gw_sip.domain,
did=trunk.did,
media_pipeline=media_pipeline,
on_leg_state_change=on_leg_state_change,
on_device_registered=on_device_registered,
on_incoming_call=on_incoming_call,
)
return SippyEngine(
sip_address=gw_sip.host,
sip_port=gw_sip.port,

175
core/logging_config.py Normal file
View File

@@ -0,0 +1,175 @@
"""
Logging configuration — human-readable text or structured JSON.
Hold Slayer's logs are shipped to Loki by the host's Alloy agent, which reads
the container's stdout. Text logs arrive there as an opaque blob: filtering on
a status code or a call ID means regex over a formatted string. JSON lines
arrive as queryable fields.
Two things about this are less obvious than they look, and both are the reason
this module exists instead of a `format=` argument on `basicConfig`:
1. **Uvicorn brings its own handlers.** `uvicorn.config.LOGGING_CONFIG` attaches
a `StreamHandler` to `uvicorn` and `uvicorn.access` with `propagate: False`,
so those records never reach the root logger's formatter. Configuring only
the root would leave the access log — the highest-volume, most useful stream
— as plain colourised text next to our JSON. `configure_logging` reaches into
those two loggers explicitly.
2. **The access record's payload is in `record.args`, not the message.** Uvicorn
logs access lines as a 5-tuple `(client_addr, method, full_path, http_version,
status_code)` and lets its `AccessFormatter` interpolate them. Formatting the
message would throw that structure away and force Loki to parse it back out,
so `JSONFormatter` unpacks the tuple into real fields.
Text mode stays the default: it is what a developer wants on a terminal, and a
JSON-only logger makes local debugging worse. Production opts in via
`LOG_FORMAT=json`.
"""
import datetime as _dt
import json
import logging
import sys
from typing import Any
# LogRecord attributes that are either already represented in our output or are
# formatting machinery. Anything on a record that is *not* here is treated as a
# caller-supplied `extra=` field and promoted into the JSON object.
_RESERVED = frozenset(
{
"args",
"asctime",
# Uvicorn passes an ANSI-colourised duplicate of the message as
# `extra={"color_message": ...}` for its own formatter to prefer. Left
# unfiltered, generic extra-promotion copies escape codes into every
# startup line — the same unreadable-in-Grafana problem as the lab's
# Asterisk logs.
"color_message",
"created",
"exc_info",
"exc_text",
"filename",
"funcName",
"levelname",
"levelno",
"lineno",
"module",
"msecs",
"message",
"msg",
"name",
"pathname",
"process",
"processName",
"relativeCreated",
"stack_info",
"taskName",
"thread",
"threadName",
}
)
# Uvicorn's own access-log tuple, in order.
_ACCESS_FIELDS = ("client_addr", "method", "path", "http_version", "status_code")
TEXT_FORMAT = "%(asctime)s | %(levelname)-7s | %(name)s | %(message)s"
class JSONFormatter(logging.Formatter):
"""Render a LogRecord as a single-line JSON object."""
def format(self, record: logging.LogRecord) -> str:
payload: dict[str, Any] = {
# RFC 3339 in UTC. `logging`'s default asctime is local-time and
# date-less, which makes correlating with Loki's own timestamps
# unnecessarily hard.
"ts": _dt.datetime.fromtimestamp(record.created, tz=_dt.UTC).isoformat(
timespec="milliseconds"
),
"level": record.levelname,
"logger": record.name,
}
if record.name == "uvicorn.access" and isinstance(record.args, tuple):
payload.update(_access_fields(record))
else:
payload["msg"] = record.getMessage()
# Thread name matters here in a way it doesn't in a single-context app:
# this process runs the asyncio loop, the Sippy ED thread, and PJSUA2
# worker threads, and "which context logged this" is usually the first
# question when debugging a call.
if record.threadName and record.threadName != "MainThread":
payload["thread"] = record.threadName
if record.exc_info:
payload["exc"] = self.formatException(record.exc_info)
if record.stack_info:
payload["stack"] = self.formatStack(record.stack_info)
for key, value in record.__dict__.items():
if key not in _RESERVED and not key.startswith("_"):
payload[key] = _safe(value)
return json.dumps(payload, default=str, separators=(",", ":"))
def _access_fields(record: logging.LogRecord) -> dict[str, Any]:
"""Unpack uvicorn's access-log arg tuple into named fields.
Falls back to the interpolated message if uvicorn ever changes the tuple's
shape — a log line with a slightly wrong shape beats an exception inside the
logging path taking out the request.
"""
args = record.args
if not isinstance(args, tuple) or len(args) != len(_ACCESS_FIELDS):
return {"msg": record.getMessage()}
fields: dict[str, Any] = dict(zip(_ACCESS_FIELDS, args))
try:
fields["status_code"] = int(fields["status_code"])
except (TypeError, ValueError):
pass
return fields
def _safe(value: Any) -> Any:
"""Keep JSON-native types; stringify everything else."""
if isinstance(value, (str, int, float, bool, type(None))):
return value
return str(value)
def configure_logging(log_format: str, log_level: str) -> None:
"""Install the root and uvicorn log handlers.
Idempotent: existing root handlers are removed first, so calling this after
uvicorn has configured itself replaces its formatting rather than adding a
second stream (which is how you get every line twice).
"""
level = getattr(logging, log_level.upper(), logging.INFO)
use_json = log_format.lower() == "json"
formatter: logging.Formatter = (
JSONFormatter() if use_json else logging.Formatter(TEXT_FORMAT, datefmt="%H:%M:%S")
)
handler = logging.StreamHandler(stream=sys.stdout)
handler.setFormatter(formatter)
root = logging.getLogger()
for existing in root.handlers[:]:
root.removeHandler(existing)
root.addHandler(handler)
root.setLevel(level)
# Uvicorn sets `propagate = False` on these and attaches its own colourised
# handlers, so they must be redirected explicitly or they bypass everything
# above. Clearing the handlers and re-enabling propagation routes them
# through the root handler like any other logger.
for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
uv = logging.getLogger(name)
uv.handlers.clear()
uv.propagate = True
uv.setLevel(level)

218
core/media_pipeline.py
View File

@@ -20,6 +20,7 @@ PJSUA2 runs in its own thread with a dedicated Endpoint.
"""
import asyncio
import gc
import logging
import threading
from collections.abc import AsyncIterator
@@ -98,6 +99,72 @@ class AudioTap:
self._active = False
def make_capture_port(stream_id: str, sample_rate: int, channels: int, frame_ms: int):
"""Build a PJSUA2 media port that forks conference audio into taps.
Defined as a factory rather than a module-level class because
``pj.AudioMediaPort`` can only be subclassed once ``pjsua2`` imports —
and the whole pipeline degrades to stub mode when it doesn't.
The returned port is a *sink*: the conference bridge transmits into it,
and every frame is copied to each registered tap. Returns ``None`` when
pjsua2 is unavailable.
"""
try:
import pjsua2 as pj
except ImportError:
return None
class _CapturePort(pj.AudioMediaPort):
"""Receives conference-bridge frames and fans them out to taps.
``onFrameReceived`` is called on a **PJSUA2 worker thread** — a third
execution context alongside the asyncio loop and the Sippy ED thread.
It must touch nothing but ``AudioTap.feed``, which is explicitly
thread-safe (it hops to the owning loop via ``call_soon_threadsafe``).
Reaching into pipeline state, the event bus, or a Sippy object from
here would be a data race.
"""
def __init__(self, stream_id: str):
super().__init__()
self.stream_id = stream_id
self.taps: list[AudioTap] = []
self._logged_error = False
def onFrameReceived(self, frame): # noqa: N802 — PJSUA2 C++ callback name
try:
if not self.taps or frame.size <= 0:
return
# frame.buf is a SWIG ByteVector of signed chars; the tap
# contract is raw little-endian 16-bit PCM.
pcm = bytes(bytearray(b & 0xFF for b in frame.buf))
for tap in self.taps:
tap.feed(pcm)
except Exception as e:
# An exception escaping into PJSUA2's C++ callback would tear
# down the worker thread and silently kill media for every
# call. Log once per port rather than on every 20ms frame.
if not self._logged_error:
self._logged_error = True
logger.error(
f" Audio capture failed for {self.stream_id}: {e}",
exc_info=True,
)
fmt = pj.MediaFormatAudio()
fmt.init(
pj.PJMEDIA_FORMAT_L16,
sample_rate,
channels,
frame_ms * 1000, # frameTimeUsec
16, # bitsPerSample
)
port = _CapturePort(stream_id)
port.createPort(f"tap-{stream_id}", fmt)
return port
# ================================================================
# Stream Entry — tracks a single media stream in the pipeline
# ================================================================
@@ -112,6 +179,8 @@ class MediaStream:
self.codec = codec
self.conf_port: Optional[int] = None # PJSUA2 conference bridge port ID
self.transport = None # PJSUA2 SipTransport
self.media = None # PJSUA2 AudioMedia for this stream
self.capture_port = None # Shared _CapturePort feeding this stream's taps
self.rtp_port: Optional[int] = None # Local RTP listen port
self.taps: list[AudioTap] = []
self.recorder = None # PJSUA2 AudioMediaRecorder
@@ -143,10 +212,11 @@ class MediaPipeline:
pipeline = MediaPipeline()
await pipeline.start()
# Add a stream for a call leg
port = pipeline.add_remote_stream("leg_1", "10.0.0.1", 20000, "PCMU")
# Media arrives from the SIP engine's onCallMediaState callback
# (PJSUA2 only surfaces RTP media for a call it owns):
# pipeline.attach_call_media("leg_1", call.getAudioMedia(i))
# Tap audio for analysis
# Tap audio for analysis — safe before or after media comes up
tap = pipeline.create_tap("leg_1")
async for frame in tap.stream():
classify(frame)
@@ -173,6 +243,7 @@ class MediaPipeline:
self._next_rtp_port = rtp_start_port
self._sample_rate = sample_rate
self._channels = channels
self._frame_ms = 20 # Must match medConfig.audioFramePtime below
self._null_audio = null_audio # Use null audio device (no sound card needed)
# State
@@ -255,11 +326,17 @@ class MediaPipeline:
tap.close()
self._taps.clear()
# Remove all streams
# Remove all streams (this releases their capture ports)
for stream_id in list(self._streams.keys()):
self.remove_stream(stream_id)
# Destroy PJSUA2 endpoint
# Destroy PJSUA2 endpoint. Every media port must be collected first:
# a port finalised after libDestroy() runs pjmedia_conf_remove_port
# against a freed conference bridge and aborts the process. Dropping
# the last Python reference is not enough on its own — force the
# collection here rather than leaving it to interpreter exit.
gc.collect()
if self._endpoint:
try:
self._endpoint.libDestroy()
@@ -270,6 +347,15 @@ class MediaPipeline:
self._ready = False
logger.info("🎵 PJSUA2 media pipeline stopped")
@property
def endpoint(self):
"""The PJSUA2 Endpoint, or None in stub mode.
PJSUA2 permits exactly one Endpoint per process, so the pipeline
creates it and the SIP engine borrows it rather than making a second.
"""
return self._endpoint
@property
def is_ready(self) -> bool:
return self._ready
@@ -291,50 +377,51 @@ class MediaPipeline:
# Stream Management
# ================================================================
def add_remote_stream(
self, stream_id: str, remote_host: str, remote_port: int, codec: str = "PCMU"
) -> Optional[int]:
def attach_call_media(self, stream_id: str, audio_media) -> Optional[int]:
"""Register a call's live ``AudioMedia`` with the pipeline.
Called from ``onCallMediaState`` on a PJSUA2 worker thread, which is
the only place PJSUA2 surfaces RTP-backed media. Any tap created
before this point is attached now; taps created later find the media
already present.
There is deliberately no ``add_remote_stream(host, port)`` counterpart:
PJSUA2 has no standalone RTP media object, so media can only arrive
from a call PJSUA2 owns. See ``docs/architecture.md``.
"""
Add a remote RTP stream to the conference bridge.
stream = self._streams.get(stream_id)
if stream is None:
stream = MediaStream(stream_id, "", 0)
self._streams[stream_id] = stream
Creates a PJSUA2 transport and media port for the remote
party's RTP stream, connecting it to the conference bridge.
stream.media = audio_media
try:
stream.conf_port = audio_media.getPortId()
except Exception:
stream.conf_port = None
Args:
stream_id: Unique ID (typically the SIP leg ID)
remote_host: Remote RTP host
remote_port: Remote RTP port
codec: Audio codec (PCMU, PCMA, G729)
# Wire up taps that were requested before media came up.
pending = self._taps.get(stream_id, [])
if pending and stream.capture_port is None:
port = make_capture_port(
stream_id, self._sample_rate, self._channels, self._frame_ms
)
if port is not None:
try:
audio_media.startTransmit(port)
stream.capture_port = port
port.taps.extend(pending)
logger.info(
f" 🎤 Audio tap attached for {stream_id} "
f"({len(pending)} waiting)"
)
except Exception as e:
logger.error(
f" Failed to attach capture port for {stream_id}: {e}",
exc_info=True,
)
Returns:
Conference bridge port ID, or None if PJSUA2 not available
"""
stream = MediaStream(stream_id, remote_host, remote_port, codec)
stream.rtp_port = self.allocate_rtp_port(stream_id)
if self._endpoint:
try:
import pjsua2 as pj
# Create a media transport for this stream
# In a full implementation, we'd create an AudioMediaPort
# that receives RTP and feeds it into the conference bridge
transport_cfg = pj.TransportConfig()
transport_cfg.port = stream.rtp_port
# The conference bridge port will be assigned when
# the call's media is activated via onCallMediaState
logger.info(
f" 📡 Added stream {stream_id}: "
f"local={stream.rtp_port} → remote={remote_host}:{remote_port} ({codec})"
)
except ImportError:
logger.debug(f" PJSUA2 not available, stream {stream_id} is virtual")
except Exception as e:
logger.error(f" Failed to add stream {stream_id}: {e}")
self._streams[stream_id] = stream
logger.info(f" 📡 Media attached for {stream_id} (conf port {stream.conf_port})")
return stream.conf_port
def remove_stream(self, stream_id: str) -> None:
@@ -350,6 +437,19 @@ class MediaPipeline:
tap.close()
self._taps.pop(stream_id, None)
# Release the capture port while the conference bridge still exists.
# A port garbage-collected after Endpoint.libDestroy() calls
# pjmedia_conf_remove_port against a freed bridge and aborts the
# process on a native assertion — a hard crash, not an exception.
if stream.capture_port is not None:
try:
if stream.media is not None:
stream.media.stopTransmit(stream.capture_port)
except Exception as e:
logger.debug(f" stopTransmit failed for {stream_id}: {e}")
stream.capture_port.taps.clear()
stream.capture_port = None
# Stop recording
if stream.recorder:
try:
@@ -426,16 +526,28 @@ class MediaPipeline:
self._taps[stream_id] = []
self._taps[stream_id].append(tap)
if self._endpoint and stream and stream.conf_port is not None:
if self._endpoint and stream and stream.media is not None:
try:
import pjsua2 as pj
# Create an AudioMediaPort that captures frames
# and feeds them to the tap
# In PJSUA2, we'd subclass AudioMediaPort and implement
# onFrameReceived to call tap.feed(frame_data)
logger.info(f" 🎤 Audio tap created for {stream_id} (PJSUA2)")
# One capture port per stream, shared by every tap on it:
# the bridge would otherwise mix each additional port back
# into the conference and the call would echo.
if stream.capture_port is None:
port = make_capture_port(
stream_id, self._sample_rate, self._channels, self._frame_ms
)
if port is not None:
# The stream's media transmits into the capture port,
# not the reverse — the port is a sink.
stream.media.startTransmit(port)
stream.capture_port = port
logger.info(f" 🎤 Audio tap created for {stream_id} (PJSUA2)")
if stream.capture_port is not None:
stream.capture_port.taps.append(tap)
except Exception as e:
logger.error(f" Failed to create PJSUA2 tap for {stream_id}: {e}")
logger.error(
f" Failed to create PJSUA2 tap for {stream_id}: {e}", exc_info=True
)
else:
logger.info(f" 🎤 Audio tap created for {stream_id} (virtual)")

488
core/pjsua_engine.py Normal file
View File

@@ -0,0 +1,488 @@
"""
PJSUA2 SIP engine — call control *and* media in one library.
Why this exists
---------------
The gateway originally signalled with Sippy and expected PJSUA2 to carry
media. That cannot work: **PJSUA2 exposes no standalone RTP media object**.
Every ``AudioMedia`` subclass in the Python bindings is a file player,
recorder, tone generator or capture port, and RTP is reachable only through
``pj.Call.getAudioMedia()`` — on a dialog PJSUA2 itself owns. A design where
another stack owns the dialog can never obtain media from PJSUA2, so audio
never reached the classifier.
Owning the dialog is the price of owning the media, so this engine places the
call. See ``docs/architecture.md`` → "Media plane: why PJSUA2 places the call".
Safety
------
This engine is *only* reached through ``gateway.make_call``, which refuses
emergency numbers and enforces the concurrency cap **before** any SIP action.
Nothing here may be given a second dial path that bypasses those checks.
Threading
---------
Three execution contexts, as elsewhere in the codebase:
* the **asyncio loop** owns legs, the event bus, and the call manager;
* **PJSUA2 worker threads** run every ``on*`` callback below;
* (the Sippy ED thread is not involved — this engine replaces it.)
PJSUA2 callbacks cross to the loop through exactly one funnel,
``_post_from_pj`` → ``run_coroutine_threadsafe``. A callback must never touch
loop-owned state directly. Any thread PJSUA2 did not create must call
``libRegisterThread`` before touching a PJSUA2 object, which
``_ensure_registered`` handles.
"""
import asyncio
import gc
import logging
import threading
import uuid
from collections.abc import Callable
from core.sip_engine import SIPEngine
from models.device import Device
logger = logging.getLogger(__name__)
class PJSUAEngine(SIPEngine):
"""SIP engine backed by PJSUA2 for both signalling and media."""
def __init__(
self,
sip_address: str = "0.0.0.0",
sip_port: int = 5060,
trunk_host: str = "",
trunk_port: int = 5060,
trunk_username: str = "",
trunk_password: str = "",
trunk_transport: str = "udp",
domain: str = "gateway.local",
did: str = "",
media_pipeline=None,
on_leg_state_change: Callable | None = None,
on_device_registered: Callable | None = None,
on_incoming_call: Callable | None = None,
):
self._sip_address = sip_address
self._sip_port = sip_port
self._trunk_host = trunk_host
self._trunk_port = trunk_port
self._trunk_username = trunk_username
self._trunk_password = trunk_password
self._trunk_transport = trunk_transport
self._domain = domain
self._did = did
# The media pipeline owns the PJSUA2 Endpoint; this engine borrows it
# rather than creating a second one (PJSUA2 permits only one).
self.media_pipeline = media_pipeline
self._on_leg_state_change = on_leg_state_change
self._on_device_registered = on_device_registered
self._on_incoming_call = on_incoming_call
self._loop: asyncio.AbstractEventLoop | None = None
self._ready = False
self._account = None
self._trunk_registered = False
self._trunk_reason = "not started"
# PJSUA2-thread-owned: maps leg_id → pj.Call. Only touched from a
# PJSUA2 callback or a method that has registered itself first.
self._calls: dict[str, object] = {}
self._lock = threading.Lock()
# ================================================================
# Thread boundary
# ================================================================
def _post_from_pj(self, coro) -> None:
"""Schedule loop work from a PJSUA2 worker thread. The one funnel."""
if self._loop is None:
return
asyncio.run_coroutine_threadsafe(coro, self._loop)
def _ensure_registered(self) -> None:
"""Register the calling thread with PJSUA2 if it isn't already.
PJSUA2 aborts when a thread it does not know touches its objects.
Calls made from the asyncio loop (hangup, DTMF) hit this.
"""
try:
import pjsua2 as pj
ep = pj.Endpoint.instance()
if not ep.libIsThreadRegistered():
ep.libRegisterThread(threading.current_thread().name)
except Exception as e: # pragma: no cover - defensive
logger.debug(f" thread registration skipped: {e}")
async def _emit_leg_state(self, leg_id: str, state: str) -> None:
"""Deliver a leg-state change on the loop."""
if self._on_leg_state_change is None:
return
result = self._on_leg_state_change(leg_id, state)
if asyncio.iscoroutine(result):
await result
# ================================================================
# Lifecycle
# ================================================================
async def start(self) -> None:
"""Create the SIP transport and register with the trunk."""
self._loop = asyncio.get_running_loop()
logger.info("🔌 Starting PJSUA2 SIP engine...")
if self.media_pipeline is None or not self.media_pipeline.endpoint:
raise RuntimeError(
"PJSUAEngine requires a started MediaPipeline — PJSUA2 allows "
"only one Endpoint, so the pipeline owns it and the engine "
"borrows it."
)
import pjsua2 as pj
ep = self.media_pipeline.endpoint
transport_cfg = pj.TransportConfig()
transport_cfg.port = self._sip_port
if self._sip_address and self._sip_address != "0.0.0.0":
transport_cfg.boundAddress = self._sip_address
tp_type = (
pj.PJSIP_TRANSPORT_TCP
if self._trunk_transport.lower() == "tcp"
else pj.PJSIP_TRANSPORT_UDP
)
ep.transportCreate(tp_type, transport_cfg)
self._create_account(ep)
self._ready = True
logger.info(f"🔌 PJSUA2 SIP engine ready on {self._sip_address}:{self._sip_port}")
def _create_account(self, ep) -> None:
"""Build the account — registered to the trunk, or local-only."""
import pjsua2 as pj
engine = self
class _Account(pj.Account):
def onRegState(self, prm): # noqa: N802 — PJSUA2 callback name
try:
info = self.getInfo()
engine._trunk_registered = bool(info.regIsActive)
engine._trunk_reason = f"{prm.code} {prm.reason}".strip()
if info.regIsActive:
logger.info(" ✅ Trunk registration accepted")
else:
# A rejected REGISTER must not read as "registered":
# /health treats a registered trunk as a condition of
# being healthy.
logger.error(
f" ❌ Trunk registration failed: {engine._trunk_reason}"
)
except Exception as e:
logger.error(f" onRegState error: {e}", exc_info=True)
def onIncomingCall(self, prm): # noqa: N802 — PJSUA2 callback name
try:
engine._handle_incoming(self, prm.callId)
except Exception as e:
logger.error(f" onIncomingCall error: {e}", exc_info=True)
acc_cfg = pj.AccountConfig()
if self._trunk_host:
acc_cfg.idUri = f"sip:{self._trunk_username}@{self._trunk_host}"
acc_cfg.regConfig.registrarUri = f"sip:{self._trunk_host}:{self._trunk_port}"
cred = pj.AuthCredInfo(
"digest", "*", self._trunk_username, 0, self._trunk_password
)
acc_cfg.sipConfig.authCreds.append(cred)
else:
# No trunk configured: a local-only account still lets devices
# register and inbound calls arrive.
acc_cfg.idUri = f"sip:gateway@{self._domain}"
self._trunk_reason = "No SIP trunk configured"
self._account = _Account()
self._account.create(acc_cfg)
if self._trunk_host:
logger.info(f" Registering with trunk: {self._trunk_host}:{self._trunk_port}")
async def stop(self) -> None:
"""Hang up everything and drop the account."""
logger.info("🔌 Stopping PJSUA2 SIP engine...")
self._ready = False
self._ensure_registered()
had_calls = bool(self._calls)
for leg_id in list(self._calls.keys()):
try:
await self.hangup(leg_id)
except Exception as e:
logger.debug(f" hangup during shutdown failed for {leg_id}: {e}")
# hangup() only queues the BYE. Give PJSUA2 a moment to send it and
# tear the media down, or the account is deleted with a call still
# active ("deleting account 0 while call 0 is still active") and the
# far end is left waiting on a dialog nobody closed.
if had_calls:
await asyncio.sleep(0.5)
# Drop every PJSUA2 object before the pipeline destroys the endpoint.
# A Call or Account finalised after libDestroy() aborts the process on
# a native assertion, exactly as a stray media port does — and a Call
# still alive keeps delivering callbacks into a half-torn-down
# interpreter. Dropping the last reference is not enough on its own,
# so force the collection here.
with self._lock:
self._calls.clear()
self._account = None
gc.collect()
logger.info("🔌 PJSUA2 SIP engine stopped")
async def is_ready(self) -> bool:
return self._ready
# ================================================================
# Calls
# ================================================================
def _make_call_class(self):
"""Build the pj.Call subclass bound to this engine."""
import pjsua2 as pj
engine = self
class _Call(pj.Call):
def __init__(self, acc, leg_id: str, call_id=pj.PJSUA_INVALID_ID):
super().__init__(acc, call_id)
self.leg_id = leg_id
def onCallState(self, prm): # noqa: N802 — PJSUA2 callback name
# PJSUA2 keeps delivering callbacks while the interpreter is
# tearing down, when module globals may already be cleared —
# hence the local alias and the bare except. A raise here
# escapes into C++ and takes the worker thread with it.
state_map = _STATE_MAP
try:
info = self.getInfo()
state = state_map.get(info.state)
if state is None:
return
if state == "terminated":
engine._forget_call(self.leg_id)
engine._post_from_pj(engine._emit_leg_state(self.leg_id, state))
except Exception:
try:
logger.error(" onCallState error", exc_info=True)
except Exception:
pass
def onCallMediaState(self, prm): # noqa: N802 — PJSUA2 callback name
"""Media is up — hand the audio to the pipeline.
This is the callback the whole refactor exists for: it is the
only place PJSUA2 surfaces an RTP-backed AudioMedia.
"""
try:
info = self.getInfo()
for i, mi in enumerate(info.media):
if (
mi.type == pj.PJMEDIA_TYPE_AUDIO
and mi.status == pj.PJSUA_CALL_MEDIA_ACTIVE
):
engine._attach_media(self.leg_id, self.getAudioMedia(i))
break
except Exception:
try:
logger.error(" onCallMediaState error", exc_info=True)
except Exception:
pass
return _Call
def _attach_media(self, leg_id: str, audio_media) -> None:
"""Register a live AudioMedia with the pipeline (PJSUA2 thread)."""
if self.media_pipeline is None:
return
try:
self.media_pipeline.attach_call_media(leg_id, audio_media)
logger.info(f" 🎵 Media active for {leg_id}")
except Exception as e:
logger.error(f" Failed to attach media for {leg_id}: {e}", exc_info=True)
def _forget_call(self, leg_id: str) -> None:
with self._lock:
self._calls.pop(leg_id, None)
if self.media_pipeline is not None:
try:
self.media_pipeline.remove_stream(leg_id)
except Exception as e:
logger.debug(f" stream cleanup failed for {leg_id}: {e}")
async def make_call(self, number: str, caller_id: str | None = None) -> str:
"""Place an outbound call. Reached only via gateway.make_call."""
if not self._ready:
raise RuntimeError("SIP engine not ready")
import pjsua2 as pj
leg_id = f"leg_{uuid.uuid4().hex[:12]}"
target = (
f"sip:{number}@{self._trunk_host}:{self._trunk_port}"
if self._trunk_host
else f"sip:{number}@{self._domain}"
)
logger.info(f"📞 Placing call to {target} (leg: {leg_id})")
self._ensure_registered()
call_cls = self._make_call_class()
call = call_cls(self._account, leg_id)
prm = pj.CallOpParam(True)
call.makeCall(target, prm)
with self._lock:
self._calls[leg_id] = call
return leg_id
async def hangup(self, call_leg_id: str) -> None:
import pjsua2 as pj
with self._lock:
call = self._calls.get(call_leg_id)
if call is None:
return
self._ensure_registered()
try:
call.hangup(pj.CallOpParam(True))
except Exception as e:
logger.debug(f" hangup failed for {call_leg_id}: {e}")
self._forget_call(call_leg_id)
async def send_dtmf(self, call_leg_id: str, digits: str) -> None:
"""Send DTMF as RFC 2833 — the in-band path a real IVR expects."""
with self._lock:
call = self._calls.get(call_leg_id)
if call is None:
logger.warning(f" send_dtmf: no call for {call_leg_id}")
return
self._ensure_registered()
call.dialDtmf(digits)
logger.info(f" Sent DTMF '{digits}' on {call_leg_id}")
async def call_device(self, device: Device) -> str:
"""Ring a registered device (transfer target)."""
if not self._ready:
raise RuntimeError("SIP engine not ready")
import pjsua2 as pj
leg_id = f"leg_{uuid.uuid4().hex[:12]}"
target = device.sip_uri or f"sip:{device.id}@{self._domain}"
logger.info(f"📞 Ringing device {device.id} at {target} (leg: {leg_id})")
self._ensure_registered()
call_cls = self._make_call_class()
call = call_cls(self._account, leg_id)
call.makeCall(target, pj.CallOpParam(True))
with self._lock:
self._calls[leg_id] = call
return leg_id
def _handle_incoming(self, account, call_id) -> None:
"""Inbound INVITE (PJSUA2 thread) — answer and hand to the receptionist."""
import pjsua2 as pj
leg_id = f"leg_{uuid.uuid4().hex[:12]}"
call_cls = self._make_call_class()
call = call_cls(account, leg_id, call_id)
try:
info = call.getInfo()
remote = info.remoteUri
except Exception:
remote = "unknown"
with self._lock:
self._calls[leg_id] = call
call.answer(pj.CallOpParam(True))
logger.info(f"📞 Inbound call {leg_id} from {remote}")
if self._on_incoming_call is not None:
result = self._on_incoming_call(leg_id, remote)
if asyncio.iscoroutine(result):
self._post_from_pj(result)
# ================================================================
# Bridging
# ================================================================
async def bridge_calls(self, leg_a: str, leg_b: str) -> str:
"""Join two legs in the conference bridge."""
bridge_id = f"bridge_{uuid.uuid4().hex[:8]}"
if self.media_pipeline is not None:
self.media_pipeline.bridge_streams(leg_a, leg_b)
logger.info(f" 🌉 Bridged {leg_a} ↔ {leg_b} ({bridge_id})")
return bridge_id
async def unbridge(self, bridge_id: str) -> None:
logger.info(f" Unbridged {bridge_id}")
def get_audio_stream(self, call_leg_id: str):
if self.media_pipeline is not None:
return self.media_pipeline.get_audio_tap(call_leg_id)
return None
# ================================================================
# Status
# ================================================================
async def get_registered_devices(self) -> list[dict]:
return []
async def get_trunk_status(self) -> dict:
return {
"registered": self._trunk_registered,
"host": self._trunk_host or "not configured",
"port": self._trunk_port,
"transport": self._trunk_transport,
"username": self._trunk_username,
"reason": None if self._trunk_registered else self._trunk_reason,
}
# Populated lazily: the pjsua2 constants are unavailable until import, and
# the module must import cleanly in stub mode.
_STATE_MAP: dict = {}
def _init_state_map() -> None:
global _STATE_MAP
if _STATE_MAP:
return
try:
import pjsua2 as pj
except ImportError:
return
_STATE_MAP = {
pj.PJSIP_INV_STATE_CALLING: "trying",
pj.PJSIP_INV_STATE_EARLY: "ringing",
pj.PJSIP_INV_STATE_CONNECTING: "trying",
pj.PJSIP_INV_STATE_CONFIRMED: "connected",
pj.PJSIP_INV_STATE_DISCONNECTED: "terminated",
}
_init_state_map()

132
core/rate_limit.py Normal file
View File

@@ -0,0 +1,132 @@
"""
Rate limiting for the unauthenticated edge.
**Scope, and why it is narrow.** Every REST/WS/MCP surface is owner-only: an
unauthenticated request is rejected by `resolve_bearer`/`is_owner` before any
handler runs, and real spend control for outbound calls is
`max_concurrent_calls` in `gateway.make_call`. Blanket per-endpoint limits would
therefore mostly rate-limit the single legitimate operator. What is genuinely
exposed is the handful of `/auth/*` routes that must answer before an identity
exists — those are limited here, and nothing else.
**What this defends against.** Not credential guessing: PATs are
`secrets.token_urlsafe(32)` (256 bits) compared by SHA-256 digest, so brute
force is not a practical threat. The concern is *unauthenticated work an
attacker controls*: `/auth/callback` makes an outbound token-exchange round-trip
to Casdoor on every request, and `/auth/me` opens a DB session and runs a token
lookup. Both are free to trigger and neither is cheap to serve.
**Fixed-window, in-process, no dependency.** One operator and a single process
mean a shared counter store would be infrastructure without a purpose. The
trade-off of a fixed window is a burst of up to 2× the limit across a boundary;
that is irrelevant at these thresholds, and it costs one dict lookup with no
background task.
**Client identity is the socket peer, deliberately.** `X-Forwarded-For` is
attacker-controlled unless a trusted proxy overwrites it, and this app does not
currently establish that trust (`_public_base_url` reads forwarded headers, but
only to build URLs). Keying on a spoofable header would let one client present
as thousands and make the limiter worse than useless. Behind the estate's
reverse proxy this means the limit applies per-proxy rather than per-caller —
correct for exhaustion, and honest about what it can enforce. If per-caller
limits are ever needed, that needs an explicit trusted-proxy config, not a
silent `X-Forwarded-For` read.
"""
import time
from fastapi import HTTPException, Request
# Requests allowed per window, per client, per route. Sized to be invisible to a
# human — a dashboard load touches /auth/me once — while capping automated
# hammering.
DEFAULT_LIMIT = 30
DEFAULT_WINDOW_SECONDS = 60
# Stop the bucket store growing without bound under a spray of source addresses.
# Eviction is oldest-first and only runs when the cap is exceeded.
MAX_TRACKED_CLIENTS = 10_000
class RateLimiter:
"""Fixed-window request counter, keyed by (route, client)."""
def __init__(
self,
limit: int = DEFAULT_LIMIT,
window_seconds: int = DEFAULT_WINDOW_SECONDS,
max_clients: int = MAX_TRACKED_CLIENTS,
):
self.limit = limit
self.window_seconds = window_seconds
self.max_clients = max_clients
# key -> [window_start, count]. Insertion-ordered, which is what makes
# oldest-first eviction a cheap `next(iter(...))`.
self._buckets: dict[str, list[float]] = {}
def check(
self, key: str, limit: int | None = None, now: float | None = None
) -> tuple[bool, int]:
"""Record a hit. Returns (allowed, retry_after_seconds).
`retry_after` is 0 when allowed, and the seconds remaining in the
current window when not. `limit` overrides the instance default for
routes that want a tighter cap.
"""
now = time.monotonic() if now is None else now
effective = self.limit if limit is None else limit
bucket = self._buckets.get(key)
if bucket is None or now - bucket[0] >= self.window_seconds:
self._buckets.pop(key, None) # re-insert so ordering tracks recency
self._buckets[key] = [now, 1]
self._evict_if_needed()
return True, 0
bucket[1] += 1
if bucket[1] > effective:
remaining = self.window_seconds - (now - bucket[0])
return False, max(1, int(remaining) + 1)
return True, 0
def _evict_if_needed(self) -> None:
while len(self._buckets) > self.max_clients:
self._buckets.pop(next(iter(self._buckets)))
def reset(self) -> None:
self._buckets.clear()
def client_key(request: Request, scope: str) -> str:
"""Identify the caller for limiting purposes.
Uses the socket peer, never a forwarded header — see the module docstring.
"""
client = request.client.host if request.client else "unknown"
return f"{scope}:{client}"
_limiter = RateLimiter()
def get_limiter() -> RateLimiter:
return _limiter
def rate_limit(scope: str, limit: int | None = None):
"""FastAPI dependency factory limiting one route.
Applied per-route rather than as middleware so the authenticated surfaces —
which are already owner-gated — pay nothing.
"""
async def _dependency(request: Request) -> None:
allowed, retry_after = get_limiter().check(client_key(request, scope), limit=limit)
if not allowed:
raise HTTPException(
status_code=429,
detail="Too many requests",
headers={"Retry-After": str(retry_after)},
)
return _dependency

17
core/sippy_engine.py
View File

@@ -276,17 +276,22 @@ class SippyEngine(SIPEngine):
if state == "connected":
sdp = data.get("sdp")
if sdp and self.media_pipeline:
# Signalling only: PJSUA2 surfaces RTP media exclusively
# through a call it owns, so a Sippy-owned dialog can
# never be given media — the classifier stays deaf on this
# engine. PJSUAEngine is the media-capable path; see
# docs/architecture.md. Log the negotiated endpoint so the
# SIP exchange is still debuggable.
try:
remote_rtp = self._parse_sdp_rtp_endpoint(sdp)
if remote_rtp:
leg.media_port = self.media_pipeline.add_remote_stream(
leg.leg_id,
remote_rtp["host"],
remote_rtp["port"],
remote_rtp["codec"],
logger.info(
f" {leg.leg_id}: remote RTP "
f"{remote_rtp['host']}:{remote_rtp['port']} "
f"({remote_rtp['codec']}) — no media on this engine"
)
except Exception as e:
logger.error(f" Failed to set up media for {leg.leg_id}: {e}")
logger.error(f" Failed to parse SDP for {leg.leg_id}: {e}")
elif state == "terminated":
if self.media_pipeline and leg.media_port is not None:
try:

3
docs/README.md
View File

@@ -8,6 +8,9 @@ Comprehensive documentation for the Hold Slayer AI telephony gateway.
|----------|-------------|
| [Architecture](architecture.md) | System architecture, component diagram, data flow |
| [Core Engine](core-engine.md) | SIP engine, media pipeline, call manager, event bus |
| [Dial Plan](dial-plan.md) | Number normalisation and the emergency-number guard |
| [PJSUA2 Build](pjsua2-build.md) | Building the `pjsua2` bindings (not pip-installable) |
| [Asterisk Lab](../tests/lab/README.md) | The fake PSTN used for media validation |
| [Hold Slayer Service](hold-slayer-service.md) | IVR navigation, hold detection, human detection, transfer |
| [Audio Classifier](audio-classifier.md) | Waveform analysis, feature extraction, classification logic |
| [Services](services.md) | LLM client, transcription, recording, analytics, notifications |

11
docs/api-reference.md
View File

@@ -274,10 +274,21 @@ All errors follow a consistent format:
| Status Code | Meaning |
|-------------|---------|
| `400` | Bad request (invalid parameters) |
| `401` | Not authenticated (missing or invalid bearer token) |
| `403` | Authenticated but not the owner |
| `404` | Resource not found (call, flow, device) |
| `409` | Conflict (call already ended, device already registered) |
| `429` | Rate limited — `/auth/*` routes only. Carries `Retry-After` (seconds) |
| `500` | Internal server error |
`429` applies solely to the unauthenticated `/auth/*` edge: those routes must
answer before an identity exists, and each does real work (`/auth/callback`
makes an outbound token exchange with Casdoor, `/auth/me` opens a DB session).
The owner-gated API is not rate limited — it is already restricted to a single
operator. Limits are per client, per route, in a fixed 60-second window; the
client is the **socket peer**, so behind a reverse proxy the limit applies
per-proxy. See [core/rate_limit.py](../core/rate_limit.py).
## WebSocket
### Event Stream

147
docs/architecture.md
View File

@@ -1,6 +1,16 @@
# Architecture
Hold Slayer is a single-process async Python application built on FastAPI. It acts as an intelligent B2BUA (Back-to-Back User Agent) sitting between your SIP trunk (PSTN access) and your desk phone/softphone.
Hold Slayer is a single-process async Python application built on FastAPI. It
acts as an intelligent B2BUA (Back-to-Back User Agent) sitting between your SIP
trunk (PSTN access) and your desk phone/softphone.
> **Two SIP engines, selected by `SIP_ENGINE`.** `sippy` (the default) signals
> only — PJSUA2 will not surface an RTP stream for a dialog it does not own, so
> **no audio reaches the classifier** on that path. `pjsua2`
> (`core/pjsua_engine.py`) places the call itself and is the only mode where
> audio reaches the classifier; it is opt-in while being proven against the lab.
> Read [Media plane: why PJSUA2 places the call](#media-plane-why-pjsua2-places-the-call)
> before changing anything in `core/`.
## System Diagram
@@ -27,8 +37,13 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
│ └────┬─────┘ └─────┬─────┘ └──────────────┘ │
│ │ │ │
│ ┌────┴──────────────┴───────────────────┐ │
│ │ Sippy B2BUA Engine │ │
│ │ (SIP calls, DTMF, conference bridge) │ │
│ │ SIP Engine │ │
│ │ signalling + call control │ │
│ └────┬──────────────────────────────────┘ │
│ │ │
│ ┌────┴──────────────────────────────────┐ │
│ │ Media Pipeline (PJSUA2) │ │
│ │ RTP, conference bridge, taps, record │ │
│ └────┬──────────────────────────────────┘ │
│ │ │
└───────┼─────────────────────────────────────────────────────────┘
@@ -70,8 +85,8 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
| Component | File | Purpose |
|-----------|------|---------|
| Sippy Engine | `core/sippy_engine.py` | SIP signaling (INVITE, BYE, REGISTER, DTMF) |
| Media Pipeline | `core/media_pipeline.py` | PJSUA2 RTP media handling, conference bridge, recording |
| Sippy Engine | `core/sippy_engine.py` | SIP signalling (INVITE, BYE, REGISTER, DTMF) |
| Media Pipeline | `core/media_pipeline.py` | PJSUA2 RTP media, conference bridge, taps, recording |
| Recording | `services/recording.py` | WAV file management and storage |
| Analytics | `services/call_analytics.py` | Call metrics, hold time stats, trends |
| Notifications | `services/notification.py` | WebSocket + SMS alerts |
@@ -84,9 +99,10 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
POST /api/v1/calls/hold-slayer { number, intent, call_flow_id }
│
2. Gateway.make_call()
├── CallManager.create_call() → track state
├── SippyEngine.make_call() → SIP INVITE to trunk
└── MediaPipeline.add_stream() → RTP media setup
├── is_emergency_number() → REFUSE 911/112 (before anything else)
├── concurrency cap check → refuse past max_concurrent_calls
├── CallManager.create_call() → track state
└── sip_engine.make_call() → place the call, media follows
│
3. HoldSlayer.run_with_flow() or run_exploration()
├── AudioClassifier.classify() → analyze 3s audio windows
@@ -99,14 +115,14 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
├── TranscriptionService.transcribe() → STT on speech audio
│
├── LLMClient.analyze_ivr_menu() → pick menu option (fallback)
│ └── SippyEngine.send_dtmf() → press the button
│ └── sip_engine.send_dtmf() → press the button
│
└── detect_hold_to_human_transition()
└── HUMAN_DETECTED! → transfer
│
4. Transfer
├── SippyEngine.bridge() → connect call legs
├── MediaPipeline.bridge_streams() → bridge RTP
├── SippyEngine.bridge_calls() → join the two call legs
├── MediaPipeline.bridge_streams() → bridge RTP in the conf bridge
├── EventBus.publish(TRANSFER_STARTED)
└── NotificationService → "Pick up your phone!"
│
@@ -117,44 +133,95 @@ Hold Slayer is a single-process async Python application built on FastAPI. It ac
→ Analytics tracking
```
The emergency guard and the concurrency cap are the first two steps of
`make_call` for a reason, and their order is load-bearing — see
[.claude/rules/call-safety.md](../.claude/rules/call-safety.md).
## Threading Model
Hold Slayer is primarily single-threaded async (asyncio), with one exception:
- **Main thread**: FastAPI + all async services (event bus, hold slayer, classifier, etc.)
- **Sippy thread**: Sippy B2BUA runs its own event loop in a dedicated daemon thread. The `SippyEngine` bridges async↔sync via `asyncio.run_in_executor()`.
- **PJSUA2**: Runs in the main thread using null audio device (no sound card needed — headless server mode).
The README's "single-process async" is a simplification. There are **three**
execution contexts, and the boundaries between them are the highest-leverage
invariant in the codebase.
```
Main Thread (asyncio)
├── FastAPI (uvicorn)
├── EventBus
├── CallManager
├── HoldSlayer
asyncio loop (main thread) Sippy ED thread PJSUA2 worker threads
├── FastAPI (uvicorn) └── ED2 dispatcher └── media / RTP
├── EventBus ├── SIP signalling └── onFrameReceived
├── CallManager ├── UA objects
├── HoldSlayer └── DTMF relay
├── AudioClassifier
├── TranscriptionService
├── LLMClient
├── MediaPipeline (PJSUA2)
├── NotificationService
└── RecordingService
Sippy Thread (daemon)
└── Sippy B2BUA event loop
├── SIP signaling
├── DTMF relay
└── Call leg management
```
**Crossing the boundaries — one funnel each way:**
| Direction | Mechanism | Notes |
|---|---|---|
| Sippy ED → loop | `_post_from_ed` → `asyncio.run_coroutine_threadsafe` → `_on_engine_event` | The single funnel where Sippy-thread events mutate loop state |
| loop → Sippy ED | `_run_on_sippy` → `ED2.callFromThread` | Anything touching a Sippy UA object |
| PJSUA2 worker → loop | `AudioTap.feed` → `loop.call_soon_threadsafe` | The **only** thing a PJSUA2 callback may touch |
`onFrameReceived` runs on a PJSUA2 worker thread every 20 ms. It must call
nothing but `AudioTap.feed`; reaching into pipeline state, the event bus, or a
Sippy object from there is a data race. An exception escaping into PJSUA2's C++
callback tears down the worker thread and silently kills media for every call,
which is why the capture port catches and logs once rather than per frame.
Full detail: [.claude/rules/concurrency-threads.md](../.claude/rules/concurrency-threads.md).
## Design Decisions
### Why Sippy B2BUA + PJSUA2?
### Media plane: why PJSUA2 places the call
We split SIP signaling and media handling into two separate libraries:
The original split was *Sippy signals, PJSUA2 carries media*. It does not work,
for a reason that is not obvious until you try it:
- **Sippy B2BUA** handles SIP signaling (INVITE, BYE, REGISTER, re-INVITE, DTMF relay). It's battle-tested for telephony and handles the complex SIP state machine.
- **PJSUA2** handles RTP media (audio streams, conference bridge, recording, tone generation). It provides a clean C++/Python API for media manipulation without needing to deal with raw RTP.
**PJSUA2 exposes no standalone RTP media object.** Every `AudioMedia` subclass
in the Python bindings is a file player, recorder, tone generator, or capture
port. RTP is reachable only through `pj.Call.getAudioMedia()`, after
`onCallMediaState` fires on a dialog **PJSUA2 itself owns**. There is no
"give me an AudioMedia for this remote host:port" API to call.
This split lets us tap into the audio stream (for classification and STT) without interfering with SIP signaling, and bridge calls through a conference bridge for clean transfer.
So a design where Sippy owns the dialog can never obtain a media stream from
PJSUA2. `MediaPipeline.add_remote_stream()` is not unfinished work — it is a
function that cannot be written against this API. The consequence is that audio
never reaches the classifier: `create_tap` builds a valid capture port with
nothing to attach it to.
**The resolution: PJSUA2 places the call; Sippy keeps every other role.**
| Concern | Owner |
|---|---|
| Emergency guard, concurrency cap | `gateway.make_call` — unchanged, still first |
| Trunk registration | PJSUA2 `Account` |
| Outbound INVITE / answer / hangup | PJSUA2 `Call` |
| RTP, conference bridge, taps, recording | PJSUA2 media |
| DTMF | PJSUA2 `Call.dialDtmf` (RFC 2833) |
| Device registration, routing, leg bridging | Sippy / gateway |
| Inbound call dispatch | PJSUA2 `Account.onIncomingCall` |
Sippy remains the SBC-shaped layer — it is where device registrations, routing
decisions and B2BUA leg-joining live. What moves is the raw dialog for a trunk
call, because owning the dialog is the price of owning the media.
Alternatives considered and rejected:
- **Terminate RTP ourselves** (aiortc or raw sockets) and feed PCM into
`AudioTap` directly, keeping Sippy on the wire. Preserves the split, but
means owning jitter buffering, packet loss concealment and ulaw/alaw
transcoding — precisely the work PJSUA2 exists to do.
- **A loopback `pj.Call` mirroring each real leg**, so PJSUA2 has a dialog it
owns. Avoids touching call placement, but adds a phantom call per real call
and the SDP juggling is fragile.
> **Safety note for this refactor:** `is_emergency_number()` stays the first
> check in `gateway.make_call`, above the concurrency cap and above any SIP
> action, regardless of which library dials. A new outbound path that reaches
> the SIP layer without passing that guard is a serious regression even if
> every test passes.
### Why asyncio Queue-based EventBus?
@@ -164,11 +231,13 @@ This split lets us tap into the audio stream (for classification and STT) withou
- **Dead subscriber cleanup** — full queues are automatically removed
- **Event history** — late joiners can catch up on recent events
If scaling to multiple gateway processes becomes necessary, the EventBus interface can be backed by Redis pub/sub without changing consumers.
If scaling to multiple gateway processes becomes necessary, the EventBus
interface can be backed by Redis pub/sub without changing consumers.
### Why OpenAI-compatible LLM API?
The LLM client uses raw HTTP (httpx) against any OpenAI-compatible endpoint. This means:
The LLM client uses raw HTTP (httpx) against any OpenAI-compatible endpoint.
This means:
- **Ollama** (local, free) — `http://localhost:11434/v1`
- **LM Studio** (local, free) — `http://localhost:1234/v1`
@@ -176,3 +245,11 @@ The LLM client uses raw HTTP (httpx) against any OpenAI-compatible endpoint. Thi
- **OpenAI** (cloud) — `https://api.openai.com/v1`
No SDK dependency. No vendor lock-in. Switch models by changing one env var.
## Testing against a fake PSTN
`tests/lab/` runs an Asterisk instance that answers calls, plays an IVR, holds
with music and connects a "human" — so the gateway has something real to dial
that is not the PSTN. `SIP_TRUNK_HOST` is just an address, so the production
code path runs unmodified; while it points at the lab there is no route to the
PSTN at all. See [tests/lab/README.md](../tests/lab/README.md).

44
docs/configuration.md
View File

@@ -34,14 +34,36 @@ SSO-disabled-off-loopback.
| `SIP_TRUNK_DID` | Your phone number (E.164) | — | Yes |
| `SIP_TRUNK_TRANSPORT` | Transport protocol (`udp`, `tcp`, `tls`) | `udp` | No |
### Server
| Variable | Description | Default | Required |
|----------|-------------|---------|----------|
| `HOST` | Bind address. Off-loopback requires `CASDOOR_ENABLED=true` | `0.0.0.0` | No |
| `PORT` | Bind port | `8000` | No |
| `DEBUG` | SQLAlchemy echo + uvicorn reload | `false` | No |
| `LOG_LEVEL` | Root log level (`debug`/`info`/`warning`/`error`) | `info` | No |
| `LOG_FORMAT` | `text` (human-readable) or `json` (structured, for Loki) | `text` | No |
`LOG_FORMAT=json` renders one JSON object per line, including uvicorn's access
log — `method`, `path`, `status_code` (numeric, so it can be range-filtered) and
`client_addr` arrive as queryable fields rather than a formatted string. The
Docker image sets it; `text` is the default so local development stays readable.
### Safety
| Variable | Description | Default | Required |
|----------|-------------|---------|----------|
| `MAX_CONCURRENT_CALLS` | Cap on simultaneous outbound calls | `4` | No |
| `USE_MOCK_SIP` | Run the mock SIP engine — no real calls. Must be asked for explicitly; an unconfigured trunk without it fails startup | `false` | No |
| `SIP_ENGINE` | `sippy` (signalling only — no audio reaches the classifier) or `pjsua2` (call control + media) | `sippy` | No |
### Gateway
| Variable | Description | Default | Required |
|----------|-------------|---------|----------|
| `GATEWAY_SIP_PORT` | Port for device SIP registration | `5080` | No |
| `GATEWAY_RTP_PORT_MIN` | Minimum RTP port | `10000` | No |
| `GATEWAY_RTP_PORT_MAX` | Maximum RTP port | `20000` | No |
| `GATEWAY_HOST` | Bind address | `0.0.0.0` | No |
| `GATEWAY_SIP_HOST` | Bind address for the device-registration listener | `0.0.0.0` | No |
| `GATEWAY_SIP_PORT` | Port for device SIP registration | `5060` | No |
| `GATEWAY_SIP_DOMAIN` | SIP domain devices register against | `gateway.local` | No |
### LLM
@@ -65,7 +87,7 @@ SSO-disabled-off-loopback.
| Variable | Description | Default | Required |
|----------|-------------|---------|----------|
| `DATABASE_URL` | PostgreSQL or SQLite connection string | `sqlite+aiosqlite:///./hold_slayer.db` | No |
| `DATABASE_URL` | PostgreSQL connection string. Startup exits with a readable error if unset | — | Yes |
### Notifications
@@ -73,6 +95,18 @@ SSO-disabled-off-loopback.
|----------|-------------|---------|----------|
| `NOTIFY_SMS_NUMBER` | Phone number for SMS alerts (E.164) | — | No |
### Receptionist
| Variable | Description | Default | Required |
|----------|-------------|---------|----------|
| `RECEPTIONIST_ENABLED` | Answer inbound calls with the AI receptionist | `true` | No |
| `RECEPTIONIST_GREETING_TEMPLATE` | Spoken greeting | `"Hi, you've reached Robert's line. Who's calling, and what's this about?"` | No |
| `RECEPTIONIST_MESSAGE_PROMPT` | Spoken prompt before recording a message | `"Please leave your message after the tone."` | No |
| `RECEPTIONIST_LLM_PERSONA` | System prompt shaping the receptionist's decisions | See `config.py` | No |
| `RECEPTIONIST_LISTEN_TIMEOUT_S` | Seconds to wait for the caller to speak | `15.0` | No |
| `RECEPTIONIST_END_OF_UTTERANCE_SILENCE_S` | Silence marking the end of a turn | `1.2` | No |
| `RECEPTIONIST_MESSAGE_MAX_SECONDS` | Voicemail cap | `90` | No |
### Audio Classifier
| Variable | Description | Default | Required |

24
main.py
View File

@@ -26,6 +26,7 @@ from api import call_flows, call_history, calls, devices, routing, tokens, webso
from auth import get_current_owner, init_jwks_client, is_owner, resolve_from_header_or_query
from config import Settings, get_settings
from core.gateway import AIPSTNGateway, build_sip_engine
from core.logging_config import configure_logging
from db.database import close_db, init_db, session_scope
from mcp_server.server import create_mcp_server
from models.call import CallMode
@@ -39,13 +40,12 @@ from services.routing import RoutingService
from services.transcription import TranscriptionService
from services.tts import TTSService
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)-7s | %(name)s | %(message)s",
datefmt="%H:%M:%S",
stream=sys.stdout,
)
# Configure logging at import so anything logged during module import and
# startup config checks is formatted. Uvicorn installs its own handlers *after*
# importing this module, so `lifespan` calls `configure_logging` again to take
# them over — see core/logging_config.py.
_startup_settings = get_settings()
configure_logging(_startup_settings.log_format, _startup_settings.log_level)
logger = logging.getLogger(__name__)
@@ -150,6 +150,12 @@ def _check_startup_config(settings: Settings) -> None:
async def lifespan(app: FastAPI):
"""Startup: Initialize database, SIP engine, and services."""
settings = get_settings()
# Re-apply: under `uvicorn main:app` the server installs its own handlers on
# `uvicorn`/`uvicorn.access` after importing this module, which would emit
# colourised text alongside our JSON. This takes them back over.
configure_logging(settings.log_format, settings.log_level)
_check_startup_config(settings)
# Prefetch Casdoor's JWKS so the first authenticated request doesn't pay
@@ -622,4 +628,8 @@ if __name__ == "__main__":
port=settings.port,
reload=settings.debug,
log_level=settings.log_level,
# Suppress uvicorn's own dictConfig: ours is installed at import and in
# lifespan, and letting uvicorn apply its default would attach a second,
# colourised handler — every line twice, one of them not JSON.
log_config=None,
)

66
tests/lab/README.md
View File

@@ -63,6 +63,72 @@ GATEWAY_SIP_PORT=21062 # must differ from the Asterisk port
> engine is assigned — `main.py`'s lifespan calls `build_sip_engine()` after
> construction. A harness that skips that step silently tests the mock.
## The softphone (transfer target)
**Asterisk is the registrar for devices, not Hold Slayer.** The gateway reaches
a desk phone by dialling extension `2001`, which Asterisk routes to whatever
has registered as `softphone`. This deliberately avoids Hold Slayer's own SIP
listener, which answers `200 OK` to any REGISTER with no digest challenge.
The `pjsua` CLI built alongside the Python bindings is the test device — same
library stack as the gateway, no extra dependency. It needs an RPATH patch like
the bindings did:
```bash
cp ~/src/pjproject/pjsip-apps/bin/pjsua-x86_64-pc-linux-gnu ~/.local/bin/pjsua
patchelf --set-rpath $HOME/.local/lib ~/.local/bin/pjsua
```
Register it (config file avoids shell-quoting pain):
```bash
cat > softphone.cfg <<'EOF'
--null-audio
--auto-answer=200
--max-calls=4
--local-port=21070
--id=sip:softphone@127.0.0.1
--registrar=sip:127.0.0.1:21061
--realm=asterisk
--username=softphone
--password=labphone
--log-level=3
EOF
# pjsua is an interactive console app: it exits ~8s after start if stdin is
# closed or /dev/null. Hold a fifo open on stdin — `script -qfc` and
# `setsid </dev/null` both look like they work (registration succeeds) and
# then the process dies, leaving a stale contact in Asterisk that routes
# INVITEs to a port nobody is listening on.
mkfifo sp.fifo
setsid sh -c 'exec 3<>sp.fifo; pjsua --config-file softphone.cfg <&3 >softphone.log 2>&1' &
```
Verify — **check the port is actually bound**, not just that Asterisk holds a
contact, since a stale registration outlives the process:
```bash
ss -lnup | grep 21070 # must be listening
docker compose -f docker-compose.lab.yml exec asterisk \
asterisk -rx "pjsip show contacts" # must show softphone
```
Then place a call to `2001`. Both legs should show `Up` under one bridge id:
```bash
docker compose -f docker-compose.lab.yml exec asterisk \
asterisk -rx "core show channels concise"
```
> **`--realm=asterisk`, not `--realm='*'`** — the wildcard fails with
> `PJSIP_EFAILEDCREDENTIAL` against Asterisk's digest challenge.
> **Qualify is off** for this AOR (`qualify_frequency = 0`): the pjsua console
> does not answer `OPTIONS`, so polling marks a working softphone `Unavail` and
> the dialplan refuses to ring it. The `2001` guard therefore tests
> `PJSIP_AOR(softphone,contact)` rather than `DEVICE_STATE`. A real hardphone
> answers OPTIONS and can have qualify re-enabled.
## Useful commands
```bash

31
tests/lab/dialplan/asterisk.conf
View File

@@ -1,24 +1,17 @@
; Minimal Asterisk core config for the lab.
[directories](!)
astetcdir => /etc/asterisk
astmoddir => /usr/lib/asterisk/modules
astvarlibdir => /var/lib/asterisk
astdbdir => /var/lib/asterisk
astkeydir => /var/lib/asterisk
astdatadir => /var/lib/asterisk
astagidir => /var/lib/asterisk/agi-bin
astspooldir => /var/spool/asterisk
astrundir => /var/run/asterisk
astlogdir => /var/log/asterisk
astsbindir => /usr/sbin
;
; Deliberately does NOT set [directories] or runuser/rungroup: the image's
; compiled-in defaults are correct, and it runs as the `asterisk` user via a
; USER directive. Overriding either risks breaking the container for no gain
; (an earlier version of this file did both).
[options]
; Log to stdout so Docker's json-file driver captures it and Alloy ships it.
; A file-based log inside the container would be invisible to Loki.
; Log to stdout so Docker's json-file driver captures it and Alloy ships it
; to Loki. A file-based log inside the container would be invisible.
verbose = 3
debug = 0
; No ANSI colour. Asterisk colourises the console by default and the escape
; codes travel through Docker into Loki, where every line arrives wrapped in
; \x1b[0;30m — unreadable in Grafana and awkward to filter on. This setting
; lives here, not in logger.conf, and only takes effect if this file is
; actually mounted into the container.
nocolor = yes
dumpcore = no
; Never run as root inside the container.
runuser = asterisk
rungroup = asterisk

17
tests/lab/dialplan/extensions.conf
View File

@@ -128,6 +128,23 @@ exten => 1008,1,NoOp(LAB 1008: silence)
same => n,Wait(40)
same => n,Hangup()
; --- 2001: ring the registered softphone ----------------------------------
; The transfer target. Asterisk is the registrar for devices, so the gateway
; reaches a desk phone by dialling this rather than by registering it itself.
; Fails fast when nothing is registered — a silent 30s ring would look like a
; gateway bug rather than an absent softphone.
exten => 2001,1,NoOp(LAB 2001: ring softphone)
; Count registered contacts rather than DEVICE_STATE: device state follows
; the OPTIONS qualify, which is off for this AOR (the pjsua CLI does not
; answer OPTIONS), so a registered softphone would still read UNAVAILABLE.
same => n,GotoIf($[${PJSIP_AOR(softphone,contact)} = ""]?nodevice)
same => n,Dial(PJSIP/softphone,30)
same => n,Hangup()
same => n(nodevice),NoOp(LAB 2001: no softphone registered)
same => n,Answer()
same => n,Playback(lab-speech)
same => n,Hangup()
; --- echo test ------------------------------------------------------------
; Not a scenario — a debugging aid. Echoes audio back so you can confirm
; bidirectional RTP by ear when something looks wrong.

18
tests/lab/dialplan/logger.conf
View File

@@ -3,6 +3,22 @@
; the container would put the logs where nothing can see them.
[general]
dateformat = %F %T
; Colour is disabled in asterisk.conf (`nocolor = yes`), not here — Asterisk
; colourises the console by default and the escape codes travel through Docker
; into Loki, where every line arrives wrapped in \x1b[0;30m. That file must be
; mounted for the setting to take effect.
[logfiles]
console => notice,warning,error
; Warnings and errors only. That covers what matters when a lab call
; misbehaves: failed authentication, no-matching-endpoint, playback failures.
;
; `notice` and `verbose` are both excluded because the "Remote UNIX
; connection" pairs the healthcheck generates arrive on those channels, and
; they swamped everything else (the Loki stream measured 100% healthcheck
; noise before this). The healthcheck itself is also reduced to one CLI call
; on a 60s interval in docker-compose — the two changes work together.
;
; To trace a call's dialplan execution, raise verbosity at runtime rather than
; leaving it on:
; asterisk -rx "core set verbose 3"
console => warning,error

67
tests/lab/dialplan/pjsip.conf
View File

@@ -34,14 +34,21 @@ local_net = {{ asterisk_local_net }}
; ---------------------------------------------------------------------------
; Hold Slayer authenticates as this endpoint to place calls into the lab.
; Identify the endpoint by source address. Asterisk's default matching uses
; the From-header domain, which Hold Slayer populates from its SIP bind
; address (0.0.0.0 on a wildcard bind) — never a value Asterisk can match.
; Matching on where the packet actually came from sidesteps that.
; Identify the endpoint by source address *and port*. Asterisk's default
; matching uses the From-header domain, which Hold Slayer populates from its
; SIP bind address (0.0.0.0 on a wildcard bind) — never a value Asterisk can
; match. Matching on where the packet came from sidesteps that.
;
; The port is essential when the softphone runs on the same host: a
; host-only match claims *every* packet from that address, so the
; softphone's REGISTER would be attributed to this endpoint and checked
; against the gateway's password ("Failed to authenticate", confusingly).
; Endpoints that authenticate by username (the softphone) must not be
; covered by an identify block.
[hold-slayer]
type = identify
endpoint = hold-slayer
match = {{ asterisk_match_host }}
match = {{ asterisk_match_host }}:{{ asterisk_gateway_port }}
[hold-slayer]
type = endpoint
@@ -68,6 +75,56 @@ auth_type = userpass
username = {{ asterisk_sip_username }}
password = {{ asterisk_sip_password }}
; ---------------------------------------------------------------------------
; Softphone endpoint — the transfer target
; ---------------------------------------------------------------------------
; Asterisk is the registrar for devices, not Hold Slayer. A softphone REGISTERs
; here and the gateway transfers a live call to it by dialling extension 2001.
;
; Deliberate: Hold Slayer's own SIP listener answers 200 OK to any REGISTER
; with no digest challenge, so anything on the network could register as a
; device and receive transferred calls. Keeping registration in Asterisk means
; the lab does not depend on that path, and the softphone is authenticated.
;
; Test with the pjsua CLI built alongside the Python bindings:
; pjsua --null-audio --auto-answer=200 \
; --id=sip:softphone@<asterisk-host> \
; --registrar=sip:<asterisk-host>:21061 \
; --realm='*' --username=softphone --password=<pw> \
; --local-port=<free port>
[softphone]
type = endpoint
context = hold-slayer-lab
disallow = all
allow = ulaw
allow = alaw
auth = softphone-auth
aors = softphone
dtmf_mode = rfc4733
direct_media = no
force_rport = yes
rewrite_contact = yes
rtp_symmetric = yes
[softphone-auth]
type = auth
auth_type = userpass
username = {{ asterisk_softphone_username }}
password = {{ asterisk_softphone_password }}
[softphone]
type = aor
; The device's contact is learned from its REGISTER rather than configured —
; a softphone's port is not known in advance.
max_contacts = 1
remove_existing = yes
; No qualify: the pjsua CLI does not answer OPTIONS while sitting at its
; console prompt, so polling marks a perfectly working softphone Unavail and
; the dialplan refuses to ring it. Registration itself is the liveness signal
; here. A real hardphone answers OPTIONS and can have qualify re-enabled.
qualify_frequency = 0
[hold-slayer]
type = aor
max_contacts = 2

14
tests/lab/docker-compose.lab.yml
View File

@@ -21,8 +21,22 @@ services:
- ./dialplan/pjsip.local.conf:/etc/asterisk/pjsip.conf:ro
- ./dialplan/rtp.local.conf:/etc/asterisk/rtp.conf:ro
- ./dialplan/logger.conf:/etc/asterisk/logger.conf:ro
# Carries `nocolor = yes`: without it every log line reaches Loki
# wrapped in ANSI escape codes.
- ./dialplan/asterisk.conf:/etc/asterisk/asterisk.conf:ro
# The image ships no sound files at all. These are generated by
# sounds/generate.py; Asterisk resolves Playback(lab-music) to
# lab-music.sln here (8kHz signed-linear, no transcoding).
- ./sounds:/var/lib/asterisk/sounds/en:ro
# The image's default command is `-vvvdddf` — verbosity 3 and debug 3
# forced on the command line, which overrides both asterisk.conf and
# logger.conf. That makes every healthcheck CLI connection log a
# "Remote UNIX connection" pair: ~2900 lines/day of pure noise that
# completely buried the real SIP events in Loki.
#
# -f foreground (required: Docker needs PID 1 to stay), -T timestamps,
# -W colour off, -U run as asterisk, -p realtime priority. No -v, no -d:
# warnings and errors still log, and verbosity can be raised at runtime
# with `asterisk -rx "core set verbose 3"` when tracing a call.
command: ["/usr/sbin/asterisk", "-f", "-T", "-W", "-U", "asterisk", "-p"]
restart: unless-stopped

61
tests/lab/sounds/generate.py
View File

@@ -12,7 +12,6 @@ finds `lab-music.sln`.
python generate.py [outdir]
"""
import struct
import sys
from pathlib import Path
@@ -29,13 +28,14 @@ def _write_sln(path: Path, samples: np.ndarray) -> None:
print(f" {path.name}: {len(pcm) / RATE:.1f}s ({path.stat().st_size} bytes)")
def make_music(seconds: float = 30.0) -> np.ndarray:
def make_music(seconds: float = 30.0, seed: int = 7) -> np.ndarray:
"""Sustained multi-harmonic tones — what the classifier must call MUSIC.
A chord progression with stable pitch and strong harmonic structure. The
steady spectrum across a long window is what distinguishes music from
speech; this deliberately has no pauses.
"""
rng = np.random.default_rng(seed)
t = np.linspace(0, seconds, int(RATE * seconds), endpoint=False)
# A-minor-ish progression, one chord per 2s bar.
chords = [(220.0, 261.6, 329.6), (196.0, 246.9, 293.7),
@@ -48,12 +48,21 @@ def make_music(seconds: float = 30.0) -> np.ndarray:
break
mask = (t >= start) & (t < end)
for j, freq in enumerate(chord):
# Fundamental plus two harmonics, decaying — a plucked-string feel.
for h, amp in ((1, 0.30), (2, 0.12), (3, 0.05)):
# Fundamental plus four harmonics, decaying — a plucked-string
# feel. Enough harmonics to keep spectral flatness inside the
# music score's 0.05-0.4 band: with only three, some windows fall
# *below* 0.05 (too pure to read as music) and score as speech.
for h, amp in ((1, 0.30), (2, 0.12), (3, 0.05), (4, 0.03), (5, 0.02)):
out[mask] += amp / (j + 1) * np.sin(2 * np.pi * freq * h * t[mask])
# Gentle per-bar envelope so bars are distinguishable but never silent.
env = 0.8 + 0.2 * np.sin(2 * np.pi * (t[mask] - start) / bar)
out[mask] *= env
# Recording-style noise floor. Windows straddling a chord change have a
# momentarily sparse spectrum and land just *under* the music score's
# 0.05 flatness floor, scoring as speech. This is well below the level
# that would disturb tonality — every real recording has one.
out += rng.normal(0, 0.004, len(out))
return out * 0.45
@@ -74,16 +83,44 @@ def make_speech(seconds: float = 8.0, seed: int = 1337) -> np.ndarray:
mask = (t >= pos) & (t < pos + syl)
if mask.any():
local = t[mask] - pos
f0 = rng.uniform(95, 165) # fundamental — adult speaking range
# Two formants, swept slightly across the syllable. The ranges
# deliberately avoid the DTMF bands (rows 697-941, columns
# 1209-1633): a formant pair landing on both trips the Goertzel
# detector and the whole utterance is classified as a keypress.
f1 = rng.uniform(300, 620) + rng.uniform(-40, 40) * local / syl
f2 = rng.uniform(1750, 2600) + rng.uniform(-120, 120) * local / syl
sig = (0.50 * np.sin(2 * np.pi * f0 * local)
frac = local / syl
# Pitch CONTOUR, not a constant. This is the single feature that
# separates this fixture from music. `_detect_tonality` looks for
# an autocorrelation peak > 0.5 in the 50-1000 Hz lag range; a
# fixed f0 is perfectly periodic there, scores is_tonal=True, and
# hands the music score a free 0.3 that speech cannot outrun.
# Real voices glide and jitter, so the periodicity never locks.
f0_start = rng.uniform(95, 165)
f0_end = f0_start * rng.uniform(0.72, 1.38) # rise or fall
f0 = f0_start + (f0_end - f0_start) * frac
# Cycle-to-cycle jitter on top of the glide (~2% is human).
f0 *= 1.0 + 0.02 * rng.standard_normal(len(local))
# Integrate frequency to phase — with a varying f0, `2*pi*f*t`
# would be wrong (that is a chirp only if f is the *instantaneous*
# rate, which it is not once f0 itself moves).
ph0 = 2 * np.pi * np.cumsum(f0) / RATE
# Two formants, swept across the syllable. The ranges deliberately
# avoid the DTMF bands (rows 697-941, columns 1209-1633): a formant
# pair landing on both trips the Goertzel detector and the whole
# utterance is classified as a keypress.
f1 = rng.uniform(300, 620) + rng.uniform(-40, 40) * frac
f2 = rng.uniform(1750, 2600) + rng.uniform(-120, 120) * frac
sig = (0.50 * np.sin(ph0)
+ 0.30 * np.sin(2 * np.pi * f1 * local)
+ 0.18 * np.sin(2 * np.pi * f2 * local))
# Aspiration noise — HIGH-PASSED, not broadband. Real speech noise
# sits above the formants; flat noise puts energy in every
# Goertzel bin, so the strongest DTMF row and column both clear
# the detector's `total_power * 0.1` threshold and every syllable
# reads as a keypress. A first-difference filter (y[n]-y[n-1]) is
# a cheap +6dB/octave tilt that leaves the 697-1633 Hz DTMF bands
# comparatively empty. The 0.09 level is chosen for margin: it puts
# spectral flatness at ~0.46, mid-way through the 0.1-0.5 band the
# speech score rewards, rather than on either edge.
noise = rng.standard_normal(len(local) + 1)
sig += 0.09 * np.diff(noise)
# Raised-cosine envelope: no clicks at syllable edges.
sig *= np.sin(np.pi * local / syl) ** 0.6
out[mask] += sig

187
tests/test_graceful_degradation.py Normal file
View File

@@ -0,0 +1,187 @@
"""
Graceful-degradation tests.
Every external dependency — STT, LLM, TTS — is reachable over the network and
can be down. The gateway's rule is that a dead dependency degrades the call
rather than killing it, and says so: each failure publishes an `ERROR` event
naming the service, so a down Speaches reads as "transcription failed" rather
than "the AI is making bad decisions".
The behaviour is already implemented across the services; these tests exist so a
later refactor can't quietly remove it. The failure mode being guarded against is
silent: an un-caught exception in one of these paths aborts a live phone call.
"""
from unittest.mock import AsyncMock, MagicMock
import pytest
from config import Settings
from core.event_bus import EventBus
from models.events import EventType
def _gateway():
"""A gateway stand-in with a real event bus, so events can be asserted."""
gw = MagicMock()
gw.settings = Settings(database_url="sqlite+aiosqlite:///:memory:")
gw.event_bus = EventBus()
gw.call_manager = MagicMock()
gw.call_manager.add_transcript = AsyncMock()
return gw
def _hold_slayer(gateway, transcription):
from services.audio_classifier import AudioClassifier
from services.hold_slayer import HoldSlayerService
return HoldSlayerService(
gateway=gateway,
call_manager=gateway.call_manager,
sip_engine=MagicMock(),
classifier=AudioClassifier(gateway.settings.classifier),
transcription=transcription,
settings=gateway.settings,
)
async def _errors_for(bus: EventBus, coro):
"""Run `coro` while subscribed, returning the ERROR events it published."""
import asyncio
sub = bus.subscribe(event_types={EventType.ERROR})
try:
result = await coro
seen = []
while True:
try:
seen.append(sub._queue.get_nowait())
except asyncio.QueueEmpty:
break
return result, seen
finally:
bus.unsubscribe(sub)
class TestClassifierWithoutSTT:
"""The classifier is spectral: it must not depend on STT at all."""
def _classifier(self):
from services.audio_classifier import AudioClassifier
return AudioClassifier(Settings(database_url="sqlite+aiosqlite:///:memory:").classifier)
def test_classify_takes_only_audio(self):
# A transcript parameter would make STT a hard dependency of hold
# detection — the thing this checkbox is about.
import inspect
params = set(inspect.signature(self._classifier().classify_chunk).parameters)
assert params == {"audio_data"}
async def test_classifies_with_no_stt_service_anywhere(self):
# Silence is the cheapest deterministic input; the point is that a
# classification is produced at all with no STT in the picture.
result = await self._classifier().classify(b"\x00\x00" * 16000)
assert result.audio_type is not None
class TestTranscriptionDegradation:
async def test_hold_slayer_transcribe_returns_empty_on_failure(self):
gw = _gateway()
stt = MagicMock()
stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
svc = _hold_slayer(gw, stt)
text, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))
assert text == "" # empty transcript, not an exception
assert len(errors) == 1
assert errors[0].data["service"] == "transcription"
async def test_error_event_names_the_service(self):
# "transcription failed" vs "the AI decided badly" — the whole reason
# transcribe() raises instead of swallowing.
gw = _gateway()
stt = MagicMock()
stt.transcribe = AsyncMock(side_effect=RuntimeError("Connection refused"))
svc = _hold_slayer(gw, stt)
_, errors = await _errors_for(gw.event_bus, svc._transcribe("call-1", b"\x00" * 320))
assert "Connection refused" in errors[0].data["error"]
async def test_service_error_survives_a_dead_event_bus(self):
# Degradation reporting must not itself become a failure path.
gw = _gateway()
gw.event_bus.publish = AsyncMock(side_effect=RuntimeError("bus down"))
stt = MagicMock()
stt.transcribe = AsyncMock(side_effect=RuntimeError("stt down"))
svc = _hold_slayer(gw, stt)
assert await svc._transcribe("call-1", b"\x00" * 320) == ""
async def test_transcription_marks_itself_unavailable(self):
# /health reads this flag; a failure that doesn't record itself makes
# the probe lie.
import httpx
from services.transcription import TranscriptionService
svc = TranscriptionService(Settings(database_url="sqlite+aiosqlite:///:memory:").speaches)
client = MagicMock()
client.post = AsyncMock(side_effect=httpx.ConnectError("refused"))
svc._client = client
svc._client.is_closed = False
with pytest.raises(Exception):
await svc.transcribe(b"\x00" * 320)
assert svc.available is False
class TestReceptionistDegradation:
def _receptionist(self, gateway, **kw):
from services.receptionist import ReceptionistService
return ReceptionistService(gateway=gateway, **kw)
async def test_llm_failure_falls_back_to_a_usable_decision(self):
gw = _gateway()
svc = self._receptionist(gw)
call = MagicMock(id="call-1", remote_number="+15551234567")
llm = MagicMock()
llm.chat_json = AsyncMock(side_effect=RuntimeError("LLM down"))
import services.llm_client as llm_mod
original = llm_mod.get_llm
llm_mod.get_llm = lambda: llm
try:
result, errors = await _errors_for(
gw.event_bus, svc._classify(call, "I need to speak to someone", None)
)
finally:
llm_mod.get_llm = original
# A decision still comes back, so the call can proceed.
assert result["recommended_action"] in {"ring", "message", "reject"}
assert errors[0].data["service"] == "llm"
async def test_no_transcription_service_yields_empty_not_crash(self):
# transcription=None is a valid wiring (STT not configured).
gw = _gateway()
svc = self._receptionist(gw, transcription=None)
assert svc.transcription is None
class TestHealthReportsDegradation:
"""A degraded gateway must read as degraded — /health may not lie."""
def test_availability_helper_distinguishes_unknown_from_down(self):
# Four distinct states, because "not wired up" and "wired up but the
# remote is refusing connections" are different operator problems.
import main
assert main._availability(None) == "not attached"
assert main._availability(MagicMock(available=None)) == "unknown (no requests yet)"
assert main._availability(MagicMock(available=True)) == "ok"
assert main._availability(MagicMock(available=False)) == "unreachable"

109
tests/test_lab_fixtures.py Normal file
View File

@@ -0,0 +1,109 @@
"""
Lab audio fixtures — the classifier must agree with what each one claims to be.
These guard the *fixtures*, not the classifier. `tests/lab/sounds/generate.py`
synthesises music/speech/silence that the Asterisk lab plays down a real call;
if a fixture drifts into the wrong class, every lab result built on it is
quietly meaningless — a hold-music scenario that never classifies as music
proves nothing about the hold slayer.
The first version of these fixtures passed on the opening 3s window and drifted
to MUSIC after, which a single-window check would not have caught. Hence the
sweep across every window.
Skipped when the fixtures have not been generated: they are gitignored (~680K,
reproducible from a fixed seed), so a fresh checkout has none until
`python tests/lab/sounds/generate.py` runs.
"""
import subprocess
import sys
from pathlib import Path
import numpy as np
import pytest
from config import Settings
from models.call import AudioClassification
from services.audio_classifier import SAMPLE_RATE, AudioClassifier
SOUNDS_DIR = Path(__file__).parent / "lab" / "sounds"
GENERATOR = SOUNDS_DIR / "generate.py"
# The lab writes 8 kHz .sln; the classifier works at 16 kHz.
LAB_RATE = 8000
WINDOW_SAMPLES = SAMPLE_RATE * 3 # classifier's 3s analysis window
FIXTURES = [
("lab-music.sln", AudioClassification.MUSIC),
("lab-speech.sln", AudioClassification.LIVE_HUMAN),
("lab-silence.sln", AudioClassification.SILENCE),
]
def _load_16k(path: Path) -> np.ndarray:
"""Load an 8 kHz .sln and upsample to the classifier's 16 kHz."""
return np.repeat(np.fromfile(path, dtype="<i2"), SAMPLE_RATE // LAB_RATE)
def _windows(samples: np.ndarray, step: int):
"""Yield successive analysis windows; at least one, even for short files."""
end = max(1, len(samples) - WINDOW_SAMPLES)
for offset in range(0, end, step):
yield samples[offset : offset + WINDOW_SAMPLES].astype("<i2").tobytes()
@pytest.fixture(scope="module")
def classifier():
return AudioClassifier(settings=Settings().classifier)
@pytest.mark.parametrize("filename,expected", FIXTURES)
def test_fixture_classifies_correctly_in_every_window(filename, expected, classifier):
"""Every window must classify correctly — not just the first.
Stepped at half the window length so windows overlap: a fixture that only
works on aligned boundaries would still be a trap in a live call, where
the window has no relationship to where the audio started.
"""
path = SOUNDS_DIR / filename
if not path.exists():
pytest.skip(f"{filename} not generated — run {GENERATOR}")
samples = _load_16k(path)
results = [
classifier.classify_chunk(w).audio_type
for w in _windows(samples, step=WINDOW_SAMPLES // 2)
]
wrong = [(i, r.value) for i, r in enumerate(results) if r is not expected]
assert not wrong, (
f"{filename} must classify as {expected.value} in all "
f"{len(results)} windows; wrong: {wrong}"
)
def test_generator_is_deterministic(tmp_path):
"""Same bytes on every run — the whole point of synthesising them.
Real hold music varies per call, so a classifier regression on the PSTN is
indistinguishable from noise. Fixed-seed audio makes the answer binary.
"""
if not GENERATOR.exists():
pytest.skip("generator not present")
def run(target: Path) -> dict[str, bytes]:
subprocess.run(
[sys.executable, str(GENERATOR), str(target)],
check=True,
capture_output=True,
)
return {p.name: p.read_bytes() for p in sorted(target.glob("*.sln"))}
first = run(tmp_path / "a")
second = run(tmp_path / "b")
assert first, "generator produced no .sln files"
assert first.keys() == second.keys()
for name in first:
assert first[name] == second[name], f"{name} differs between runs"

183
tests/test_logging_config.py Normal file
View File

@@ -0,0 +1,183 @@
"""
Structured-logging tests.
The two things worth guarding are the ones that are easy to break silently:
uvicorn's access logger must actually route through our formatter (it sets
`propagate = False` and brings its own handler), and its arg tuple must land as
real JSON fields rather than a pre-formatted string.
"""
import json
import logging
import logging.config
import pytest
from config import Settings
from core.logging_config import JSONFormatter, configure_logging
def _record(name="test.logger", level=logging.INFO, msg="hello", args=None, **extra):
record = logging.LogRecord(
name=name, level=level, pathname=__file__, lineno=1, msg=msg, args=args, exc_info=None
)
for key, value in extra.items():
setattr(record, key, value)
return record
def _emit(record):
return json.loads(JSONFormatter().format(record))
class TestJSONFormatter:
def test_emits_single_line_json_with_core_fields(self):
out = JSONFormatter().format(_record())
assert "\n" not in out
payload = json.loads(out)
assert payload["msg"] == "hello"
assert payload["level"] == "INFO"
assert payload["logger"] == "test.logger"
def test_timestamp_is_utc_rfc3339_with_date(self):
# logging's default asctime is local-time and date-less, which is
# exactly what makes text logs hard to correlate in Loki.
ts = _emit(_record())["ts"]
assert ts.endswith("+00:00")
assert "T" in ts
def test_interpolates_message_args(self):
assert _emit(_record(msg="call %s ended", args=("abc123",)))["msg"] == "call abc123 ended"
def test_extra_fields_are_promoted(self):
payload = _emit(_record(call_id="c-1", duration=12.5))
assert payload["call_id"] == "c-1"
assert payload["duration"] == 12.5
def test_uvicorn_color_message_is_dropped(self):
# Uvicorn ships an ANSI-coloured copy of the message via extra=; letting
# it through puts escape codes in Loki.
payload = _emit(
_record(name="uvicorn.error", msg="Started", color_message="\x1b[36mStarted\x1b[0m")
)
assert "color_message" not in payload
assert "\x1b" not in json.dumps(payload)
def test_non_json_native_extra_is_stringified(self):
payload = _emit(_record(obj=object()))
assert isinstance(payload["obj"], str)
def test_secretstr_extra_stays_masked(self):
# Config secrets are SecretStr precisely so an accidental log can't leak
# them; stringification must preserve that.
from pydantic import SecretStr
payload = _emit(_record(secret=SecretStr("hs_pat_supersecret")))
assert "supersecret" not in json.dumps(payload)
def test_exception_is_captured(self):
try:
raise ValueError("boom")
except ValueError:
import sys
record = _record(level=logging.ERROR, msg="failed")
record.exc_info = sys.exc_info()
payload = _emit(record)
assert "ValueError: boom" in payload["exc"]
def test_thread_name_included_only_off_main_thread(self):
# Which execution context logged a line is the first question when
# debugging a call across the asyncio/Sippy/PJSUA2 boundary.
on_main = _record()
on_main.threadName = "MainThread"
assert "thread" not in _emit(on_main)
off_main = _record()
off_main.threadName = "sippy-ed"
assert _emit(off_main)["thread"] == "sippy-ed"
class TestAccessLogFields:
def _access(self, status=200):
return _emit(
_record(
name="uvicorn.access",
msg='%s - "%s %s HTTP/%s" %d',
args=("127.0.0.1:5050", "GET", "/api/v1/calls", "1.1", status),
)
)
def test_arg_tuple_becomes_structured_fields(self):
payload = self._access()
assert payload["method"] == "GET"
assert payload["path"] == "/api/v1/calls"
assert payload["client_addr"] == "127.0.0.1:5050"
assert payload["http_version"] == "1.1"
def test_status_code_is_a_number_not_a_string(self):
# So Loki can range-filter on it (status_code >= 500).
assert self._access(503)["status_code"] == 503
def test_access_line_is_not_pre_formatted(self):
# The whole point: no interpolated request line to regex back apart.
assert "msg" not in self._access()
def test_unexpected_arg_shape_falls_back_to_message(self):
# A logging path that raises would take out the request; degrade instead.
payload = _emit(_record(name="uvicorn.access", msg="just a string", args=None))
assert payload["msg"] == "just a string"
class TestConfigureLogging:
@pytest.fixture(autouse=True)
def _restore(self):
root = logging.getLogger()
saved = root.handlers[:], root.level
yield
root.handlers[:] = saved[0]
root.setLevel(saved[1])
def _uvicorn_defaults(self):
"""Reproduce what uvicorn does to its loggers at startup."""
from uvicorn.config import LOGGING_CONFIG
logging.config.dictConfig(LOGGING_CONFIG)
def test_takes_over_uvicorn_handlers(self):
self._uvicorn_defaults()
access = logging.getLogger("uvicorn.access")
assert access.propagate is False # uvicorn's default, the problem
configure_logging("json", "info")
assert access.propagate is True
assert access.handlers == []
def test_json_mode_installs_json_formatter(self):
configure_logging("json", "info")
assert isinstance(logging.getLogger().handlers[0].formatter, JSONFormatter)
def test_text_mode_does_not(self):
configure_logging("text", "info")
assert not isinstance(logging.getLogger().handlers[0].formatter, JSONFormatter)
def test_is_idempotent(self):
# Called at import and again in lifespan; a second call must replace the
# handler, not add one, or every line is emitted twice.
configure_logging("json", "info")
configure_logging("json", "info")
assert len(logging.getLogger().handlers) == 1
def test_respects_log_level(self):
configure_logging("json", "warning")
assert logging.getLogger().level == logging.WARNING
class TestSettings:
def test_defaults_to_text(self):
# A JSON-only default would make local development worse.
assert Settings(database_url="sqlite+aiosqlite:///:memory:").log_format == "text"
def test_reads_log_format_env(self, monkeypatch):
monkeypatch.setenv("LOG_FORMAT", "json")
assert Settings(database_url="sqlite+aiosqlite:///:memory:").log_format == "json"

185
tests/test_rate_limit.py Normal file
View File

@@ -0,0 +1,185 @@
"""
Rate-limiting tests.
The limiter guards the unauthenticated `/auth/*` edge — the only routes that
must answer before an identity exists. Everything else is owner-gated, so a
limit there would mostly throttle the single legitimate operator.
The properties worth pinning: the cap actually blocks, windows expire, clients
and routes don't share a bucket, the store can't grow without bound, and
identity comes from the socket peer rather than a spoofable header.
"""
import httpx
import pytest
from fastapi import Depends, FastAPI
import main
from core.rate_limit import RateLimiter, client_key, get_limiter, rate_limit
@pytest.fixture(autouse=True)
def _clean_limiter():
get_limiter().reset()
yield
get_limiter().reset()
class TestRateLimiter:
def test_allows_up_to_the_limit(self):
rl = RateLimiter(limit=3, window_seconds=60)
assert [rl.check("k", now=100.0)[0] for _ in range(3)] == [True, True, True]
def test_blocks_past_the_limit(self):
rl = RateLimiter(limit=3, window_seconds=60)
for _ in range(3):
rl.check("k", now=100.0)
allowed, retry_after = rl.check("k", now=100.0)
assert allowed is False
assert retry_after > 0
def test_window_expiry_resets_the_count(self):
rl = RateLimiter(limit=2, window_seconds=60)
rl.check("k", now=100.0)
rl.check("k", now=100.0)
assert rl.check("k", now=100.0)[0] is False
# A full window later, the caller is welcome again.
assert rl.check("k", now=161.0)[0] is True
def test_retry_after_shrinks_as_the_window_drains(self):
rl = RateLimiter(limit=1, window_seconds=60)
rl.check("k", now=100.0)
early = rl.check("k", now=110.0)[1]
late = rl.check("k", now=150.0)[1]
assert early > late >= 1
def test_clients_do_not_share_a_bucket(self):
rl = RateLimiter(limit=1, window_seconds=60)
assert rl.check("auth:me:1.1.1.1", now=100.0)[0] is True
# A different client must be unaffected by the first one's usage.
assert rl.check("auth:me:2.2.2.2", now=100.0)[0] is True
def test_routes_do_not_share_a_bucket(self):
rl = RateLimiter(limit=1, window_seconds=60)
assert rl.check("auth:me:1.1.1.1", now=100.0)[0] is True
assert rl.check("auth:callback:1.1.1.1", now=100.0)[0] is True
def test_per_call_limit_overrides_the_default(self):
rl = RateLimiter(limit=100, window_seconds=60)
rl.check("k", limit=1, now=100.0)
assert rl.check("k", limit=1, now=100.0)[0] is False
def test_bucket_store_is_bounded(self):
# Otherwise a spray of source addresses is itself a memory exhaustion
# vector — the thing the limiter exists to prevent.
rl = RateLimiter(limit=5, window_seconds=60, max_clients=10)
for i in range(50):
rl.check(f"client-{i}", now=100.0 + i)
assert len(rl._buckets) <= 10
def test_eviction_drops_oldest_first(self):
rl = RateLimiter(limit=5, window_seconds=600, max_clients=3)
for i in range(4):
rl.check(f"client-{i}", now=100.0 + i)
assert "client-0" not in rl._buckets
assert "client-3" in rl._buckets
class TestClientKey:
def _request(self, peer: str | None, headers: dict | None = None):
scope = {
"type": "http",
"method": "GET",
"path": "/auth/me",
"headers": [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()],
"client": (peer, 12345) if peer else None,
}
from starlette.requests import Request
return Request(scope)
def test_uses_the_socket_peer(self):
assert client_key(self._request("10.0.0.5"), "auth:me") == "auth:me:10.0.0.5"
def test_ignores_x_forwarded_for(self):
# Trusting a spoofable header would let one client present as
# thousands, making the limiter worse than useless.
key = client_key(
self._request("10.0.0.5", {"X-Forwarded-For": "1.2.3.4"}), "auth:me"
)
assert key == "auth:me:10.0.0.5"
assert "1.2.3.4" not in key
def test_missing_peer_does_not_crash(self):
assert client_key(self._request(None), "auth:me") == "auth:me:unknown"
class TestDependency:
"""The FastAPI integration: a 429 with a Retry-After header."""
def _app(self, limit=2):
app = FastAPI()
@app.get("/limited", dependencies=[Depends(rate_limit("test", limit=limit))])
async def limited():
return {"ok": True}
@app.get("/unlimited")
async def unlimited():
return {"ok": True}
return app
async def _get(self, app, path, n=1):
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
return [await c.get(path) for _ in range(n)]
async def test_returns_429_past_the_limit(self):
responses = await self._get(self._app(limit=2), "/limited", n=3)
assert [r.status_code for r in responses] == [200, 200, 429]
async def test_429_carries_retry_after(self):
responses = await self._get(self._app(limit=1), "/limited", n=2)
blocked = responses[-1]
assert blocked.status_code == 429
assert int(blocked.headers["retry-after"]) >= 1
async def test_unlimited_routes_are_untouched(self):
responses = await self._get(self._app(limit=1), "/unlimited", n=10)
assert {r.status_code for r in responses} == {200}
class TestAuthRoutesAreLimited:
"""The wiring: the unauthenticated edge is covered, the rest is not."""
def _is_limited(self, path: str) -> bool:
"""True if the route carries a dependency built by `rate_limit`.
Identified by the closure's qualname rather than a string search, so
this can't pass on an unrelated dependency that happens to stringify
similarly.
"""
for route in main.app.routes:
if getattr(route, "path", None) != path:
continue
return any(
getattr(d.dependency, "__qualname__", "").startswith("rate_limit")
for d in getattr(route, "dependencies", [])
)
raise AssertionError(f"route {path} not found")
@pytest.mark.parametrize(
"path", ["/auth/login", "/auth/callback", "/auth/me", "/auth/refresh-callback"]
)
def test_unauthenticated_auth_routes_are_limited(self, path):
assert self._is_limited(path)
def test_owner_gated_routes_are_not_limited(self):
# They're already behind is_owner; limiting them would throttle the
# single legitimate operator.
assert not self._is_limited("/api/v1/calls/active")
def test_logout_is_not_limited(self):
# Pure redirect builder — no I/O, nothing to exhaust.
assert not self._is_limited("/auth/logout")