go-sip/internal/callflow/flow.go

package callflow

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"git.ipao.vip/rogee/go-sip/internal/ai"
	"git.ipao.vip/rogee/go-sip/internal/media"
)

// MediaSession is the only transport boundary of the shared call flow.
// Real mode uses Asterisk ExternalMedia/RTP; mock mode uses MemorySession; the
// sequencing is shared while ASR-only deliberately omits opening/reply TTS.
type MediaSession interface {
	ReadPayload(context.Context) ([]byte, error)
	SendPCM16(context.Context, []byte, int) error
	Stats() media.RTPStats
}

type Result struct {
	Turn          ai.TurnResult
	Turns         []ai.TurnResult
	Inbound       []byte
	InboundTurns  [][]byte
	OutboundTurns [][]byte
	RTP           media.RTPStats
}

func Execute(ctx context.Context, session MediaSession, pipeline ai.Pipeline, snapshot ai.Snapshot, opening string, turnWindow time.Duration) (Result, error) {
	if turnWindow <= 0 {
		turnWindow = 5 * time.Second
	}
	return ExecuteWithCapture(ctx, session, pipeline, snapshot, opening, CaptureConfig{
		FirstSpeechTimeout: turnWindow,
		MaxDuration:        turnWindow,
		MaxTurns:           1,
	})
}

func ExecuteWithCapture(ctx context.Context, session MediaSession, pipeline ai.Pipeline, snapshot ai.Snapshot, opening string, capture CaptureConfig) (Result, error) {
	if session == nil || pipeline == nil {
		return Result{}, errors.New("media session and AI pipeline are required")
	}
	if snapshot.Mode != ai.ModeFullAI && snapshot.Mode != ai.ModeASROnly {
		return Result{}, fmt.Errorf("unsupported callflow AI mode %q", snapshot.Mode)
	}
	maxTurns := capture.MaxTurns
	if maxTurns <= 0 {
		maxTurns = 1
	}
	result := Result{}
	if snapshot.Mode == ai.ModeFullAI {
		openingPCM, err := pipeline.Synthesize(ctx, snapshot, opening)
		if err != nil {
			return Result{}, err
		}
		result.OutboundTurns = append(result.OutboundTurns, clonePCM(openingPCM))
		if err := session.SendPCM16(ctx, openingPCM, 16000); err != nil {
			return result, err
		}
	}
	for turnIndex := 0; turnIndex < maxTurns; turnIndex++ {
		inbound, err := captureTurn(ctx, session, capture)
		if err != nil {
			return result, fmt.Errorf("capturing RTP turn %d after opening/reply prompt: %w", turnIndex+1, err)
		}
		result.Inbound = inbound
		result.InboundTurns = append(result.InboundTurns, clonePCM(inbound))
		if len(inbound) < 3200 {
			return result, fmt.Errorf("captured audio turn %d is too short", turnIndex+1)
		}
		turn, err := pipeline.RunTurn(ctx, snapshot, inbound)
		if err != nil {
			return result, fmt.Errorf("run AI turn %d: %w", turnIndex+1, err)
		}
		turn.InvalidCall, turn.InvalidReason = invalidCallReason(turn.Transcript, turn.Reply)
		result.Turn = turn
		result.Turns = append(result.Turns, turn)
		if turn.InvalidCall {
			result.RTP = session.Stats()
			return result, nil
		}
		if snapshot.Mode == ai.ModeASROnly {
			result.RTP = session.Stats()
			continue
		}
		if err := session.SendPCM16(ctx, turn.AudioPCM16, 16000); err != nil {
			return result, fmt.Errorf("send AI reply turn %d: %w", turnIndex+1, err)
		}
		result.OutboundTurns = append(result.OutboundTurns, clonePCM(turn.AudioPCM16))
		result.RTP = session.Stats()
	}
	result.RTP = session.Stats()
	return result, nil
}

func clonePCM(pcm []byte) []byte {
	return append([]byte(nil), pcm...)
}

func invalidCallReason(transcript, reply string) (bool, string) {
	if strings.Contains(reply, ai.InvalidCallMarker) {
		return true, "llm_invalid_call_marker"
	}
	normalized := strings.NewReplacer(" ", "", "　", "", "。", "", "，", "", ",", "", ".", "").Replace(strings.TrimSpace(transcript))
	for _, marker := range []string{"打错", "不需要", "不用", "没兴趣", "不考虑", "不方便", "别打", "拒绝", "骚扰", "语音信箱", "自动语音", "请按键", "空号"} {
		if strings.Contains(normalized, marker) {
			return true, "transcript_invalid_intent"
		}
	}
	return false, ""
}

// MemorySession is a bounded transport adapter for mock/mixed-flow tests. It
// has no SIP semantics and never bypasses the shared call flow.
type MemorySession struct {
	inbound  [][]byte
	index    int
	outbound []byte
	stats    media.RTPStats
}

func NewMemorySession(pcm []byte) *MemorySession {
	const frameBytes = 640
	frames := make([][]byte, 0, (len(pcm)+frameBytes-1)/frameBytes)
	for offset := 0; offset < len(pcm); offset += frameBytes {
		end := offset + frameBytes
		if end > len(pcm) {
			end = len(pcm)
		}
		frames = append(frames, append([]byte(nil), pcm[offset:end]...))
	}
	return &MemorySession{inbound: frames}
}

func (m *MemorySession) ReadPayload(ctx context.Context) ([]byte, error) {
	if m.index < len(m.inbound) {
		payload := m.inbound[m.index]
		m.index++
		m.stats.ReceivedPackets++
		m.stats.ReceivedBytes += uint64(len(payload))
		return append([]byte(nil), payload...), nil
	}
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	default:
		return nil, context.DeadlineExceeded
	}
}

func (m *MemorySession) SendPCM16(ctx context.Context, pcm []byte, _ int) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	m.outbound = append(m.outbound, pcm...)
	m.stats.SentBytes += uint64(len(pcm))
	m.stats.SentPackets += uint64((len(pcm) + 639) / 640)
	return nil
}

func (m *MemorySession) Stats() media.RTPStats { return m.stats }
func (m *MemorySession) OutboundPCM() []byte   { return append([]byte(nil), m.outbound...) }