diff --git a/backend/services/twilio_service.py b/backend/services/twilio_service.py
new file mode 100644
index 0000000..185a807
--- /dev/null
+++ b/backend/services/twilio_service.py
@@ -0,0 +1,100 @@
+"""Twilio call queue and media stream service"""
+
+import time
+import threading
+from typing import Optional
+
+
+class TwilioService:
+    """Manages Twilio call queue, channel allocation, and media streams"""
+
+    FIRST_REAL_CHANNEL = 3
+
+    def __init__(self):
+        self._queue: list[dict] = []
+        self.active_calls: dict[str, dict] = {}
+        self._allocated_channels: set[int] = set()
+        self._caller_counter: int = 0
+        self._lock = threading.Lock()
+
+    def add_to_queue(self, call_sid: str, phone: str):
+        with self._lock:
+            self._queue.append({
+                "call_sid": call_sid,
+                "phone": phone,
+                "queued_at": time.time(),
+            })
+        print(f"[Twilio] Caller {phone} added to queue (SID: {call_sid})")
+
+    def remove_from_queue(self, call_sid: str):
+        with self._lock:
+            self._queue = [c for c in self._queue if c["call_sid"] != call_sid]
+        print(f"[Twilio] Caller {call_sid} removed from queue")
+
+    def get_queue(self) -> list[dict]:
+        now = time.time()
+        with self._lock:
+            return [
+                {
+                    "call_sid": c["call_sid"],
+                    "phone": c["phone"],
+                    "wait_time": int(now - c["queued_at"]),
+                }
+                for c in self._queue
+            ]
+
+    def allocate_channel(self) -> int:
+        with self._lock:
+            ch = self.FIRST_REAL_CHANNEL
+            while ch in self._allocated_channels:
+                ch += 1
+            self._allocated_channels.add(ch)
+            return ch
+
+    def release_channel(self, channel: int):
+        with self._lock:
+            self._allocated_channels.discard(channel)
+
+    def take_call(self, call_sid: str) -> dict:
+        caller = None
+        with self._lock:
+            for c in self._queue:
+                if c["call_sid"] == call_sid:
+                    caller = c
+                    break
+            if caller:
+                self._queue = [c for c in self._queue if c["call_sid"] != call_sid]
+
+        if not caller:
+            raise ValueError(f"Call {call_sid} not in queue")
+
+        channel = self.allocate_channel()
+        self._caller_counter += 1
+        name = f"Caller #{self._caller_counter}"
+
+        call_info = {
+            "call_sid": call_sid,
+            "phone": caller["phone"],
+            "channel": channel,
+            "name": name,
+            "started_at": time.time(),
+        }
+        self.active_calls[call_sid] = call_info
+        print(f"[Twilio] {name} ({caller['phone']}) taken on air — channel {channel}")
+        return call_info
+
+    def hangup(self, call_sid: str):
+        call_info = self.active_calls.pop(call_sid, None)
+        if call_info:
+            self.release_channel(call_info["channel"])
+            print(f"[Twilio] {call_info['name']} hung up — channel {call_info['channel']} released")
+
+    def reset(self):
+        with self._lock:
+            for call_info in self.active_calls.values():
+                self._allocated_channels.discard(call_info["channel"])
+            self._queue.clear()
+            self.active_calls.clear()
+            self._allocated_channels.clear()
+            self._caller_counter = 0
+        print("[Twilio] Service reset")
diff --git a/tests/test_twilio_service.py b/tests/test_twilio_service.py
new file mode 100644
index 0000000..6070c15
--- /dev/null
+++ b/tests/test_twilio_service.py
@@ -0,0 +1,67 @@
+import sys
+sys.path.insert(0, "/Users/lukemacneil/ai-podcast")
+
+from backend.services.twilio_service import TwilioService
+
+
+def test_queue_starts_empty():
+    svc = TwilioService()
+    assert svc.get_queue() == []
+
+
+def test_add_caller_to_queue():
+    svc = TwilioService()
+    svc.add_to_queue("CA123", "+15125550142")
+    q = svc.get_queue()
+    assert len(q) == 1
+    assert q[0]["call_sid"] == "CA123"
+    assert q[0]["phone"] == "+15125550142"
+    assert "wait_time" in q[0]
+
+
+def test_remove_caller_from_queue():
+    svc = TwilioService()
+    svc.add_to_queue("CA123", "+15125550142")
+    svc.remove_from_queue("CA123")
+    assert svc.get_queue() == []
+
+
+def test_allocate_channel():
+    svc = TwilioService()
+    ch1 = svc.allocate_channel()
+    ch2 = svc.allocate_channel()
+    assert ch1 == 3
+    assert ch2 == 4
+    svc.release_channel(ch1)
+    ch3 = svc.allocate_channel()
+    assert ch3 == 3
+
+
+def test_take_call():
+    svc = TwilioService()
+    svc.add_to_queue("CA123", "+15125550142")
+    result = svc.take_call("CA123")
+    assert result["call_sid"] == "CA123"
+    assert result["channel"] >= 3
+    assert svc.get_queue() == []
+    assert svc.active_calls["CA123"]["channel"] == result["channel"]
+
+
+def test_hangup_real_caller():
+    svc = TwilioService()
+    svc.add_to_queue("CA123", "+15125550142")
+    svc.take_call("CA123")
+    ch = svc.active_calls["CA123"]["channel"]
+    svc.hangup("CA123")
+    assert "CA123" not in svc.active_calls
+    assert ch not in svc._allocated_channels
+
+
+def test_caller_counter_increments():
+    svc = TwilioService()
+    svc.add_to_queue("CA1", "+15125550001")
+    svc.add_to_queue("CA2", "+15125550002")
+    r1 = svc.take_call("CA1")
+    r2 = svc.take_call("CA2")
+    assert r1["name"] == "Caller #1"
+    assert r2["name"] == "Caller #2"