refactor(agent): remove unreachable legacy ARI call runtime
This commit is contained in:
@@ -123,6 +123,7 @@
|
||||
- Agent 旧 spool 代码：旧执行状态、实时事件、上传尝试、失败事实及上传锁只由旧模块彼此调用，没有现行命令或录音恢复调用；删除专属实现与测试。录音恢复仍复用的原子写、目录同步、文件名校验等小函数移至 `internal/agent/file_state.go`，通过现行录音重试/未知结果测试确认恢复能力保持。删除源码不删除任何磁盘 spool；旧文件在当前恢复根目录触发上述拒绝启动，不能视作已经补传或处置。
|
||||
- Agent Proto 旧消息：先用描述符结构测试复现 `ConfigReference` 等废弃定义仍可被发现，再按当前八个服务方法及现行 Go 引用追踪字段依赖；删除无调用者的 26 个旧消息和 3 个旧枚举，不改现行请求/响应的字段号。使用本地 Buf 重新生成 Go 类型，核对七项来源/hash；现行录音客户端、获批执行、控制及会话测试保持通过。固定 `agent.v1` 仍是有效的内部协议值；这不构成真实 Agent/Asterisk 或外部 SaaS 验收。
|
||||
- 隔离 MQ 校验脚本：审计发现脚本因测试文件改名仍使用旧筛选式，`internal/mq` 和 `internal/dispatcher` 输出 `no tests to run` 但退出成功；先复现两个空匹配，再改为现行测试名并要求三项均明确 `PASS`，跳过或空匹配均失败。隔离 RabbitMQ 实跑 `TestBrokerSharedResultQueueAndNoConfigure`、`TestRuntimeIsolatedControlBacklogExecuteAndSharedResult` 与 `TestCurrentDispatcherCommandStartsWithIsolatedMQHTTPAndAgent` 全部通过；这只证明本机隔离链路，不代替真实 MQ/SaaS 应用收讫。
|
||||
- 旧 ARI/媒体执行包：`internal/callruntime` 仅实现旧 `ai.Snapshot` 的独立 `Run` 入口，无当前 Agent/Dispatcher 调用者；删除该包及仅针对该路径的测试。现行获批执行、录音恢复和完整 AI Mock 测试继续覆盖唯一现行入口；删除死代码不表示真实 Asterisk/ARI 或线路已验证，历史验收报告保留当时包名与覆盖率事实。
|
||||
|
||||
## 验收台账
|
||||
|
||||
|
||||
@@ -1,67 +0,0 @@
|
||||
package callruntime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.ipao.vip/rogee/go-sip/internal/ai"
|
||||
)
|
||||
|
||||
func validConfig() Config {
|
||||
return Config{ARIURL: "http://127.0.0.1:1/ari", ARIWebsocketURL: "ws://127.0.0.1:1/ari/events", ARIApplication: "test", ARIUsername: "local", ARIPassword: "isolated-test-only", Endpoint: "PJSIP/local-test", MediaBind: "127.0.0.1", MediaPort: 20000, MediaFormat: "slin16", MediaSampleRate: 16000, PayloadType: 96, Pipeline: ai.MockPipeline{}}
|
||||
}
|
||||
|
||||
func TestRuntimeRejectsInvalidConfigurationBeforeNetwork(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
change func(*Config)
|
||||
}{
|
||||
{"ARI URL", func(c *Config) { c.ARIURL = "" }},
|
||||
{"websocket", func(c *Config) { c.ARIWebsocketURL = "" }},
|
||||
{"application", func(c *Config) { c.ARIApplication = "" }},
|
||||
{"username", func(c *Config) { c.ARIUsername = "" }},
|
||||
{"password", func(c *Config) { c.ARIPassword = "" }},
|
||||
{"endpoint", func(c *Config) { c.Endpoint = "" }},
|
||||
{"media bind", func(c *Config) { c.MediaBind = "" }},
|
||||
{"low media port", func(c *Config) { c.MediaPort = 1023 }},
|
||||
{"high media port", func(c *Config) { c.MediaPort = 65536 }},
|
||||
{"media format", func(c *Config) { c.MediaFormat = "invalid" }},
|
||||
{"pipeline", func(c *Config) { c.Pipeline = nil }},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
cfg := validConfig()
|
||||
tc.change(&cfg)
|
||||
if err := cfg.validate(); err == nil {
|
||||
t.Fatal("invalid configuration accepted")
|
||||
}
|
||||
if _, err := Run(context.Background(), cfg); err == nil {
|
||||
t.Fatal("invalid configuration reached runtime")
|
||||
}
|
||||
})
|
||||
}
|
||||
if err := validConfig().validate(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRuntimeNormalizationPreservesExplicitSettings(t *testing.T) {
|
||||
cfg := validConfig()
|
||||
cfg.AnswerTimeout = time.Second
|
||||
cfg.TurnWindow = 2 * time.Second
|
||||
cfg.FirstSpeechTimeout = 3 * time.Second
|
||||
cfg.MaxTurnDuration = 4 * time.Second
|
||||
cfg.EndSilence = 5 * time.Second
|
||||
cfg.MaxCallDuration = 6 * time.Second
|
||||
cfg.VoiceThreshold = 22
|
||||
cfg.MaxTurns = 9
|
||||
cfg.OpeningPrompt = "configured opening"
|
||||
actual := cfg.normalized()
|
||||
if actual.AnswerTimeout != cfg.AnswerTimeout || actual.TurnWindow != cfg.TurnWindow || actual.FirstSpeechTimeout != cfg.FirstSpeechTimeout || actual.MaxTurnDuration != cfg.MaxTurnDuration || actual.EndSilence != cfg.EndSilence || actual.MaxCallDuration != cfg.MaxCallDuration || actual.VoiceThreshold != cfg.VoiceThreshold || actual.MaxTurns != cfg.MaxTurns || actual.OpeningPrompt != cfg.OpeningPrompt {
|
||||
t.Fatal("explicit conversation settings were overwritten")
|
||||
}
|
||||
defaults := (Config{}).normalized()
|
||||
if defaults.MediaFormat != "slin16" || defaults.MediaSampleRate != 16000 || defaults.AnswerTimeout <= 0 || defaults.TurnWindow <= 0 {
|
||||
t.Fatal("invalid normalized transport settings")
|
||||
}
|
||||
}
|
||||
@@ -1,85 +0,0 @@
|
||||
package callruntime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/CyCoreSystems/ari/v5"
|
||||
)
|
||||
|
||||
type eventSubscription struct{ events chan ari.Event }
|
||||
|
||||
func (s eventSubscription) Events() <-chan ari.Event { return s.events }
|
||||
func (s eventSubscription) Cancel() {}
|
||||
func events(values ...ari.Event) eventSubscription {
|
||||
s := eventSubscription{make(chan ari.Event, len(values))}
|
||||
for _, event := range values {
|
||||
s.events <- event
|
||||
}
|
||||
close(s.events)
|
||||
return s
|
||||
}
|
||||
|
||||
func TestWaitForAnswerUsesOnlyTargetChannel(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
event ari.Event
|
||||
wantError string
|
||||
}{
|
||||
{name: "stasis", event: &ari.StasisStart{EventData: ari.EventData{Type: "StasisStart"}, Channel: ari.ChannelData{ID: "call"}}},
|
||||
{name: "up", event: &ari.ChannelStateChange{EventData: ari.EventData{Type: "ChannelStateChange"}, Channel: ari.ChannelData{ID: "call", State: "Up"}}},
|
||||
{name: "hangup", event: &ari.ChannelHangupRequest{EventData: ari.EventData{Type: "ChannelHangupRequest"}, Channel: ari.ChannelData{ID: "call", State: "Down"}, Cause: 17}, wantError: "cause=17"},
|
||||
{name: "destroyed", event: &ari.ChannelDestroyed{EventData: ari.EventData{Type: "ChannelDestroyed"}, Channel: ari.ChannelData{ID: "call"}}, wantError: "ended before StasisStart"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
other := &ari.StasisStart{EventData: ari.EventData{Type: "StasisStart"}, Channel: ari.ChannelData{ID: "another-call"}}
|
||||
err := waitForStasisStart(context.Background(), events(nil, other, tc.event), "call", time.Second)
|
||||
if tc.wantError == "" && err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tc.wantError != "" && (err == nil || !strings.Contains(err.Error(), tc.wantError)) {
|
||||
t.Fatalf("unexpected answer result: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
if err := waitForStasisStart(context.Background(), events(), "call", time.Second); err == nil || !strings.Contains(err.Error(), "subscription closed") {
|
||||
t.Fatal("closed subscription accepted")
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
if err := waitForStasisStart(ctx, eventSubscription{make(chan ari.Event)}, "call", time.Second); !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("cancellation lost: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLifecycleCancellationIsScopedToChannel(t *testing.T) {
|
||||
for _, kind := range []string{"ChannelHangupRequest", "ChannelDestroyed"} {
|
||||
t.Run(kind, func(t *testing.T) {
|
||||
var event ari.Event
|
||||
if kind == "ChannelDestroyed" {
|
||||
event = &ari.ChannelDestroyed{EventData: ari.EventData{Type: kind}, Channel: ari.ChannelData{ID: "call"}}
|
||||
} else {
|
||||
event = &ari.ChannelHangupRequest{EventData: ari.EventData{Type: kind}, Channel: ari.ChannelData{ID: "call"}}
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
other := &ari.ChannelDestroyed{EventData: ari.EventData{Type: "ChannelDestroyed"}, Channel: ari.ChannelData{ID: "other"}}
|
||||
watchChannelLifecycle(ctx, events(other, event), "call", cancel)
|
||||
if !errors.Is(ctx.Err(), context.Canceled) {
|
||||
t.Fatal("matching hangup failed to cancel")
|
||||
}
|
||||
})
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
watchChannelLifecycle(ctx, events(&ari.ChannelDestroyed{EventData: ari.EventData{Type: "ChannelDestroyed"}, Channel: ari.ChannelData{ID: "other"}}), "call", cancel)
|
||||
if ctx.Err() != nil {
|
||||
t.Fatal("another call ended this call")
|
||||
}
|
||||
watchChannelLifecycle(ctx, events(nil), "call", cancel)
|
||||
cancel()
|
||||
watchChannelLifecycle(ctx, eventSubscription{make(chan ari.Event)}, "call", cancel)
|
||||
}
|
||||
@@ -1,86 +0,0 @@
|
||||
package callruntime
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/sha256"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.ipao.vip/rogee/go-sip/internal/callflow"
|
||||
)
|
||||
|
||||
func TestConversationRecordingFactsMatchFiles(t *testing.T) {
|
||||
directory := t.TempDir()
|
||||
incoming := [][]byte{{1, 0, 2, 0}, make([]byte, 32000)}
|
||||
outgoing := [][]byte{{4, 0}, {5, 0, 6, 0}}
|
||||
in, out, err := persistConversationRecordings(directory, "channel/with:separators", callflow.Result{InboundTurns: incoming, OutboundTurns: outgoing})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(in) != 2 || len(out) != 2 {
|
||||
t.Fatal("lost recording segments")
|
||||
}
|
||||
for side, facts := range [][]RecordingFact{in, out} {
|
||||
expected := [][][]byte{incoming, outgoing}[side]
|
||||
for index, fact := range facts {
|
||||
if filepath.Dir(fact.Path) != directory {
|
||||
t.Fatal("recording escaped directory")
|
||||
}
|
||||
raw, err := os.ReadFile(fact.Path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sum := sha256.Sum256(raw)
|
||||
if fact.SHA256 != hex.EncodeToString(sum[:]) || fact.Bytes != len(raw) || fact.DurationMS != int64(len(expected[index]))*1000/32000 {
|
||||
t.Fatal("recording fact does not match content")
|
||||
}
|
||||
if string(raw[:4]) != "RIFF" || string(raw[8:12]) != "WAVE" || string(raw[36:40]) != "data" {
|
||||
t.Fatal("invalid WAV header")
|
||||
}
|
||||
if binary.LittleEndian.Uint32(raw[24:28]) != 16000 || binary.LittleEndian.Uint32(raw[40:44]) != uint32(len(expected[index])) || !bytes.Equal(raw[44:], expected[index]) {
|
||||
t.Fatal("incorrect WAV format or payload")
|
||||
}
|
||||
info, err := os.Stat(fact.Path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if info.Mode().Perm() != 0600 {
|
||||
t.Fatal("recording is not restricted")
|
||||
}
|
||||
}
|
||||
}
|
||||
if in[0].Segment != "inbound_turn_01" || out[0].Segment != "outbound_segment_00" {
|
||||
t.Fatal("segment identities changed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordingFailuresAreExplicit(t *testing.T) {
|
||||
directory := t.TempDir()
|
||||
if err := writeWAV(filepath.Join(directory, "odd.wav"), []byte{1}, 16000); err == nil {
|
||||
t.Fatal("odd PCM accepted")
|
||||
}
|
||||
blocker := filepath.Join(directory, "file")
|
||||
if err := os.WriteFile(blocker, []byte("blocked"), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, _, err := persistConversationRecordings(filepath.Join(blocker, "child"), "channel", callflow.Result{InboundTurns: [][]byte{{0, 0}}}); err == nil {
|
||||
t.Fatal("directory creation error hidden")
|
||||
}
|
||||
for _, flow := range []callflow.Result{{InboundTurns: [][]byte{{1}}}, {OutboundTurns: [][]byte{{1}}}} {
|
||||
if _, _, err := persistConversationRecordings(directory, "channel", flow); err == nil {
|
||||
t.Fatal("invalid segment accepted")
|
||||
}
|
||||
}
|
||||
for _, dir := range []string{"", directory} {
|
||||
in, out, err := persistConversationRecordings(dir, "empty", callflow.Result{})
|
||||
if err != nil || len(in) != 0 || len(out) != 0 {
|
||||
t.Fatal("empty recording produced facts")
|
||||
}
|
||||
}
|
||||
if _, _, err := fileDigestAndSize(filepath.Join(directory, "missing.wav")); err == nil {
|
||||
t.Fatal("missing recording silently returned a checksum")
|
||||
}
|
||||
}
|
||||
@@ -1,442 +0,0 @@
|
||||
package callruntime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.ipao.vip/rogee/go-sip/internal/ai"
|
||||
"git.ipao.vip/rogee/go-sip/internal/callflow"
|
||||
"git.ipao.vip/rogee/go-sip/internal/media"
|
||||
"github.com/CyCoreSystems/ari/v5"
|
||||
"github.com/CyCoreSystems/ari/v5/client/native"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultAnswerTimeout = 45 * time.Second
|
||||
defaultTurnWindow = 5 * time.Second
|
||||
defaultFirstSpeechWait = 20 * time.Second
|
||||
defaultMaxTurnDuration = 12 * time.Second
|
||||
defaultEndSilence = 900 * time.Millisecond
|
||||
defaultVoiceThreshold = 350
|
||||
defaultConversationTurns = 3
|
||||
defaultCallDuration = 2 * time.Minute
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
ARIURL string
|
||||
ARIWebsocketURL string
|
||||
ARIApplication string
|
||||
ARIUsername string
|
||||
ARIPassword string
|
||||
|
||||
Endpoint string
|
||||
CallerID string
|
||||
MediaBind string
|
||||
MediaPort int
|
||||
MediaFormat string
|
||||
MediaSampleRate int
|
||||
PayloadType uint8
|
||||
|
||||
RecordingDirectory string
|
||||
AnswerTimeout time.Duration
|
||||
TurnWindow time.Duration
|
||||
FirstSpeechTimeout time.Duration
|
||||
MaxTurnDuration time.Duration
|
||||
EndSilence time.Duration
|
||||
VoiceThreshold int
|
||||
MaxTurns int
|
||||
MaxCallDuration time.Duration
|
||||
OpeningPrompt string
|
||||
|
||||
Snapshot ai.Snapshot
|
||||
Pipeline ai.Pipeline
|
||||
}
|
||||
|
||||
type RecordingFact struct {
|
||||
Segment string
|
||||
Path string
|
||||
SHA256 string
|
||||
Bytes int // Complete file size, including the WAV header.
|
||||
DurationMS int64
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
ChannelID string
|
||||
Transcript string
|
||||
Reply string
|
||||
InvalidCall bool
|
||||
InvalidReason string
|
||||
Turns []ai.TurnResult
|
||||
InboundPath string
|
||||
InboundSHA256 string
|
||||
InboundBytes int
|
||||
OutboundPath string
|
||||
OutboundSHA256 string
|
||||
OutboundBytes int
|
||||
InboundRecordings []RecordingFact
|
||||
OutboundRecordings []RecordingFact
|
||||
RTP media.RTPStats
|
||||
}
|
||||
|
||||
func (c Config) validate() error {
|
||||
if c.ARIApplication == "" || c.ARIURL == "" || c.ARIWebsocketURL == "" {
|
||||
return errors.New("ARI URL, websocket URL and application are required")
|
||||
}
|
||||
if c.ARIPassword == "" || c.ARIUsername == "" {
|
||||
return errors.New("ARI credentials are required")
|
||||
}
|
||||
if c.Endpoint == "" {
|
||||
return errors.New("endpoint is required")
|
||||
}
|
||||
if c.MediaBind == "" || c.MediaPort < 1024 || c.MediaPort > 65535 {
|
||||
return errors.New("valid media bind and port are required")
|
||||
}
|
||||
if err := media.ValidateFormat(media.Format(c.MediaFormat), c.PayloadType, c.MediaSampleRate); err != nil {
|
||||
return err
|
||||
}
|
||||
if c.Pipeline == nil {
|
||||
return errors.New("AI pipeline is required")
|
||||
}
|
||||
if c.AnswerTimeout <= 0 {
|
||||
c.AnswerTimeout = defaultAnswerTimeout
|
||||
}
|
||||
if c.TurnWindow <= 0 {
|
||||
c.TurnWindow = defaultTurnWindow
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c Config) normalized() Config {
|
||||
if c.AnswerTimeout <= 0 {
|
||||
c.AnswerTimeout = defaultAnswerTimeout
|
||||
}
|
||||
if c.TurnWindow <= 0 {
|
||||
c.TurnWindow = defaultTurnWindow
|
||||
}
|
||||
if c.MediaFormat == "" {
|
||||
c.MediaFormat = string(media.FormatSLIN16)
|
||||
}
|
||||
if c.MediaSampleRate <= 0 {
|
||||
c.MediaSampleRate = 16000
|
||||
}
|
||||
if c.FirstSpeechTimeout <= 0 {
|
||||
c.FirstSpeechTimeout = defaultFirstSpeechWait
|
||||
}
|
||||
if c.MaxTurnDuration <= 0 {
|
||||
c.MaxTurnDuration = defaultMaxTurnDuration
|
||||
}
|
||||
if c.EndSilence <= 0 {
|
||||
c.EndSilence = defaultEndSilence
|
||||
}
|
||||
if c.VoiceThreshold <= 0 {
|
||||
c.VoiceThreshold = defaultVoiceThreshold
|
||||
}
|
||||
if c.MaxTurns <= 0 {
|
||||
c.MaxTurns = defaultConversationTurns
|
||||
}
|
||||
if c.MaxCallDuration <= 0 {
|
||||
c.MaxCallDuration = defaultCallDuration
|
||||
}
|
||||
if c.OpeningPrompt == "" {
|
||||
c.OpeningPrompt = "您好，请说出您想咨询的内容。"
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func Run(ctx context.Context, cfg Config) (Result, error) {
|
||||
cfg = cfg.normalized()
|
||||
if err := cfg.validate(); err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
runCtx, cancelRun := context.WithTimeout(ctx, cfg.MaxCallDuration)
|
||||
defer cancelRun()
|
||||
stream, err := media.ListenRTPWithFormat(fmt.Sprintf("%s:%d", cfg.MediaBind, cfg.MediaPort), cfg.PayloadType, media.Format(cfg.MediaFormat), cfg.MediaSampleRate)
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("bind external media: %w", err)
|
||||
}
|
||||
defer stream.Close()
|
||||
|
||||
client, err := native.Connect(&native.Options{
|
||||
URL: cfg.ARIURL,
|
||||
WebsocketURL: cfg.ARIWebsocketURL,
|
||||
Application: cfg.ARIApplication,
|
||||
Username: cfg.ARIUsername,
|
||||
Password: cfg.ARIPassword,
|
||||
SubscribeAll: true,
|
||||
})
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("connect ARI: %w", err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
starts := client.Bus().Subscribe(nil, ari.Events.All)
|
||||
defer starts.Cancel()
|
||||
originate, err := client.Channel().Originate(nil, ari.OriginateRequest{
|
||||
Endpoint: cfg.Endpoint,
|
||||
// ari.OriginateRequest.Timeout is specified in seconds.
|
||||
Timeout: int(cfg.AnswerTimeout / time.Second),
|
||||
CallerID: cfg.CallerID,
|
||||
App: cfg.ARIApplication,
|
||||
Formats: cfg.MediaFormat,
|
||||
})
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("originate %s: %w", cfg.Endpoint, err)
|
||||
}
|
||||
channelID := originate.ID()
|
||||
channelKey := ari.NewKey(ari.ChannelKey, channelID)
|
||||
var bridgeKey *ari.Key
|
||||
var externalKey *ari.Key
|
||||
defer func() {
|
||||
if externalKey != nil {
|
||||
_ = client.Channel().Hangup(externalKey, "normal")
|
||||
}
|
||||
if channelKey != nil {
|
||||
_ = client.Channel().Hangup(channelKey, "normal")
|
||||
}
|
||||
if bridgeKey != nil {
|
||||
_ = client.Bridge().Delete(bridgeKey)
|
||||
}
|
||||
}()
|
||||
|
||||
if err := waitForStasisStart(runCtx, starts, channelID, cfg.AnswerTimeout); err != nil {
|
||||
return Result{}, err
|
||||
}
|
||||
callCtx, cancelCall := context.WithCancel(runCtx)
|
||||
defer cancelCall()
|
||||
go watchChannelLifecycle(callCtx, starts, channelID, cancelCall)
|
||||
if err := client.Channel().Answer(channelKey); err != nil && !strings.Contains(strings.ToLower(err.Error()), "already") {
|
||||
return Result{}, fmt.Errorf("answer channel: %w", err)
|
||||
}
|
||||
|
||||
bridgeID := "agent-call-" + channelID
|
||||
bridge, err := client.Bridge().Create(ari.NewKey(ari.BridgeKey, bridgeID), "mixing", bridgeID)
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("create media bridge: %w", err)
|
||||
}
|
||||
bridgeKey = ari.NewKey(ari.BridgeKey, bridge.ID())
|
||||
if err := client.Bridge().AddChannel(bridgeKey, channelID); err != nil {
|
||||
return Result{}, fmt.Errorf("add call channel to bridge: %w", err)
|
||||
}
|
||||
external, err := client.Channel().ExternalMedia(nil, ari.ExternalMediaOptions{
|
||||
App: cfg.ARIApplication,
|
||||
ExternalHost: fmt.Sprintf("%s:%d", cfg.MediaBind, cfg.MediaPort),
|
||||
Encapsulation: "rtp",
|
||||
Transport: "udp",
|
||||
ConnectionType: "client",
|
||||
Format: cfg.MediaFormat,
|
||||
Direction: "both",
|
||||
})
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("create external media channel: %w", err)
|
||||
}
|
||||
externalKey = ari.NewKey(ari.ChannelKey, external.ID())
|
||||
address, err := external.GetVariable("UNICASTRTP_LOCAL_ADDRESS")
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("read ExternalMedia RTP address: %w", err)
|
||||
}
|
||||
port, err := external.GetVariable("UNICASTRTP_LOCAL_PORT")
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("read ExternalMedia RTP port: %w", err)
|
||||
}
|
||||
if err := stream.SetPeer(net.JoinHostPort(strings.TrimSpace(address), strings.TrimSpace(port))); err != nil {
|
||||
return Result{}, fmt.Errorf("configure ExternalMedia RTP peer: %w", err)
|
||||
}
|
||||
if err := client.Bridge().AddChannel(bridgeKey, external.ID()); err != nil {
|
||||
return Result{}, fmt.Errorf("add external media to bridge: %w", err)
|
||||
}
|
||||
|
||||
flowResult, flowErr := callflow.ExecuteWithCapture(callCtx, stream, cfg.Pipeline, cfg.Snapshot, cfg.OpeningPrompt, callflow.CaptureConfig{
|
||||
FirstSpeechTimeout: cfg.FirstSpeechTimeout,
|
||||
MaxDuration: cfg.MaxTurnDuration,
|
||||
EndSilence: cfg.EndSilence,
|
||||
VoiceThreshold: cfg.VoiceThreshold,
|
||||
MaxTurns: cfg.MaxTurns,
|
||||
})
|
||||
inboundRecordings, outboundRecordings, recordingErr := persistConversationRecordings(cfg.RecordingDirectory, channelID, flowResult)
|
||||
if recordingErr != nil {
|
||||
return Result{}, recordingErr
|
||||
}
|
||||
turn := flowResult.Turn
|
||||
result := Result{
|
||||
ChannelID: channelID,
|
||||
Transcript: turn.Transcript,
|
||||
Reply: turn.Reply,
|
||||
InvalidCall: turn.InvalidCall,
|
||||
InvalidReason: turn.InvalidReason,
|
||||
Turns: append([]ai.TurnResult(nil), flowResult.Turns...),
|
||||
InboundRecordings: inboundRecordings,
|
||||
OutboundRecordings: outboundRecordings,
|
||||
RTP: flowResult.RTP,
|
||||
}
|
||||
if len(inboundRecordings) > 0 {
|
||||
last := inboundRecordings[len(inboundRecordings)-1]
|
||||
result.InboundPath, result.InboundBytes, result.InboundSHA256 = last.Path, last.Bytes, last.SHA256
|
||||
}
|
||||
if len(outboundRecordings) > 0 {
|
||||
last := outboundRecordings[len(outboundRecordings)-1]
|
||||
result.OutboundPath, result.OutboundBytes, result.OutboundSHA256 = last.Path, last.Bytes, last.SHA256
|
||||
}
|
||||
if flowErr != nil {
|
||||
stats := stream.Stats()
|
||||
return result, fmt.Errorf("execute call flow: %w (rtp rx_packets=%d rx_bytes=%d tx_packets=%d tx_bytes=%d)", flowErr, stats.ReceivedPackets, stats.ReceivedBytes, stats.SentPackets, stats.SentBytes)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func watchChannelLifecycle(ctx context.Context, sub ari.Subscription, channelID string, cancel context.CancelFunc) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case event, ok := <-sub.Events():
|
||||
if !ok || event == nil {
|
||||
return
|
||||
}
|
||||
matched := false
|
||||
for _, key := range event.Keys() {
|
||||
if key != nil && key.ID == channelID {
|
||||
matched = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !matched {
|
||||
continue
|
||||
}
|
||||
switch event.GetType() {
|
||||
case "ChannelHangupRequest", "ChannelDestroyed":
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func waitForStasisStart(ctx context.Context, sub ari.Subscription, channelID string, timeout time.Duration) error {
|
||||
waitCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
var recent []string
|
||||
for {
|
||||
select {
|
||||
case <-waitCtx.Done():
|
||||
return fmt.Errorf("waiting for channel %s answer/StasisStart (recent_events=%s): %w", channelID, strings.Join(recent, ","), waitCtx.Err())
|
||||
case event, ok := <-sub.Events():
|
||||
if !ok {
|
||||
return errors.New("ARI event subscription closed")
|
||||
}
|
||||
if event == nil {
|
||||
continue
|
||||
}
|
||||
typ := event.GetType()
|
||||
recent = append(recent, typ)
|
||||
if len(recent) > 8 {
|
||||
recent = recent[len(recent)-8:]
|
||||
}
|
||||
matched := false
|
||||
for _, key := range event.Keys() {
|
||||
if key != nil && key.ID == channelID {
|
||||
matched = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !matched {
|
||||
continue
|
||||
}
|
||||
if typ == "StasisStart" {
|
||||
return nil
|
||||
}
|
||||
if typ == "ChannelStateChange" {
|
||||
if state, ok := event.(*ari.ChannelStateChange); ok && strings.EqualFold(state.Channel.State, "Up") {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
if typ == "ChannelHangupRequest" {
|
||||
if hangup, ok := event.(*ari.ChannelHangupRequest); ok {
|
||||
return fmt.Errorf("channel %s ended before StasisStart: %s cause=%d soft=%t state=%s", channelID, typ, hangup.Cause, hangup.Soft, hangup.Channel.State)
|
||||
}
|
||||
return fmt.Errorf("channel %s ended before StasisStart: %s", channelID, typ)
|
||||
}
|
||||
if typ == "ChannelDestroyed" {
|
||||
return fmt.Errorf("channel %s ended before StasisStart: %s", channelID, typ)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func persistConversationRecordings(directory, channelID string, flowResult callflow.Result) ([]RecordingFact, []RecordingFact, error) {
|
||||
if directory == "" || (len(flowResult.InboundTurns) == 0 && len(flowResult.OutboundTurns) == 0) {
|
||||
return nil, nil, nil
|
||||
}
|
||||
if err := os.MkdirAll(directory, 0o700); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
stamp := time.Now().UTC().Format("20060102T150405.000000000Z")
|
||||
name := strings.NewReplacer("/", "_", "\\", "_", ":", "_").Replace(channelID)
|
||||
inbound := make([]RecordingFact, 0, len(flowResult.InboundTurns))
|
||||
for index, pcm := range flowResult.InboundTurns {
|
||||
path := filepath.Join(directory, fmt.Sprintf("%s-%s-inbound-turn-%02d.wav", stamp, name, index+1))
|
||||
if err := writeWAV(path, pcm, 16000); err != nil {
|
||||
return inbound, nil, err
|
||||
}
|
||||
digest, size, err := fileDigestAndSize(path)
|
||||
if err != nil {
|
||||
return inbound, nil, fmt.Errorf("hash inbound recording: %w", err)
|
||||
}
|
||||
inbound = append(inbound, RecordingFact{Segment: fmt.Sprintf("inbound_turn_%02d", index+1), Path: path, SHA256: digest, Bytes: size, DurationMS: int64(len(pcm)) * 1000 / 32000})
|
||||
}
|
||||
outbound := make([]RecordingFact, 0, len(flowResult.OutboundTurns))
|
||||
for index, pcm := range flowResult.OutboundTurns {
|
||||
path := filepath.Join(directory, fmt.Sprintf("%s-%s-outbound-segment-%02d.wav", stamp, name, index))
|
||||
if err := writeWAV(path, pcm, 16000); err != nil {
|
||||
return inbound, outbound, err
|
||||
}
|
||||
digest, size, err := fileDigestAndSize(path)
|
||||
if err != nil {
|
||||
return inbound, outbound, fmt.Errorf("hash outbound recording: %w", err)
|
||||
}
|
||||
outbound = append(outbound, RecordingFact{Segment: fmt.Sprintf("outbound_segment_%02d", index), Path: path, SHA256: digest, Bytes: size, DurationMS: int64(len(pcm)) * 1000 / 32000})
|
||||
}
|
||||
return inbound, outbound, nil
|
||||
}
|
||||
|
||||
func writeWAV(path string, pcm []byte, sampleRate int) error {
|
||||
if len(pcm)%2 != 0 {
|
||||
return errors.New("PCM16 has odd byte length")
|
||||
}
|
||||
dataSize := uint32(len(pcm))
|
||||
byteRate := uint32(sampleRate * 2)
|
||||
blockAlign := uint16(2)
|
||||
buf := make([]byte, 44+len(pcm))
|
||||
copy(buf[:4], "RIFF")
|
||||
binary.LittleEndian.PutUint32(buf[4:8], 36+dataSize)
|
||||
copy(buf[8:12], "WAVE")
|
||||
copy(buf[12:16], "fmt ")
|
||||
binary.LittleEndian.PutUint32(buf[16:20], 16)
|
||||
binary.LittleEndian.PutUint16(buf[20:22], 1)
|
||||
binary.LittleEndian.PutUint16(buf[22:24], 1)
|
||||
binary.LittleEndian.PutUint32(buf[24:28], uint32(sampleRate))
|
||||
binary.LittleEndian.PutUint32(buf[28:32], byteRate)
|
||||
binary.LittleEndian.PutUint16(buf[32:34], blockAlign)
|
||||
binary.LittleEndian.PutUint16(buf[34:36], 16)
|
||||
copy(buf[36:40], "data")
|
||||
binary.LittleEndian.PutUint32(buf[40:44], dataSize)
|
||||
copy(buf[44:], pcm)
|
||||
return os.WriteFile(path, buf, 0o600)
|
||||
}
|
||||
|
||||
func fileDigestAndSize(path string) (string, int, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
sum := sha256.Sum256(data)
|
||||
return hex.EncodeToString(sum[:]), len(data), nil
|
||||
}
|
||||
Reference in New Issue
Block a user