story/internal/callmemory/direct_test.go

package callmemory

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"net/http/httptest"
	"os"
	"strings"
	"testing"

	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/volcsubtitle"
	"github.com/google/uuid"
	"go.yaml.in/yaml/v3"
)

func TestDirectOpenAPIRequiresNoStartup(t *testing.T) {
	raw, err := os.ReadFile("../../openapi/server-voice.yaml")
	if err != nil {
		t.Fatal(err)
	}
	var doc map[string]any
	if err = yaml.Unmarshal(raw, &doc); err != nil {
		t.Fatal(err)
	}
	paths := doc["paths"].(map[string]any)
	start := paths["/api/v1/device/chat/start"].(map[string]any)["post"].(map[string]any)
	schema := start["requestBody"].(map[string]any)["content"].(map[string]any)["application/json"].(map[string]any)["schema"].(map[string]any)
	required := schema["required"].([]any)
	if len(required) != 1 || required[0] != "session_id" {
		t.Fatal("startup field still required", required)
	}
	legacy := paths["/api/v1/device/voice-runtime"].(map[string]any)["post"].(map[string]any)
	if legacy["deprecated"] != true {
		t.Fatal("legacy startup not marked optional")
	}
}

func directOptions() DirectOptions {
	return DirectOptions{Template: []byte(`{"Config":{"S2SConfig":{"Provider":"volcano","ProviderParams":{"dialog":{"system_role":"聊天"}}},"MemoryConfig":{"Enable":true}}}`), BotID: "bot", CallbackURL: "https://example.com/callbacks/v1/volcengine/subtitles", CallbackSecret: []byte(strings.Repeat("k", 32))}
}
func TestDirectParametersAndCallbackAuthentication(t *testing.T) {
	s := NewService(nil, nil)
	s.ConfigureDirect(directOptions())
	sid := uuid.NewString()
	welcome := "今天咱们聊点老手艺。"
	raw, e := s.directParameters(sid, "bot", welcome)
	if e != nil {
		t.Fatal(e)
	}
	var out map[string]any
	json.Unmarshal(raw, &out)
	p := out["params"].(map[string]any)
	if p["PlatformTaskID"] != sid || p["Config"] != nil {
		t.Fatal("unsafe template")
	}
	if p["AgentConfig"].(map[string]any)["WelcomeMessage"] != welcome || !strings.Contains(string(raw), "session_id="+sid) {
		t.Fatal("missing configuration")
	}
	r := httptest.NewRequest("POST", "/callbacks/v1/volcengine/chat?session_id="+sid, strings.NewReader(`{"message":"e30=","signature":"wrong"}`))
	w := httptest.NewRecorder()
	s.DirectCallback(w, r)
	if w.Code != 401 {
		t.Fatal(w.Code)
	}
	s.direct.Template = nil
	if _, e = s.directParameters(sid, "bot", ""); e != nil {
		t.Fatal("preconfigured bot should not require a per-call template")
	}
}

func TestDirectParametersWelcomeFallback(t *testing.T) {
	s := NewService(nil, nil)
	s.ConfigureDirect(directOptions())
	for _, welcome := range []string{"", " \t\n"} {
		raw, err := s.directParameters(uuid.NewString(), "bot", welcome)
		if err != nil {
			t.Fatal(err)
		}
		var out map[string]any
		if err = json.Unmarshal(raw, &out); err != nil {
			t.Fatal(err)
		}
		agent := out["params"].(map[string]any)["AgentConfig"].(map[string]any)
		if agent["WelcomeMessage"] != "你好，今天想聊些什么？" {
			t.Fatal("blank welcome did not fall back", agent["WelcomeMessage"])
		}
	}
}

func directFixture(t *testing.T) (fixture, deviceauth.Identity) {
	f := setup(t)
	f.s.ConfigureDirect(directOptions())
	f.exec(t, `UPDATE device SET activation_state='ACTIVE' WHERE id=$1::uuid`, f.device)
	f.exec(t, `INSERT INTO device_activation_job(device_id,tenant_id,provider_device_name,provider_bot_id,status,binding_generation,config_version,config_ack_at,engine_ready_at,sdk_registered_at,config_expires_at,ready_binding_generation) VALUES($1::uuid,$2::uuid,$3,'bot','READY',1,1,now(),now(),now(),now()+interval '1 hour',1)`, f.device, f.tenant, "test-"+f.device)
	f.exec(t, `INSERT INTO subject_voice_agent(subject_id,bot_name,bot_id,status) VALUES($1::uuid,$2,$2,'READY')`, f.subject, "elder-bot-"+f.subject)
	return f, deviceauth.Identity{ID: f.device, TenantID: f.tenant, BindingGeneration: 1}
}

