story/internal/callmemory/direct_story_test.go

package callmemory

import (
	"context"
	"encoding/json"
	"strings"
	"testing"

	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/volcsubtitle"
	"github.com/google/uuid"
)

func TestDirectStoryTrackingBoundaries(t *testing.T) {
	state := storyContext{Version: storyContextVersion}
	sid := uuid.NewString()
	state.observe(sid, "q", "ASSISTANT", "您以前在厂里做什么工作呢？", 2000)
	state.observe(sid, "short", "USER", "嗯", 3000)
	if state.UserTurns != 0 || state.EarlyInvitations != 1 || state.Questions[0].Reply != nil {
		t.Fatal(state)
	}
	state.observe(sid, "quoted", "USER", "他说不想聊以前的工作", 4000)
	if len(state.Paused) != 0 {
		t.Fatal("third-party quotation became a refusal")
	}
	state.observe(sid, "q2", "ASSISTANT", "您以前在厂里做什么工作呢？", 190000)
	state.observe(sid, "no", "USER", "我不想聊以前的工作", 190100)
	if len(state.Paused) != 1 || state.Paused[0].Topic != "work" || state.Pending != nil {
		t.Fatal(state)
	}
	state.observe(sid, "own", "USER", "我年轻时在老家种过菜", 191000)
	state.observe(sid, "q3", "ASSISTANT", "老家以前怎么种菜呢？", 192000)
	if state.Invitations != 2 || len(state.Paused) != 1 {
		t.Fatal("user-led story counted as invitation or erased prior boundary", state)
	}
	raw, _ := json.Marshal(state)
	if strings.Contains(string(raw), "不想聊") || strings.Contains(string(raw), "种过菜") {
		t.Fatal("duplicated transcript in snapshot", string(raw))
	}
	state.observe(sid, "all", "USER", "我不想聊以前的事", 193000)
	if state.Paused[len(state.Paused)-1].Topic != "all" {
		t.Fatal("global boundary not tracked")
	}
}

func storyFixture(t *testing.T) (fixture, deviceauth.Identity, *string) {
	f, d := directFixture(t)
	f.add(t)
	brief := ""
	f.s.direct.Prepare = func(_ context.Context, _, _, _, text string) error { brief = text; return nil }
	return f, d, &brief
}
func storyStart(t *testing.T, f fixture, d deviceauth.Identity) string {
	t.Helper()
	sid := uuid.NewString()
	if _, e := f.s.DirectRequest(context.Background(), d, "start", DirectInput{SessionID: sid}); e != nil {
		t.Fatal(e)
	}
	return sid
}
func storyStop(t *testing.T, f fixture, d deviceauth.Identity, sid string) {
	t.Helper()
	if _, e := f.s.DirectRequest(context.Background(), d, "end", DirectInput{SessionID: sid, Reason: "hangup", SDKStopped: true}); e != nil {
		t.Fatal(e)
	}
}
func storySubtitle(t *testing.T, f fixture, sid, speaker, text string, seq int64) {
	t.Helper()
	round := int64(1)
	if e := f.s.DirectSubtitle(context.Background(), sid, volcsubtitle.Packet{Type: "subtitle", Data: []volcsubtitle.Segment{{UserID: speaker, RoundID: &round, Sequence: &seq, Text: text, Definite: true}}}); e != nil {
		t.Fatal(e)
	}
}
func storyRead(t *testing.T, f fixture, sid string) storyContext {
	t.Helper()
	var raw []byte
	if e := f.p.QueryRow(context.Background(), `SELECT budget_snapshot->'story_context' FROM call_session WHERE id=$1::uuid`, sid).Scan(&raw); e != nil {
		t.Fatal(e)
	}
	var state storyContext
	if e := json.Unmarshal(raw, &state); e != nil {
		t.Fatal(e)
	}
	return state
}

