349 lines
11 KiB
Go
349 lines
11 KiB
Go
package call
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"slices"
|
|
"strings"
|
|
"time"
|
|
|
|
"ai-operator/internal/ai"
|
|
"ai-operator/internal/asterisk/ari"
|
|
"ai-operator/internal/config"
|
|
"ai-operator/internal/media"
|
|
"ai-operator/internal/media/asteriskws"
|
|
)
|
|
|
|
type ManagerMode string
|
|
|
|
const (
|
|
ManagerModeObserveOnly ManagerMode = "observe_only"
|
|
ManagerModeCallControl ManagerMode = "call_control"
|
|
)
|
|
|
|
type ManagerConfig struct {
|
|
Mode ManagerMode
|
|
AllowedStasisArgs []string
|
|
TestCallHangupAfter time.Duration
|
|
ProductionEnabled bool
|
|
MediaEnabled bool
|
|
MediaTestMode string
|
|
MediaCodec media.Codec
|
|
VoiceInputSampleRate int
|
|
VoiceOutputSampleRate int
|
|
MediaStarter MediaStarter
|
|
VoiceProvider ai.VoiceProvider
|
|
StartDialogue func(context.Context, *CallSession) (string, error)
|
|
HandleVoiceEvent func(context.Context, string, ai.VoiceEvent) (*ai.ToolResult, error)
|
|
EndDialogue func(context.Context, string, string) error
|
|
}
|
|
|
|
type MediaStarter interface {
|
|
StartCallMedia(ctx context.Context, session *CallSession) (MediaClient, error)
|
|
}
|
|
|
|
type diagnosticMediaClient interface {
|
|
GetStatus(ctx context.Context) error
|
|
Events() <-chan asteriskws.ControlEvent
|
|
Stats() media.Stats
|
|
}
|
|
|
|
type playbackActionClient interface {
|
|
PlayChannel(ctx context.Context, channelID string, media string) error
|
|
}
|
|
|
|
type Manager struct {
|
|
cfg ManagerConfig
|
|
actions ari.ActionClient
|
|
store *SessionStore
|
|
logger *slog.Logger
|
|
}
|
|
|
|
func DefaultManagerConfig(mode ManagerMode, hangupAfter time.Duration) ManagerConfig {
|
|
if hangupAfter == 0 {
|
|
hangupAfter = 5 * time.Second
|
|
}
|
|
return ManagerConfig{Mode: mode, AllowedStasisArgs: []string{"test"}, TestCallHangupAfter: hangupAfter, ProductionEnabled: false}
|
|
}
|
|
|
|
func NewManager(cfg ManagerConfig, actions ari.ActionClient, store *SessionStore, logger *slog.Logger) *Manager {
|
|
if store == nil {
|
|
store = NewSessionStore()
|
|
}
|
|
return &Manager{cfg: cfg, actions: actions, store: store, logger: logger}
|
|
}
|
|
|
|
func (m *Manager) Store() *SessionStore { return m.store }
|
|
|
|
func (m *Manager) HandleEvent(ctx context.Context, event ari.Event) error {
|
|
switch e := event.(type) {
|
|
case ari.StasisStartEvent:
|
|
return m.handleStasisStart(ctx, e)
|
|
case ari.StasisEndEvent:
|
|
m.endSession(e.Channel.ID, "stasis_end")
|
|
case ari.ChannelHangupRequestEvent:
|
|
m.endSession(e.Channel.ID, "hangup_request")
|
|
case ari.ChannelDestroyedEvent:
|
|
m.endSession(e.Channel.ID, "channel_destroyed")
|
|
case ari.ChannelStateChangeEvent:
|
|
m.log("channel state changed", "channel_id", e.Channel.ID, "state", e.Channel.State)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (m *Manager) handleStasisStart(ctx context.Context, e ari.StasisStartEvent) error {
|
|
if isGeneratedMediaChannel(e.Channel.ID) || isGeneratedMediaChannel(e.Channel.Name) {
|
|
m.log("media channel stasis start ignored", "channel_id", e.Channel.ID)
|
|
return nil
|
|
}
|
|
route := detectRoute(e.Args)
|
|
session := &CallSession{CallID: e.Channel.ID, AsteriskChannelID: e.Channel.ID, CallerNumber: e.Channel.Caller.Number, CallerName: e.Channel.Caller.Name, State: StateCallStarted, Language: LanguageUnknown, StasisArgs: e.Args, Route: route, StartedAt: time.Now().UTC()}
|
|
m.store.Create(session)
|
|
m.log("stasis start", "channel_id", e.Channel.ID, "caller", config.MaskPhoneNumber(e.Channel.Caller.Number), "route", route, "mode", string(m.cfg.Mode))
|
|
if m.cfg.Mode == ManagerModeObserveOnly {
|
|
m.log("observe-only, no channel control", "channel_id", e.Channel.ID)
|
|
return nil
|
|
}
|
|
if m.cfg.Mode != ManagerModeCallControl {
|
|
return nil
|
|
}
|
|
if route != "test" || !slices.Contains(m.cfg.AllowedStasisArgs, "test") {
|
|
m.log("non-test route rejected", "channel_id", e.Channel.ID, "route", route)
|
|
if m.actions != nil {
|
|
return m.actions.HangupChannel(ctx, e.Channel.ID)
|
|
}
|
|
return nil
|
|
}
|
|
systemPrompt := ""
|
|
if m.cfg.StartDialogue != nil {
|
|
prompt, err := m.cfg.StartDialogue(ctx, session)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
systemPrompt = prompt
|
|
}
|
|
if m.cfg.VoiceProvider != nil {
|
|
if err := m.cfg.VoiceProvider.StartSession(ctx, ai.VoiceSessionConfig{CallID: session.CallID, SystemPrompt: systemPrompt, InputAudioFormat: "pcm16", OutputAudioFormat: "pcm16", InputSampleRate: m.voiceInputSampleRate(), OutputSampleRate: m.voiceOutputSampleRate()}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if m.actions != nil {
|
|
if err := m.actions.AnswerChannel(ctx, e.Channel.ID); err != nil {
|
|
return err
|
|
}
|
|
session.Answered = true
|
|
if m.cfg.MediaTestMode == "playback" {
|
|
m.playDiagnostic(ctx, e.Channel.ID)
|
|
}
|
|
var pump *AudioPump
|
|
if m.cfg.MediaEnabled && m.cfg.MediaStarter != nil {
|
|
mediaCtx, mediaCancel := context.WithCancel(context.Background())
|
|
session.MediaCancel = mediaCancel
|
|
mediaClient, err := m.cfg.MediaStarter.StartCallMedia(mediaCtx, session)
|
|
if err != nil {
|
|
mediaCancel()
|
|
return err
|
|
}
|
|
if mediaClient != nil && m.cfg.VoiceProvider != nil {
|
|
pump = &AudioPump{CallID: session.CallID, Media: mediaClient, Provider: m.cfg.VoiceProvider, AsteriskCodec: m.mediaCodec(), AsteriskSampleRate: m.mediaSampleRate(), ProviderInputSampleRate: m.voiceInputSampleRate(), ProviderOutputSampleRate: m.voiceOutputSampleRate(), Logger: m.logger}
|
|
if err := pump.Start(mediaCtx); err != nil {
|
|
mediaCancel()
|
|
return err
|
|
}
|
|
}
|
|
if mediaClient != nil && m.cfg.MediaTestMode == "tone" {
|
|
m.startMediaDiagnostics(mediaCtx, session.CallID, mediaClient)
|
|
m.startTonePump(mediaCtx, session.CallID, mediaClient)
|
|
}
|
|
}
|
|
if m.cfg.VoiceProvider != nil {
|
|
m.startVoiceEventPump(session.CallID, pump)
|
|
}
|
|
go m.delayedHangup(e.Channel.ID)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (m *Manager) startVoiceEventPump(callID string, pump *AudioPump) {
|
|
if m.cfg.VoiceProvider == nil || m.cfg.HandleVoiceEvent == nil {
|
|
return
|
|
}
|
|
go func() {
|
|
for event := range m.cfg.VoiceProvider.Events() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
if pump != nil {
|
|
if err := pump.HandleProviderEvent(ctx, event); err != nil {
|
|
m.log("audio pump provider event failed", "channel_id", callID, "event_type", event.Type, "error", err)
|
|
}
|
|
}
|
|
result, err := m.cfg.HandleVoiceEvent(ctx, callID, event)
|
|
if err != nil {
|
|
m.log("voice event handling failed", "channel_id", callID, "event_type", event.Type, "error", err)
|
|
}
|
|
if result != nil {
|
|
if err := m.cfg.VoiceProvider.SendToolResult(ctx, *result); err != nil {
|
|
m.log("voice tool result send failed", "channel_id", callID, "error", err)
|
|
}
|
|
}
|
|
cancel()
|
|
}
|
|
if pump != nil {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
_ = pump.Stop(ctx)
|
|
cancel()
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (m *Manager) startTonePump(ctx context.Context, callID string, mediaClient MediaClient) {
|
|
go func() {
|
|
codec := m.mediaCodec()
|
|
sampleRate := m.mediaSampleRate()
|
|
payload := asteriskws.GenerateSineTone(codec, 440, 100*time.Millisecond, sampleRate, 0.08)
|
|
ticker := time.NewTicker(100 * time.Millisecond)
|
|
defer ticker.Stop()
|
|
for {
|
|
sendCtx, cancel := context.WithTimeout(ctx, time.Second)
|
|
err := mediaClient.SendAudio(sendCtx, payload)
|
|
cancel()
|
|
if err != nil {
|
|
m.log("live tone send failed", "channel_id", callID, "error", err)
|
|
return
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}()
|
|
m.log("live tone pump started", "channel_id", callID, "codec", string(m.mediaCodec()), "sample_rate", m.mediaSampleRate())
|
|
}
|
|
|
|
func (m *Manager) playDiagnostic(ctx context.Context, channelID string) {
|
|
player, ok := m.actions.(playbackActionClient)
|
|
if !ok {
|
|
m.log("diagnostic playback unavailable", "channel_id", channelID)
|
|
return
|
|
}
|
|
playCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
|
|
defer cancel()
|
|
if err := player.PlayChannel(playCtx, channelID, "sound:hello-world"); err != nil {
|
|
m.log("diagnostic playback failed", "channel_id", channelID, "error", err)
|
|
return
|
|
}
|
|
m.log("diagnostic playback started", "channel_id", channelID, "media", "sound:hello-world")
|
|
}
|
|
|
|
func (m *Manager) startMediaDiagnostics(ctx context.Context, callID string, mediaClient MediaClient) {
|
|
diag, ok := mediaClient.(diagnosticMediaClient)
|
|
if !ok {
|
|
return
|
|
}
|
|
go func() {
|
|
ticker := time.NewTicker(time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case ev, ok := <-diag.Events():
|
|
if !ok {
|
|
return
|
|
}
|
|
switch ev.Type {
|
|
case asteriskws.ControlStatus, asteriskws.ControlMediaXOFF, asteriskws.ControlMediaXON, asteriskws.ControlQueueDrained:
|
|
m.log("media websocket control", "channel_id", callID, "event", string(ev.Type), "raw", ev.Raw)
|
|
}
|
|
case <-ticker.C:
|
|
statusCtx, cancel := context.WithTimeout(ctx, time.Second)
|
|
_ = diag.GetStatus(statusCtx)
|
|
cancel()
|
|
st := diag.Stats()
|
|
m.log("media websocket stats", "channel_id", callID, "inbound_frames", st.InboundFrames, "inbound_bytes", st.InboundBytes, "outbound_frames", st.OutboundFrames, "outbound_bytes", st.OutboundBytes, "xoff", st.XOFFCount, "xon", st.XONCount)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (m *Manager) delayedHangup(channelID string) {
|
|
time.Sleep(m.cfg.TestCallHangupAfter)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer cancel()
|
|
if m.actions != nil {
|
|
_ = m.actions.HangupChannel(ctx, channelID)
|
|
}
|
|
}
|
|
|
|
func (m *Manager) mediaCodec() media.Codec {
|
|
if m.cfg.MediaCodec != "" {
|
|
return m.cfg.MediaCodec
|
|
}
|
|
return media.CodecSLIN16
|
|
}
|
|
|
|
func (m *Manager) mediaSampleRate() int {
|
|
switch m.mediaCodec() {
|
|
case media.CodecULaw, media.CodecALaw:
|
|
return 8000
|
|
default:
|
|
return 16000
|
|
}
|
|
}
|
|
|
|
func (m *Manager) voiceInputSampleRate() int {
|
|
if m.cfg.VoiceInputSampleRate > 0 {
|
|
return m.cfg.VoiceInputSampleRate
|
|
}
|
|
return 24000
|
|
}
|
|
|
|
func (m *Manager) voiceOutputSampleRate() int {
|
|
if m.cfg.VoiceOutputSampleRate > 0 {
|
|
return m.cfg.VoiceOutputSampleRate
|
|
}
|
|
return 24000
|
|
}
|
|
|
|
func (m *Manager) endSession(channelID, reason string) {
|
|
session, ok := m.store.Get(channelID)
|
|
if !ok {
|
|
return
|
|
}
|
|
now := time.Now().UTC()
|
|
session.EndedAt = &now
|
|
session.State = StateEnded
|
|
duration := now.Sub(session.StartedAt).String()
|
|
if session.MediaCancel != nil {
|
|
session.MediaCancel()
|
|
}
|
|
if m.cfg.VoiceProvider != nil {
|
|
_ = m.cfg.VoiceProvider.Close(context.Background())
|
|
}
|
|
if m.cfg.EndDialogue != nil {
|
|
_ = m.cfg.EndDialogue(context.Background(), channelID, reason)
|
|
}
|
|
m.store.Delete(channelID)
|
|
m.log("session ended", "channel_id", channelID, "route", session.Route, "answered", session.Answered, "duration", duration, "reason", reason)
|
|
}
|
|
|
|
func detectRoute(args []string) string {
|
|
if slices.Contains(args, "test") {
|
|
return "test"
|
|
}
|
|
if slices.Contains(args, "production") {
|
|
return "production"
|
|
}
|
|
return "unknown"
|
|
}
|
|
|
|
func (m *Manager) log(msg string, args ...any) {
|
|
if m.logger != nil {
|
|
m.logger.Info(msg, args...)
|
|
}
|
|
}
|
|
|
|
func isGeneratedMediaChannel(value string) bool {
|
|
return strings.Contains(value, "aiop-media-") || strings.Contains(value, "aiop-selftest-")
|
|
}
|