func TestDirectWelcomeFrozenPerSessionAndElder(t *testing.T) {
	f, d := directFixture(t)
	ctx := context.Background()
	firstWelcome := "今天咱们聊点老手艺。"
	nextWelcome := "我给您讲个老机器的小故事。"
	f.exec(t, `UPDATE subject_voice_agent SET agent_config=$2::jsonb WHERE subject_id=$1::uuid`, f.subject, jsonBytes(map[string]any{"WelcomeMessage": firstWelcome}))
	f.add(t)
	prepared := 0
	f.s.direct.Prepare = func(_ context.Context, subject, bot, session, contextText string) error {
		prepared++
		if subject != f.subject || bot != "elder-bot-"+f.subject || !validUUID(session) || contextText != "" {
			t.Fatal("incorrect elder or platform memory was injected", subject, bot, session, contextText)
		}
		return nil
	}
	checkWelcome := func(result any, expected string) {
		t.Helper()
		params := result.(map[string]any)["params"].(map[string]any)
		if params["AgentConfig"].(map[string]any)["WelcomeMessage"] != expected {
			t.Fatal("wrong runtime welcome", params)
		}
		if params["Config"] != nil || params["MemoryConfig"] != nil || params["memory"] != nil || strings.Contains(string(jsonBytes(params)), "喜欢喝淡茶") {
			t.Fatal("runtime exposed server configuration or memory", params)
		}
	}
	sid := uuid.NewString()
	in := DirectInput{SessionID: sid}
	first, err := f.s.DirectRequest(ctx, d, "start", in)
	if err != nil {
		t.Fatal(err)
	}
	checkWelcome(first, firstWelcome)
	f.exec(t, `UPDATE subject_voice_agent SET agent_config=$2::jsonb WHERE subject_id=$1::uuid`, f.subject, jsonBytes(map[string]any{"WelcomeMessage": nextWelcome}))
	retry, err := f.s.DirectRequest(ctx, d, "start", in)
	if err != nil {
		t.Fatal(err)
	}
	checkWelcome(retry, firstWelcome)
	if prepared != 1 || string(jsonBytes(first.(map[string]any)["params"])) != string(jsonBytes(retry.(map[string]any)["params"])) {
		t.Fatal("retry changed frozen parameters or prepared again", prepared)
	}
	if _, err = f.s.DirectRequest(ctx, d, "end", DirectInput{SessionID: sid, Reason: "hangup", SDKStopped: true}); err != nil {
		t.Fatal(err)
	}
	next, err := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: uuid.NewString()})
	if err != nil {
		t.Fatal(err)
	}
	checkWelcome(next, nextWelcome)
	if prepared != 2 {
		t.Fatal("new session did not prepare once", prepared)
	}
	other, otherDevice := directFixture(t)
	otherCall, err := other.s.DirectRequest(ctx, otherDevice, "start", DirectInput{SessionID: uuid.NewString()})
	if err != nil {
		t.Fatal(err)
	}
	checkWelcome(otherCall, "你好，今天想聊些什么？")
}

