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) } }