liqiang
b119135836
Build latest book artifacts / build (push) Canceled after 0s
dependency resolution / resolve (3.11) (push) Canceled after 0s
dependency resolution / resolve (3.13) (push) Canceled after 0s
deploy-pages / build (push) Canceled after 0s
deploy-pages / deploy (push) Canceled after 0s
i18n consistency check / check (push) Canceled after 0s
provider adoption tests / test (chapter2/context-compression) (push) Canceled after 0s
provider adoption tests / test (chapter2/prompt-injection) (push) Canceled after 0s
provider adoption tests / test (chapter2/system-hint) (push) Canceled after 0s
provider adoption tests / test (chapter3/log-sanitization) (push) Canceled after 0s
web-search-agent tests / test (push) Canceled after 0s
web-search-agent tests / agentbook (push) Canceled after 0s
699 lines
27 KiB
Python
699 lines
27 KiB
Python
"""Browser-to-aiortc voice Agent for Phone Agent add-on.
|
|
|
|
The browser microphone is the only user-semantic input. The server buffers its
|
|
decoded RTP audio, runs real Whisper ASR, sends that transcript to a real external
|
|
LLM, synthesizes the Agent reply, and places the resulting PCM on the WebRTC
|
|
downlink track. The data channel carries captions and control events only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import uuid
|
|
from collections import Counter, deque
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
from agent import CallPlan, conversation_turn, direct_plan, react_plan
|
|
from aiortc import AudioStreamTrack, RTCPeerConnection, RTCSessionDescription
|
|
from aiortc.mediastreams import MediaStreamError
|
|
from av import AudioFrame, AudioResampler
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI, HTTPException, Request, Response
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field, model_validator
|
|
from speech import SystemSpeechSynthesizer, WhisperASR, sha256_bytes
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
STATIC = HERE / "static"
|
|
load_dotenv(HERE / ".env")
|
|
CALLS: dict[str, dict[str, Any]] = {}
|
|
PEERS: dict[str, RTCPeerConnection] = {}
|
|
RUNTIMES: dict[str, CallRuntime] = {}
|
|
MAX_MICROPHONE_PCM_BYTES = 16_000 * 2 * 60
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _sha256_text(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _safe_error(exc: BaseException) -> str:
|
|
text = f"{type(exc).__name__}: {exc}"
|
|
for name, secret in os.environ.items():
|
|
if (
|
|
any(marker in name.upper() for marker in ("KEY", "TOKEN", "SECRET", "PASSWORD"))
|
|
and len(secret) >= 8
|
|
):
|
|
text = text.replace(secret, "[REDACTED]")
|
|
return text[:1000]
|
|
|
|
|
|
class CreateCall(BaseModel):
|
|
mode: Literal["direct", "react"]
|
|
task: str = Field(default="", max_length=4000)
|
|
callee_name: str = Field(default="", max_length=200)
|
|
goal: str = Field(default="", max_length=2000)
|
|
context: str = Field(default="", max_length=4000)
|
|
instructions: str = Field(default="", max_length=4000)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_arm(self) -> CreateCall:
|
|
if self.mode == "react" and not self.task.strip():
|
|
raise ValueError("the ReAct arm requires a natural-language task")
|
|
if self.mode == "direct":
|
|
missing = [
|
|
name
|
|
for name in ("callee_name", "goal", "context", "instructions")
|
|
if not getattr(self, name).strip()
|
|
]
|
|
if missing:
|
|
raise ValueError("the direct arm requires: " + ", ".join(missing))
|
|
return self
|
|
|
|
|
|
class EventEnvelope(BaseModel):
|
|
event: dict[str, Any]
|
|
|
|
|
|
class CompleteTask(BaseModel):
|
|
result: str = Field(min_length=1, max_length=2000)
|
|
appointment_time: str = Field(min_length=1, max_length=300)
|
|
confirmation_number: str = Field(min_length=1, max_length=300)
|
|
notes: str = Field(min_length=1, max_length=2000)
|
|
|
|
|
|
class FinishCall(BaseModel):
|
|
reason: str = Field(default="user_hangup", max_length=200)
|
|
|
|
|
|
@dataclass
|
|
class QueuedUtterance:
|
|
receipt: dict[str, Any]
|
|
pcm: bytes
|
|
offset: int = 0
|
|
|
|
|
|
class SynthesizedAudioTrack(AudioStreamTrack):
|
|
"""Continuous WebRTC audio track whose non-silent segments are real TTS PCM."""
|
|
|
|
def __init__(self, record: dict[str, Any]) -> None:
|
|
super().__init__()
|
|
self.record = record
|
|
self.queue: deque[QueuedUtterance] = deque()
|
|
|
|
def enqueue(self, pcm: bytes, receipt: dict[str, Any]) -> None:
|
|
if not pcm:
|
|
raise ValueError("cannot enqueue empty TTS audio")
|
|
receipt["enqueued_on_webrtc_track"] = True
|
|
receipt["transmitted_samples"] = 0
|
|
receipt["delivery_complete"] = False
|
|
self.queue.append(QueuedUtterance(receipt=receipt, pcm=pcm))
|
|
|
|
async def recv(self) -> AudioFrame:
|
|
frame = await super().recv()
|
|
target = bytearray(frame.samples * 2)
|
|
cursor = 0
|
|
while cursor < len(target) and self.queue:
|
|
utterance = self.queue[0]
|
|
available = len(utterance.pcm) - utterance.offset
|
|
take = min(len(target) - cursor, available)
|
|
target[cursor : cursor + take] = utterance.pcm[
|
|
utterance.offset : utterance.offset + take
|
|
]
|
|
utterance.offset += take
|
|
cursor += take
|
|
samples = take // 2
|
|
utterance.receipt["transmitted_samples"] += samples
|
|
self.record["transport"]["server_sent_tts_samples"] += samples
|
|
if utterance.offset == len(utterance.pcm):
|
|
utterance.receipt["delivery_complete"] = True
|
|
utterance.receipt["delivered_pcm_sha256"] = utterance.receipt["pcm_sha256"]
|
|
self.queue.popleft()
|
|
frame.planes[0].update(bytes(target))
|
|
return frame
|
|
|
|
|
|
class CallRuntime:
|
|
def __init__(self, record: dict[str, Any]) -> None:
|
|
self.record = record
|
|
self.tts = SystemSpeechSynthesizer()
|
|
self.asr = WhisperASR()
|
|
self.output_track = SynthesizedAudioTrack(record)
|
|
self.microphone_pcm = bytearray()
|
|
self.resampler = AudioResampler(format="s16", layout="mono", rate=16_000)
|
|
self.channel: Any | None = None
|
|
self.commit_started = False
|
|
|
|
|
|
app = FastAPI(title="Phone Agent add-on WebRTC Voice Agent", version="3.0")
|
|
app.mount("/static", StaticFiles(directory=STATIC), name="static")
|
|
|
|
|
|
def _record(call_id: str) -> dict[str, Any]:
|
|
record = CALLS.get(call_id)
|
|
if record is None:
|
|
raise HTTPException(status_code=404, detail="unknown call")
|
|
return record
|
|
|
|
|
|
def _public(record: dict[str, Any]) -> dict[str, Any]:
|
|
return json.loads(json.dumps(record, ensure_ascii=False))
|
|
|
|
|
|
def _hash_is_valid(value: Any) -> bool:
|
|
return (
|
|
isinstance(value, str)
|
|
and len(value) == 64
|
|
and all(character in "0123456789abcdef" for character in value)
|
|
)
|
|
|
|
|
|
def _llm_receipt_is_real(receipt: dict[str, Any]) -> bool:
|
|
usage = receipt.get("usage") or {}
|
|
return bool(
|
|
receipt.get("execution") == "real_external_llm"
|
|
and receipt.get("external_request_completed") is True
|
|
and receipt.get("provider_response_id")
|
|
and receipt.get("provider_model")
|
|
and receipt.get("finish_reason")
|
|
and int(usage.get("total_tokens", 0)) > 0
|
|
and float(receipt.get("latency_seconds", 0)) > 0
|
|
and _hash_is_valid(receipt.get("request_sha256"))
|
|
and _hash_is_valid(receipt.get("raw_response_sha256"))
|
|
and _hash_is_valid(receipt.get("response_content_sha256"))
|
|
and receipt.get("fallback_used") is False
|
|
and receipt.get("mock") is False
|
|
and receipt.get("probe_only") is False
|
|
and receipt.get("credential_fields_retained") is False
|
|
)
|
|
|
|
|
|
def _acceptance(record: dict[str, Any]) -> dict[str, Any]:
|
|
transport = record["transport"]
|
|
stats = transport["rtc_stats"]
|
|
transcript = record["transcript"]
|
|
llm_receipts = record["models"]["llm_receipts"]
|
|
asr_receipts = record["models"]["asr_receipts"]
|
|
tts_receipts = record["models"]["tts_receipts"]
|
|
planning_receipts = [item for item in llm_receipts if item.get("purpose") == "react_planning"]
|
|
dialogue_receipts = [
|
|
item for item in llm_receipts if item.get("purpose") == "post_asr_dialogue"
|
|
]
|
|
all_model_receipts = [*llm_receipts, *asr_receipts, *tts_receipts]
|
|
completion = record.get("completion") or {}
|
|
first_agent = next((turn for turn in transcript if turn.get("speaker") == "agent"), {})
|
|
checks = {
|
|
"sdp_offer_answer_negotiated": bool(transport["sdp_negotiated"]),
|
|
"ice_connected": bool(transport["ice_connected_observed"]),
|
|
"data_channel_open": bool(transport["data_channel_open"]),
|
|
"browser_microphone_track": bool(transport["local_audio_track"]),
|
|
"server_downlink_audio_track": bool(transport["remote_audio_track"]),
|
|
"outbound_audio_rtp": int(stats["outbound_packets"]) > 0
|
|
and int(stats["outbound_bytes"]) > 0,
|
|
"inbound_audio_rtp": int(stats["inbound_packets"]) > 0 and int(stats["inbound_bytes"]) > 0,
|
|
"server_buffered_microphone_rtp": (
|
|
int(transport["server_received_audio_frames"]) > 0
|
|
and int(transport["server_received_audio_pcm_bytes"]) > 0
|
|
),
|
|
"real_asr_consumed_microphone_audio": bool(
|
|
asr_receipts
|
|
and all(
|
|
item.get("execution") == "real_local_inference"
|
|
and item.get("input_source") == "browser_microphone_rtp"
|
|
and _hash_is_valid(item.get("input_wav_sha256"))
|
|
and _hash_is_valid(item.get("checkpoint_sha256"))
|
|
and item.get("fallback_used") is False
|
|
for item in asr_receipts
|
|
)
|
|
),
|
|
"external_react_planner_or_fixed_direct_control": (
|
|
record["mode"] == "direct"
|
|
and record["input_contract"]["fields_supplied_by_caller"]
|
|
== ["callee_name", "goal", "context", "instructions"]
|
|
)
|
|
or (
|
|
record["mode"] == "react"
|
|
and len(planning_receipts) == 1
|
|
and _llm_receipt_is_real(planning_receipts[0])
|
|
),
|
|
"real_external_post_asr_dialogue": len(dialogue_receipts) == 1
|
|
and _llm_receipt_is_real(dialogue_receipts[0]),
|
|
"real_tts_assets_synthesized": len(tts_receipts) >= 2
|
|
and all(
|
|
item.get("execution") == "real_speech_synthesis"
|
|
and int(item.get("sample_count", 0)) > 0
|
|
and _hash_is_valid(item.get("wav_sha256"))
|
|
and _hash_is_valid(item.get("pcm_sha256"))
|
|
and item.get("fallback_used") is False
|
|
for item in tts_receipts
|
|
),
|
|
"tts_audio_transmitted_on_downlink": len(tts_receipts) >= 2
|
|
and all(
|
|
item.get("enqueued_on_webrtc_track") is True
|
|
and item.get("delivery_complete") is True
|
|
and int(item.get("transmitted_samples", 0)) == int(item.get("sample_count", -1))
|
|
and item.get("delivered_pcm_sha256") == item.get("pcm_sha256")
|
|
for item in tts_receipts
|
|
),
|
|
"media_is_canonical_transcript_source": bool(transcript)
|
|
and all(
|
|
(turn.get("speaker") == "user" and turn.get("source") == "asr.microphone_rtp")
|
|
or (turn.get("speaker") == "agent" and turn.get("source") == "tts.webrtc_downlink")
|
|
for turn in transcript
|
|
),
|
|
"data_channel_is_control_and_caption_only": int(
|
|
record["event_counts"].get("semantic_user_messages", 0)
|
|
)
|
|
== 0,
|
|
"missing_fields_were_clarified_aloud": (
|
|
first_agent.get("purpose") == "missing_field_clarification"
|
|
and first_agent.get("source") == "tts.webrtc_downlink"
|
|
and (record["mode"] == "direct" or bool(record["plan"].get("missing_information")))
|
|
),
|
|
"explicit_confirmation_observed": record.get("explicit_confirmation_observed") is True,
|
|
"structured_completion_saved": bool(
|
|
completion.get("appointment_time")
|
|
and completion.get("confirmation_number")
|
|
and completion.get("result")
|
|
),
|
|
"no_mock_probe_or_fallback": bool(all_model_receipts)
|
|
and all(
|
|
item.get("mock") is False
|
|
and item.get("probe_only") is False
|
|
and item.get("fallback_used") is False
|
|
for item in all_model_receipts
|
|
),
|
|
"privacy_boundary_preserved": record["privacy"]["private_audio_retained"] is False
|
|
and record["privacy"]["private_transcripts_retained"] is False,
|
|
}
|
|
return {"checks": checks, "passed": all(checks.values()) and not record["errors"]}
|
|
|
|
|
|
def _retained_media_path(record: dict[str, Any], filename: str) -> Path | None:
|
|
if os.getenv("PHONE_SAFE_SYNTHETIC_ACCEPTANCE") != "1":
|
|
return None
|
|
root = os.getenv("PHONE_EVIDENCE_DIR")
|
|
if not root:
|
|
raise RuntimeError("safe acceptance media retention requires PHONE_EVIDENCE_DIR")
|
|
path = Path(root).resolve() / "media" / record["mode"] / filename
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def _relative_evidence_path(path: Path | None) -> str | None:
|
|
if path is None:
|
|
return None
|
|
root = Path(os.environ["PHONE_EVIDENCE_DIR"]).resolve()
|
|
return str(path.relative_to(root))
|
|
|
|
|
|
@app.get("/")
|
|
async def index() -> FileResponse:
|
|
return FileResponse(STATIC / "index.html")
|
|
|
|
|
|
@app.get("/api/health")
|
|
async def health() -> dict[str, Any]:
|
|
return {
|
|
"ok": True,
|
|
"experiment": "phone-agent",
|
|
"model_provider": os.getenv("PHONE_MODEL_PROVIDER", "ark"),
|
|
"model_credential_present": bool(
|
|
os.getenv("ARK_API_KEY")
|
|
or os.getenv("OPENAI_API_KEY")
|
|
or os.getenv("OPENROUTER_API_KEY")
|
|
),
|
|
"speech_paths": "browser microphone RTP -> Whisper ASR; system TTS -> WebRTC RTP",
|
|
}
|
|
|
|
|
|
@app.post("/api/calls")
|
|
async def create_call(request: CreateCall) -> dict[str, Any]:
|
|
try:
|
|
if request.mode == "direct":
|
|
plan = direct_plan(
|
|
callee_name=request.callee_name,
|
|
goal=request.goal,
|
|
context=request.context,
|
|
instructions=request.instructions,
|
|
)
|
|
supplied = ["callee_name", "goal", "context", "instructions"]
|
|
else:
|
|
plan = await asyncio.to_thread(react_plan, request.task)
|
|
supplied = ["task"]
|
|
except (RuntimeError, TypeError, ValueError, json.JSONDecodeError) as exc:
|
|
raise HTTPException(status_code=422, detail=_safe_error(exc)) from exc
|
|
|
|
plan_dict = plan.to_dict()
|
|
planner_receipt = plan_dict.pop("planner_receipt", None)
|
|
llm_receipts = [planner_receipt] if planner_receipt else []
|
|
call_id = "rtc_" + uuid.uuid4().hex[:20]
|
|
safe_acceptance = os.getenv("PHONE_SAFE_SYNTHETIC_ACCEPTANCE") == "1"
|
|
record = {
|
|
"schema_version": 3,
|
|
"experiment": "phone-agent",
|
|
"call_id": call_id,
|
|
"created_at_utc": _now(),
|
|
"finished_at_utc": None,
|
|
"status": "planned",
|
|
"mode": request.mode,
|
|
"input_contract": {
|
|
"fields_supplied_by_caller": supplied,
|
|
"natural_language_task": request.task,
|
|
},
|
|
"plan": plan_dict,
|
|
"models": {
|
|
"planner": plan.planner_model,
|
|
"dialogue_models": [],
|
|
"llm_receipts": llm_receipts,
|
|
"asr_receipts": [],
|
|
"tts_receipts": [],
|
|
},
|
|
"transport": {
|
|
"kind": "webrtc",
|
|
"pstn_used": False,
|
|
"e164_required": False,
|
|
"sdp_negotiated": False,
|
|
"offer_sha256": None,
|
|
"answer_sha256": None,
|
|
"ice_connection_state": "new",
|
|
"ice_connected_observed": False,
|
|
"data_channel_open": False,
|
|
"local_audio_track": False,
|
|
"remote_audio_track": False,
|
|
"rtc_stats": {
|
|
"inbound_packets": 0,
|
|
"inbound_bytes": 0,
|
|
"outbound_packets": 0,
|
|
"outbound_bytes": 0,
|
|
},
|
|
"server_received_audio_frames": 0,
|
|
"server_received_audio_samples": 0,
|
|
"server_received_audio_pcm_bytes": 0,
|
|
"server_sent_tts_samples": 0,
|
|
},
|
|
"privacy": {
|
|
"safe_synthetic_acceptance": safe_acceptance,
|
|
"private_audio_retained": False,
|
|
"private_transcripts_retained": False,
|
|
"safe_synthetic_media_retained": safe_acceptance,
|
|
},
|
|
"event_counts": {},
|
|
"transcript": [],
|
|
"explicit_confirmation_observed": False,
|
|
"completion": None,
|
|
"errors": [],
|
|
"acceptance": {"checks": {}, "passed": False},
|
|
}
|
|
CALLS[call_id] = record
|
|
return {"call_id": call_id, "join_url": f"/?call_id={call_id}", "plan": plan_dict}
|
|
|
|
|
|
async def _consume_microphone(track: Any, runtime: CallRuntime) -> None:
|
|
record = runtime.record
|
|
try:
|
|
while True:
|
|
frame = await track.recv()
|
|
record["transport"]["server_received_audio_frames"] += 1
|
|
record["transport"]["server_received_audio_samples"] += int(
|
|
getattr(frame, "samples", 0)
|
|
)
|
|
for resampled in runtime.resampler.resample(frame):
|
|
# Packed mono s16 has two bytes per sample. Reading the plane
|
|
# directly avoids imposing NumPy on this transport-only stack.
|
|
pcm = bytes(resampled.planes[0])[: int(resampled.samples) * 2]
|
|
if len(runtime.microphone_pcm) + len(pcm) > MAX_MICROPHONE_PCM_BYTES:
|
|
raise RuntimeError("microphone buffer exceeded the 60-second safety limit")
|
|
runtime.microphone_pcm.extend(pcm)
|
|
record["transport"]["server_received_audio_pcm_bytes"] = len(runtime.microphone_pcm)
|
|
except MediaStreamError:
|
|
return
|
|
except Exception as exc: # noqa: BLE001 - evidence must retain asynchronous media failures
|
|
record["errors"].append(
|
|
{"at": _now(), "stage": "microphone_rtp", "message": _safe_error(exc)}
|
|
)
|
|
|
|
|
|
async def _send_agent(
|
|
runtime: CallRuntime,
|
|
text: str,
|
|
*,
|
|
purpose: str,
|
|
) -> None:
|
|
cleaned = text.strip()
|
|
if not cleaned:
|
|
raise ValueError("Agent TTS text is empty")
|
|
index = len(runtime.record["models"]["tts_receipts"]) + 1
|
|
speech = await asyncio.to_thread(runtime.tts.synthesize, cleaned)
|
|
path = _retained_media_path(runtime.record, f"agent_{index:02d}.wav")
|
|
if path is not None:
|
|
path.write_bytes(speech.wav)
|
|
if sha256_bytes(path.read_bytes()) != speech.receipt["wav_sha256"]:
|
|
raise RuntimeError("retained TTS asset hash mismatch")
|
|
receipt = {
|
|
**speech.receipt,
|
|
"utterance_id": f"tts_{index:02d}",
|
|
"purpose": purpose,
|
|
"retained_safe_fixture_path": _relative_evidence_path(path),
|
|
"text_sha256": hashlib.sha256(cleaned.encode("utf-8")).hexdigest(),
|
|
}
|
|
runtime.record["models"]["tts_receipts"].append(receipt)
|
|
runtime.output_track.enqueue(speech.pcm, receipt)
|
|
runtime.record["transcript"].append(
|
|
{
|
|
"speaker": "agent",
|
|
"text": cleaned,
|
|
"source": "tts.webrtc_downlink",
|
|
"utterance_id": receipt["utterance_id"],
|
|
"purpose": purpose,
|
|
}
|
|
)
|
|
channel = runtime.channel
|
|
if channel is not None and channel.readyState == "open":
|
|
channel.send(
|
|
json.dumps(
|
|
{
|
|
"type": "agent.caption",
|
|
"text": cleaned,
|
|
"utterance_id": receipt["utterance_id"],
|
|
"source": "accessibility_mirror_of_tts",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
|
|
|
|
async def _commit_microphone(runtime: CallRuntime) -> None:
|
|
if runtime.commit_started:
|
|
raise RuntimeError("microphone audio was already committed")
|
|
runtime.commit_started = True
|
|
record = runtime.record
|
|
pcm = bytes(runtime.microphone_pcm)
|
|
path = _retained_media_path(record, "microphone_rtp_asr_input.wav")
|
|
transcript, asr_receipt = await asyncio.to_thread(
|
|
runtime.asr.transcribe, pcm, retained_wav_path=path
|
|
)
|
|
asr_receipt["retained_safe_fixture_path"] = _relative_evidence_path(path)
|
|
if path is not None and sha256_bytes(path.read_bytes()) != asr_receipt["input_wav_sha256"]:
|
|
raise RuntimeError("retained microphone/ASR input hash mismatch")
|
|
record["models"]["asr_receipts"].append(asr_receipt)
|
|
record["transcript"].append(
|
|
{
|
|
"speaker": "user",
|
|
"text": transcript,
|
|
"source": "asr.microphone_rtp",
|
|
"asr_receipt_index": len(record["models"]["asr_receipts"]) - 1,
|
|
}
|
|
)
|
|
channel = runtime.channel
|
|
if channel is not None and channel.readyState == "open":
|
|
channel.send(
|
|
json.dumps(
|
|
{
|
|
"type": "user.caption",
|
|
"text": transcript,
|
|
"source": "accessibility_mirror_of_asr",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
plan = CallPlan(**record["plan"])
|
|
dialogue = await asyncio.to_thread(
|
|
conversation_turn,
|
|
plan,
|
|
list(record["transcript"][:-1]),
|
|
transcript,
|
|
)
|
|
receipt = dialogue["llm_receipt"]
|
|
record["models"]["llm_receipts"].append(receipt)
|
|
model = dialogue["dialogue_model"]
|
|
if model not in record["models"]["dialogue_models"]:
|
|
record["models"]["dialogue_models"].append(model)
|
|
record["explicit_confirmation_observed"] = dialogue["explicit_confirmation_observed"] is True
|
|
if not dialogue["should_complete"]:
|
|
raise RuntimeError("canonical call did not reach explicit structured completion")
|
|
completion = CompleteTask(**dialogue["completion"])
|
|
await _send_agent(runtime, dialogue["assistant_message"], purpose="confirmed_completion")
|
|
record["completion"] = {
|
|
**completion.model_dump(),
|
|
"saved_at_utc": _now(),
|
|
"tool": "complete_task",
|
|
}
|
|
if channel is not None and channel.readyState == "open":
|
|
channel.send(json.dumps({"type": "tool.result", "name": "complete_task", "saved": True}))
|
|
|
|
|
|
async def _handle_data_message(runtime: CallRuntime, raw: Any) -> None:
|
|
record = runtime.record
|
|
try:
|
|
message = json.loads(raw) if isinstance(raw, str) else {}
|
|
message_type = message.get("type")
|
|
if message_type == "client.ready":
|
|
record["transport"]["data_channel_open"] = True
|
|
if not record["transcript"]:
|
|
await _send_agent(
|
|
runtime,
|
|
record["plan"]["opening_line"],
|
|
purpose="missing_field_clarification",
|
|
)
|
|
return
|
|
if message_type == "client.audio.commit":
|
|
record["event_counts"]["client.audio.commit"] = (
|
|
int(record["event_counts"].get("client.audio.commit", 0)) + 1
|
|
)
|
|
await _commit_microphone(runtime)
|
|
return
|
|
if message_type in {"user.message", "client.user_text"}:
|
|
record["event_counts"]["semantic_user_messages"] = (
|
|
int(record["event_counts"].get("semantic_user_messages", 0)) + 1
|
|
)
|
|
raise RuntimeError("semantic user text is forbidden; speak over the microphone track")
|
|
except Exception as exc: # noqa: BLE001 - asynchronous failures belong in the evidence record
|
|
error = {"at": _now(), "stage": "voice_agent_loop", "message": _safe_error(exc)}
|
|
record["errors"].append(error)
|
|
if runtime.channel is not None and runtime.channel.readyState == "open":
|
|
runtime.channel.send(json.dumps({"type": "error", "error": error}, ensure_ascii=False))
|
|
|
|
|
|
@app.post("/api/calls/{call_id}/session")
|
|
async def negotiate(call_id: str, request: Request) -> Response:
|
|
record = _record(call_id)
|
|
content_type = request.headers.get("content-type", "")
|
|
if not content_type.startswith("application/sdp"):
|
|
raise HTTPException(status_code=415, detail="expected application/sdp")
|
|
offer = (await request.body()).decode("utf-8", errors="strict")
|
|
if not offer.startswith("v=0") or len(offer) > 1_000_000:
|
|
raise HTTPException(status_code=400, detail="invalid SDP offer")
|
|
try:
|
|
runtime = CallRuntime(record)
|
|
except RuntimeError as exc:
|
|
raise HTTPException(status_code=503, detail=_safe_error(exc)) from exc
|
|
RUNTIMES[call_id] = runtime
|
|
pc = RTCPeerConnection()
|
|
PEERS[call_id] = pc
|
|
pc.addTrack(runtime.output_track)
|
|
|
|
@pc.on("track")
|
|
def on_track(track: Any) -> None:
|
|
if track.kind == "audio":
|
|
record["transport"]["local_audio_track"] = True
|
|
asyncio.create_task(_consume_microphone(track, runtime))
|
|
|
|
@pc.on("datachannel")
|
|
def on_datachannel(channel: Any) -> None:
|
|
runtime.channel = channel
|
|
|
|
@channel.on("message")
|
|
def on_message(message: Any) -> None:
|
|
asyncio.create_task(_handle_data_message(runtime, message))
|
|
|
|
@pc.on("connectionstatechange")
|
|
async def on_connectionstatechange() -> None:
|
|
state = pc.connectionState
|
|
if state in {"connected", "completed"}:
|
|
record["transport"]["ice_connected_observed"] = True
|
|
record["transport"]["ice_connection_state"] = state
|
|
elif not record["transport"]["ice_connected_observed"]:
|
|
record["transport"]["ice_connection_state"] = state
|
|
if state in {"failed", "closed"}:
|
|
await pc.close()
|
|
|
|
await pc.setRemoteDescription(RTCSessionDescription(sdp=offer, type="offer"))
|
|
answer_description = await pc.createAnswer()
|
|
await pc.setLocalDescription(answer_description)
|
|
answer = pc.localDescription.sdp
|
|
record["status"] = "connected"
|
|
record["transport"].update(
|
|
{
|
|
"sdp_negotiated": True,
|
|
"offer_sha256": _sha256_text(offer),
|
|
"answer_sha256": _sha256_text(answer),
|
|
"remote_audio_track": True,
|
|
}
|
|
)
|
|
return Response(content=answer, media_type="application/sdp")
|
|
|
|
|
|
@app.post("/api/calls/{call_id}/events")
|
|
async def save_event(call_id: str, envelope: EventEnvelope) -> dict[str, bool]:
|
|
record = _record(call_id)
|
|
event = envelope.event
|
|
event_type = str(event.get("type", "unknown"))[:200]
|
|
counts = Counter(record["event_counts"])
|
|
counts[event_type] += 1
|
|
record["event_counts"] = dict(sorted(counts.items()))
|
|
if event_type == "rtc.ready":
|
|
state = str(event.get("ice_connection_state", "unknown"))[:30]
|
|
if state in {"connected", "completed"}:
|
|
record["transport"]["ice_connected_observed"] = True
|
|
record["transport"]["ice_connection_state"] = state
|
|
for field, event_field in (
|
|
("data_channel_open", "data_channel_open"),
|
|
("local_audio_track", "local_audio_track"),
|
|
("remote_audio_track", "remote_audio_track"),
|
|
):
|
|
record["transport"][field] = bool(record["transport"][field] or event.get(event_field))
|
|
elif event_type == "rtc.stats":
|
|
stats = record["transport"]["rtc_stats"]
|
|
for field in stats:
|
|
stats[field] = max(int(stats[field]), max(0, int(event.get(field, 0))))
|
|
state = str(event.get("ice_connection_state", ""))[:30]
|
|
if state in {"connected", "completed"}:
|
|
record["transport"]["ice_connected_observed"] = True
|
|
record["transport"]["ice_connection_state"] = state
|
|
elif event_type == "error":
|
|
record["errors"].append({"at": _now(), "stage": "browser", "message": str(event)[:1000]})
|
|
return {"saved": True}
|
|
|
|
|
|
@app.post("/api/calls/{call_id}/finish")
|
|
async def finish(call_id: str, finish_request: FinishCall) -> dict[str, Any]:
|
|
record = _record(call_id)
|
|
record["finished_at_utc"] = _now()
|
|
record["finish_reason"] = finish_request.reason
|
|
record["acceptance"] = _acceptance(record)
|
|
record["status"] = "completed" if record["acceptance"]["passed"] else "ended"
|
|
result = _public(record)
|
|
peer = PEERS.pop(call_id, None)
|
|
RUNTIMES.pop(call_id, None)
|
|
if peer is not None:
|
|
await peer.close()
|
|
return result
|
|
|
|
|
|
@app.get("/api/calls/{call_id}")
|
|
async def get_call(call_id: str) -> dict[str, Any]:
|
|
record = _record(call_id)
|
|
record["acceptance"] = _acceptance(record)
|
|
return _public(record)
|