func TestDirectUsesElderBotAndRejectsUnboundDevice(t *testing.T) {
	f, d := directFixture(t)
	ctx := context.Background()
	sid := uuid.NewString()
	prepared := 0
	f.s.direct.Prepare = func(_ context.Context, subject, bot, session, contextText string) error {
		prepared++
		if subject != f.subject || bot != "elder-bot-"+f.subject || session != sid {
			t.Fatal("wrong elder preparation")
		}
		return nil
	}
	out, err := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: sid, BootID: f.boot})
	if err != nil {
		t.Fatal(err)
	}
	if out.(map[string]any)["bot_id"] != "elder-bot-"+f.subject {
		t.Fatal("used shared device bot")
	}
	if _, err = f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: sid, BootID: f.boot}); err != nil || prepared != 1 {
		t.Fatal("renewal repeated supplier update", prepared, err)
	}
	if _, err = f.s.DirectRequest(ctx, d, "end", DirectInput{SessionID: sid, Reason: "hangup", SDKStopped: true}); err != nil {
		t.Fatal(err)
	}
	f.exec(t, `UPDATE device_binding SET unbound_at=now() WHERE device_id=$1::uuid AND unbound_at IS NULL`, f.device)
	_, err = f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: uuid.NewString(), BootID: f.boot})
	if err == nil || err.Error() != "DEVICE_UNBOUND" {
		t.Fatalf("want unbound error, got %v", err)
	}
	// The end marker for an older call must still be accepted after unbinding.
	if _, err = f.s.DirectRequest(ctx, d, "end", DirectInput{SessionID: sid, Reason: "hangup", SDKStopped: true}); err != nil {
		t.Fatal(err)
	}
}
func TestDirectStartEndCallbackAndIsolation(t *testing.T) {
	f, d := directFixture(t)
	ctx := context.Background()
	sid := uuid.NewString()
	in := DirectInput{SessionID: sid, BootID: f.boot}
	first, e := f.s.DirectRequest(ctx, d, "start", in)
	if e != nil {
		t.Fatal(e)
	}
	second, e := f.s.DirectRequest(ctx, d, "start", in)
	if e != nil {
		t.Fatal(e)
	}
	if string(jsonBytes(first.(map[string]any)["params"])) != string(jsonBytes(second.(map[string]any)["params"])) {
		t.Fatal("retry changed params")
	}
	if _, e = f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: uuid.NewString(), BootID: f.boot}); e == nil {
		t.Fatal("concurrent call allowed")
	}
	end := DirectInput{SessionID: sid, Reason: "hangup", SDKStopped: true}
	if _, e = f.s.DirectRequest(ctx, d, "end", end); e != nil {
		t.Fatal(e)
	}
	if _, e = f.s.DirectRequest(ctx, d, "end", end); e != nil {
		t.Fatal(e)
	}
	if _, e = f.s.DirectRequest(ctx, d, "start", in); e == nil {
		t.Fatal("ended call resurrected")
	}
	n := int64(1)
	p := volcsubtitle.Packet{Type: "subtitle", Data: []volcsubtitle.Segment{{UserID: "input", RoundID: &n, Sequence: &n, Text: "测试文字", Definite: true}}}
	// Actual handler parsing + signature, followed by the database writer.
	envelope := volcsubtitle.Envelope{Message: base64.StdEncoding.EncodeToString(jsonBytes(p)), Signature: volcsubtitle.Signature(f.s.direct.CallbackSecret, "direct:"+sid)}
	for i := 0; i < 2; i++ {
		r := httptest.NewRequest("POST", "/callbacks/v1/volcengine/chat?session_id="+sid, strings.NewReader(string(jsonBytes(envelope))))
		w := httptest.NewRecorder()
		f.s.DirectCallback(w, r)
		if w.Code != 200 {
			t.Fatal(w.Code, w.Body.String())
		}
	}
	var count int
	f.p.QueryRow(ctx, `SELECT count(*) FROM call_utterance WHERE session_id=$1::uuid`, sid).Scan(&count)
	if count != 1 {
		t.Fatal("duplicate transcript", count)
	}
	// A later session never steals callbacks from the completed session.
	next := uuid.NewString()
	if _, e = f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: next, BootID: f.boot}); e != nil {
		t.Fatal(e)
	}
	if e = f.s.DirectSubtitle(ctx, next, volcsubtitle.Packet{Type: "subtitle", DeviceName: "another-device", Data: p.Data}); e == nil {
		t.Fatal("wrong device accepted")
	}
	p.Data[0].Text = "conflicting text"
	if e = f.s.DirectSubtitle(ctx, sid, p); e == nil {
		t.Fatal("conflicting duplicate accepted")
	}
	f.exec(t, `UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type='transcript_storage'`, f.subject)
	if e = f.s.DirectSubtitle(ctx, sid, p); e == nil {
		t.Fatal("revoked storage accepted")
	}
}
func TestDirectEndBeforeStartAndExpiredLease(t *testing.T) {
	f, d := directFixture(t)
	ctx := context.Background()
	sid := uuid.NewString()
	if _, e := f.s.DirectRequest(ctx, d, "end", DirectInput{SessionID: sid, Reason: "start_failed"}); e != nil {
		t.Fatal(e)
	}
	if _, e := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: sid, BootID: f.boot}); e == nil {
		t.Fatal("cancelled call started")
	}
	sid = uuid.NewString()
	in := DirectInput{SessionID: sid, BootID: f.boot}
	if _, e := f.s.DirectRequest(ctx, d, "start", in); e != nil {
		t.Fatal(e)
	}
	f.exec(t, `UPDATE call_session SET lease_expires_at=now()-interval '1 second' WHERE id=$1::uuid`, sid)
	if _, e := f.s.DirectRequest(ctx, d, "start", in); e == nil {
		t.Fatal("expired lease renewed")
	}
}

func TestNativeRTCAgentSpeaker(t *testing.T) {
	f, d := directFixture(t)
	sid := uuid.NewString()
	ctx := context.Background()
	if _, e := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: sid, BootID: f.boot}); e != nil {
		t.Fatal(e)
	}
	n := int64(0)
	if e := f.s.DirectSubtitle(ctx, sid, volcsubtitle.Packet{Type: "subtitle", Data: []volcsubtitle.Segment{{UserID: "bot_20260914095023000", RoundID: &n, Sequence: &n, Definite: true, Text: "你好"}}}); e != nil {
		t.Fatal(e)
	}
	var speaker string
	if e := f.p.QueryRow(ctx, `SELECT speaker FROM call_utterance WHERE session_id=$1::uuid`, sid).Scan(&speaker); e != nil || speaker != "ASSISTANT" {
		t.Fatal(speaker, e)
	}
	if isRTCAgentUser("bot_fake") || isRTCAgentUser("bot_2026091409502300x") {
		t.Fatal("unknown participant accepted as bot")
	}
}

