go-sip/internal/callflow/flow_test.go

package callflow

import (
	"context"
	"testing"
	"time"

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

func TestInitialMediaWaitIsBounded(t *testing.T) {
	raw, err := contracts.Read("examples/agent-version-full-explicit.json")
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := ai.ValidateForMode(raw, ai.ModeFullAI)
	if err != nil {
		t.Fatal(err)
	}
	_, err = Execute(context.Background(), NewMemorySession(nil), ai.MockPipeline{}, snapshot, "测试开场", time.Millisecond)
	if err == nil {
		t.Fatal("expected bounded initial media wait to fail")
	}
}

func TestSharedFlowRunsThreeConversationTurns(t *testing.T) {
	raw, err := contracts.Read("examples/agent-version-full-explicit.json")
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := ai.ValidateForMode(raw, ai.ModeFullAI)
	if err != nil {
		t.Fatal(err)
	}
	session := &scriptedTurnSession{turns: [][]byte{make([]byte, 6400), make([]byte, 6400), make([]byte, 6400)}}
	result, err := ExecuteWithCapture(context.Background(), session, ai.MockPipeline{MaxAudioBytes: 16 << 20}, snapshot, "测试开场", CaptureConfig{
		FirstSpeechTimeout: time.Second,
		MaxDuration:        2 * time.Millisecond,
		MaxTurns:           3,
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(result.Turns) != 3 || len(result.InboundTurns) != 3 || len(result.OutboundTurns) != 4 {
		t.Fatalf("expected opening plus three shared turns, got turns=%d inbound=%d outbound=%d", len(result.Turns), len(result.InboundTurns), len(result.OutboundTurns))
	}
	if result.Turn.Transcript == "" || result.Turn.Reply == "" || result.RTP.ReceivedBytes != 19200 {
		t.Fatalf("three-turn flow facts are incomplete: %+v", result)
	}
}

type scriptedTurnSession struct {
	turns  [][]byte
	index  int
	served bool
	stats  media.RTPStats
}

func (s *scriptedTurnSession) ReadPayload(ctx context.Context) ([]byte, error) {
	if !s.served && s.index < len(s.turns) {
		s.served = true
		payload := append([]byte(nil), s.turns[s.index]...)
		s.index++
		s.stats.ReceivedPackets++
		s.stats.ReceivedBytes += uint64(len(payload))
		return payload, nil
	}
	<-ctx.Done()
	s.served = false
	return nil, ctx.Err()
}

func (s *scriptedTurnSession) SendPCM16(ctx context.Context, pcm []byte, _ int) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	s.stats.SentPackets++
	s.stats.SentBytes += uint64(len(pcm))
	return nil
}

func (s *scriptedTurnSession) Stats() media.RTPStats { return s.stats }

type invalidCallPipeline struct{}

func (invalidCallPipeline) Synthesize(context.Context, ai.Snapshot, string) ([]byte, error) {
	return make([]byte, 6400), nil
}

func (invalidCallPipeline) RunTurn(context.Context, ai.Snapshot, []byte) (ai.TurnResult, error) {
	return ai.TurnResult{Transcript: "打错了", Reply: ai.InvalidCallMarker, AudioPCM16: make([]byte, 6400)}, nil
}

func TestInvalidCallStopsBeforeReply(t *testing.T) {
	raw, err := contracts.Read("examples/agent-version-full-explicit.json")
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := ai.ValidateForMode(raw, ai.ModeFullAI)
	if err != nil {
		t.Fatal(err)
	}
	session := &scriptedTurnSession{turns: [][]byte{make([]byte, 6400)}}
	result, err := ExecuteWithCapture(context.Background(), session, invalidCallPipeline{}, snapshot, "开场", CaptureConfig{
		FirstSpeechTimeout: time.Second,
		MaxDuration:        2 * time.Millisecond,
		MaxTurns:           3,
	})
	if err != nil {
		t.Fatal(err)
	}
	if !result.Turn.InvalidCall || result.Turn.InvalidReason != "llm_invalid_call_marker" || len(result.Turns) != 1 {
		t.Fatalf("invalid call was not classified: %+v", result)
	}
	if session.stats.SentPackets != 1 {
		t.Fatalf("invalid call must not send a reply, sent packets=%d", session.stats.SentPackets)
	}
}

func TestASROnlySkipsOpeningLLMAndTTS(t *testing.T) {
	raw, err := contracts.Read("examples/agent-version-asr-only.json")
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := ai.ValidateForMode(raw, ai.ModeASROnly)
	if err != nil {
		t.Fatal(err)
	}
	pipeline := &asrOnlyPipeline{}
	session := &scriptedTurnSession{turns: [][]byte{make([]byte, 6400)}}
	result, err := ExecuteWithCapture(context.Background(), session, pipeline, snapshot, "不得播放", CaptureConfig{
		FirstSpeechTimeout: time.Second,
		MaxDuration:        time.Millisecond,
		MaxTurns:           1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if pipeline.synthesizeCalls != 0 || pipeline.turnCalls != 1 {
		t.Fatalf("ASR-only provider calls = synthesize:%d turn:%d", pipeline.synthesizeCalls, pipeline.turnCalls)
	}
	if result.Turn.Transcript != "asr-only transcript" || result.Turn.Reply != "" || len(result.OutboundTurns) != 0 {
		t.Fatalf("unexpected ASR-only result: %+v", result)
	}
	if session.stats.SentPackets != 0 {
		t.Fatalf("ASR-only flow must not send opening/reply audio, sent packets=%d", session.stats.SentPackets)
	}
}

type asrOnlyPipeline struct {
	synthesizeCalls int
	turnCalls       int
}

func (p *asrOnlyPipeline) Synthesize(context.Context, ai.Snapshot, string) ([]byte, error) {
	p.synthesizeCalls++
	return nil, nil
}

func (p *asrOnlyPipeline) RunTurn(context.Context, ai.Snapshot, []byte) (ai.TurnResult, error) {
	p.turnCalls++
	return ai.TurnResult{Transcript: "asr-only transcript"}, nil
}

func TestMockModeUsesTheSharedConversationFlow(t *testing.T) {
	raw, err := contracts.Read("examples/agent-version-full-explicit.json")
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := ai.ValidateForMode(raw, ai.ModeFullAI)
	if err != nil {
		t.Fatal(err)
	}
	session := NewMemorySession(make([]byte, 6400))
	result, err := Execute(context.Background(), session, ai.MockPipeline{MaxAudioBytes: 16 << 20}, snapshot, "测试开场", 10*time.Millisecond)
	if err != nil {
		t.Fatal(err)
	}
	if result.Turn.Transcript == "" || result.Turn.Reply == "" || len(result.Turn.AudioPCM16) == 0 {
		t.Fatalf("shared flow did not complete mock ASR/LLM/TTS: %+v", result.Turn)
	}
	if result.RTP.ReceivedBytes != 6400 || result.RTP.SentPackets == 0 || len(session.OutboundPCM()) == 0 {
		t.Fatalf("shared media boundary was not exercised: %+v", result.RTP)
	}
}