diff --git a/docs/evidence/saas-dispatcher-implementation.md b/docs/evidence/saas-dispatcher-implementation.md index bdc8c9a..ffe7739 100644 --- a/docs/evidence/saas-dispatcher-implementation.md +++ b/docs/evidence/saas-dispatcher-implementation.md @@ -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 或线路已验证,历史验收报告保留当时包名与覆盖率事实。 ## 验收台账 diff --git a/internal/callruntime/config_test.go b/internal/callruntime/config_test.go deleted file mode 100644 index 2988c68..0000000 --- a/internal/callruntime/config_test.go +++ /dev/null @@ -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") - } -} diff --git a/internal/callruntime/lifecycle_test.go b/internal/callruntime/lifecycle_test.go deleted file mode 100644 index 960932c..0000000 --- a/internal/callruntime/lifecycle_test.go +++ /dev/null @@ -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) -} diff --git a/internal/callruntime/recording_test.go b/internal/callruntime/recording_test.go deleted file mode 100644 index 3aa5e65..0000000 --- a/internal/callruntime/recording_test.go +++ /dev/null @@ -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") - } -} diff --git a/internal/callruntime/runtime.go b/internal/callruntime/runtime.go deleted file mode 100644 index 1cb97a6..0000000 --- a/internal/callruntime/runtime.go +++ /dev/null @@ -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 -}