func TestDirectWithoutStartupHandshake(t *testing.T) {
	f, d := directFixture(t)
	ctx := context.Background()
	// Real state immediately after platform registration: no local SDK receipts,
	// no boot registration and an expired legacy config are all irrelevant.
	f.exec(t, `UPDATE device SET onboarding_boot_id=NULL,activation_state='PROVISIONING' WHERE id=$1::uuid`, f.device)
	f.exec(t, `UPDATE device_activation_job SET status='REGISTERED_OBSERVED',config_ack_at=NULL,engine_ready_at=NULL,config_ack_boot_id=NULL,config_expires_at=now()-interval '1 day',ready_binding_generation=NULL WHERE device_id=$1::uuid`, f.device)
	sid := uuid.NewString()
	prepared := 0
	f.s.direct.Prepare = func(context.Context, string, string, string, string) error { prepared++; return nil }
	first, err := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: sid})
	if err != nil {
		t.Fatal(err)
	}
	// Renewals are session scoped, including old firmware with arbitrary boot IDs.
	for _, boot := range []string{"", uuid.NewString(), f.boot} {
		out, err := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: sid, BootID: boot})
		if err != nil || string(jsonBytes(first.(map[string]any)["params"])) != string(jsonBytes(out.(map[string]any)["params"])) {
			t.Fatal("renewal depends on boot registration", err)
		}
	}
	if prepared != 1 {
		t.Fatal("supplier configured more than once", prepared)
	}
	n := int64(1)
	if err = f.s.DirectSubtitle(ctx, sid, volcsubtitle.Packet{Type: "subtitle", Data: []volcsubtitle.Segment{{UserID: "bot_20260914095023000", RoundID: &n, Sequence: &n, Definite: true, Text: "你好"}}}); err != nil {
		t.Fatal(err)
	}
	if _, err = f.s.DirectRequest(ctx, d, "end", DirectInput{SessionID: sid, Reason: "hangup", SDKStopped: true}); err != nil {
		t.Fatal(err)
	}
	var state, job string
	var noReceipts bool
	if err = f.p.QueryRow(ctx, `SELECT d.activation_state,j.status,j.config_ack_at IS NULL AND j.engine_ready_at IS NULL FROM device d JOIN device_activation_job j ON j.device_id=d.id WHERE d.id=$1::uuid`, f.device).Scan(&state, &job, &noReceipts); err != nil {
		t.Fatal(err)
	}
	if state != "ACTIVE" || job != "READY" || !noReceipts {
		t.Fatal("activation requires/fabricates old receipts", state, job, noReceipts)
	}
	if _, err = f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: uuid.NewString()}); err != nil {
		t.Fatal("second call failed", err)
	}
}

func TestDirectStillRequiresPlatformRegistrationAndCurrentBinding(t *testing.T) {
	f, d := directFixture(t)
	ctx := context.Background()
	f.exec(t, `UPDATE device SET onboarding_boot_id=NULL WHERE id=$1::uuid`, f.device)
	f.exec(t, `UPDATE device_activation_job SET sdk_registered_at=NULL WHERE device_id=$1::uuid`, f.device)
	if _, err := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: uuid.NewString()}); err == nil || err.Error() != "DEVICE_NOT_READY" {
		t.Fatal("unregistered supplier allowed", err)
	}
	f.exec(t, `UPDATE device_activation_job SET sdk_registered_at=now() WHERE device_id=$1::uuid`, f.device)
	stale := d
	stale.BindingGeneration = 0
	if _, err := f.s.DirectRequest(ctx, stale, "start", DirectInput{SessionID: uuid.NewString()}); err == nil || err.Error() != "DEVICE_BINDING_CHANGED" {
		t.Fatal("stale binding allowed", err)
	}
	sid := uuid.NewString()
	if _, err := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: sid}); err != nil {
		t.Fatal(err)
	}
	f.exec(t, `UPDATE device_binding SET unbound_at=now() WHERE device_id=$1::uuid AND unbound_at IS NULL`, f.device)
	for _, id := range []string{sid, uuid.NewString()} {
		if _, err := f.s.DirectRequest(ctx, d, "start", DirectInput{SessionID: id}); err == nil || err.Error() != "DEVICE_UNBOUND" {
			t.Fatal("unbound start/renewal allowed", err)
		}
	}
	if _, err := f.s.DirectRequest(ctx, d, "end", DirectInput{SessionID: sid, Reason: "disconnected", SDKStopped: true}); err != nil {
		t.Fatal("end after unbind failed", err)
	}
}