func TestDirectStorySourceContextAndRevocation(t *testing.T) {
	f, d, brief := storyFixture(t)
	sid := storyStart(t, f, d)
	storySubtitle(t, f, sid, "output", "您以前在厂里做什么工作呢？", 1)
	storySubtitle(t, f, sid, "input", "我在厂里负责修机器", 2)
	storyStop(t, f, d, sid)
	next := storyStart(t, f, d)
	if !strings.Contains(*brief, "recently_asked_questions") || !strings.Contains(*brief, "我在厂里负责修机器") {
		t.Fatal("source-backed question/reply not loaded", *brief)
	}
	if len(storyRead(t, f, next).Questions) != 1 {
		t.Fatal("question history not carried forward")
	}
	frozen := *brief
	if _, e := f.s.DirectRequest(context.Background(), d, "start", DirectInput{SessionID: next}); e != nil {
		t.Fatal(e)
	}
	if *brief != frozen {
		t.Fatal("renewal changed the frozen background")
	}
	storyStop(t, f, d, next)
	f.exec(t, `UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type='personalized_memory'`, f.subject)
	storyStart(t, f, d)
	if *brief != "" {
		t.Fatal("memory revoked but history sent to provider", *brief)
	}
}

func TestDirectStoryRefusalPersistsAndDuplicateCallbackIsIdempotent(t *testing.T) {
	f, d, brief := storyFixture(t)
	sid := storyStart(t, f, d)
	storySubtitle(t, f, sid, "output", "您以前在厂里做什么工作呢？", 1)
	storySubtitle(t, f, sid, "output", "您以前在厂里做什么工作呢？", 1)
	storySubtitle(t, f, sid, "input", "我不想聊以前的工作", 2)
	state := storyRead(t, f, sid)
	if state.Invitations != 1 || len(state.Paused) != 1 || len(state.Questions) != 1 || state.Questions[0].Reply != nil {
		t.Fatal(state)
	}
	storyStop(t, f, d, sid)
	next := storyStart(t, f, d)
	if !strings.Contains(*brief, `"topics_not_to_open_proactively":["work"]`) {
		t.Fatal("explicit boundary missing", *brief)
	}
	storySubtitle(t, f, next, "input", "我年轻时在厂里修过机器", 3)
	storyStop(t, f, d, next)
	last := storyStart(t, f, d)
	if len(storyRead(t, f, last).Paused) != 1 || !strings.Contains(*brief, `"work"`) {
		t.Fatal("voluntary story erased future boundary")
	}
}

func TestDirectStoryIsolationAndDeletedSource(t *testing.T) {
	f, d, brief := storyFixture(t)
	sid := storyStart(t, f, d)
	storySubtitle(t, f, sid, "output", "您以前在厂里做什么工作呢？", 1)
	storySubtitle(t, f, sid, "input", "我不想聊以前的工作", 2)
	storyStop(t, f, d, sid)
	f.exec(t, `UPDATE call_utterance SET text_content='[deleted]' WHERE session_id=$1::uuid`, sid)
	next := storyStart(t, f, d)
	if strings.Contains(*brief, "recently_asked_questions") || len(storyRead(t, f, next).Paused) != 0 {
		t.Fatal("deleted sources still used", *brief)
	}
	storyStop(t, f, d, next)
	g, gd, _ := storyFixture(t)
	foreign := storyStart(t, g, gd)
	storySubtitle(t, g, foreign, "output", "您以前在厂里做什么工作呢？", 1)
	storySubtitle(t, g, foreign, "input", "我不想聊以前的工作", 2)
	storyStop(t, g, gd, foreign)
	forged := storyRead(t, g, foreign)
	f.exec(t, `UPDATE call_session SET budget_snapshot=jsonb_set(budget_snapshot,'{story_context}',$2::jsonb) WHERE id=$1::uuid`, next, jsonBytes(forged))
	last := storyStart(t, f, d)
	if len(storyRead(t, f, last).Paused) != 0 || strings.Contains(*brief, "recently_asked_questions") {
		t.Fatal("foreign elder source leaked", *brief)
	}
	storyStop(t, f, d, last)
	f.exec(t, `UPDATE memory_user_state SET privacy_epoch=privacy_epoch+1 WHERE subject_id=$1::uuid`, f.subject)
	epochCall := storyStart(t, f, d)
	if len(storyRead(t, f, epochCall).Questions) != 0 {
		t.Fatal("old epoch history survived")
	}
}