story/internal/extraction/direct_story_integration_test.go

package extraction

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/outbox"
	"data-terminal-cloud/internal/saas"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

const directStoryTenant = "10000000-0000-0000-0000-000000000001"
const directStoryOwner = "10000000-0000-0000-0000-000000000101"

type directStoryGeneratorFunc func(context.Context, GenerationInput) (GenerationResult, error)

func (f directStoryGeneratorFunc) Generate(ctx context.Context, input GenerationInput) (GenerationResult, error) {
	return f(ctx, input)
}

type directStoryFixture struct {
	t                                       *testing.T
	pool                                    *pgxpool.Pool
	subject, publicSubject, device, session string
	service                                 *saas.Service
	owner                                   identity.Principal
}

func newDirectStoryFixture(t *testing.T) *directStoryFixture {
	t.Helper()
	if os.Getenv("TEST_DATABASE_URL") == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	pool, err := pgxpool.New(context.Background(), os.Getenv("TEST_DATABASE_URL"))
	if err != nil {
		t.Fatal(err)
	}
	if err := pool.Ping(context.Background()); err != nil {
		pool.Close()
		t.Fatal(err)
	}
	f := &directStoryFixture{t: t, pool: pool, subject: uuid.NewString(), publicSubject: uuid.NewString(), device: uuid.NewString(), session: uuid.NewString(),
		service: saas.NewService(pool, audit.NewRepository(), nil, nil),
		owner:   identity.Principal{Issuer: "http://localhost:58180/realms/voice-local", Subject: "00000000-0000-0000-0000-000000000101"}}
	t.Cleanup(f.cleanup)
	f.exec(`INSERT INTO subject_profile(id,public_id,tenant_id,household_id,preferred_name) VALUES($1::uuid,$2::uuid,$3::uuid,'10000000-0000-0000-0000-000000000201','DIRECT 故事合成测试')`, f.subject, f.publicSubject, directStoryTenant)
	f.exec(`INSERT INTO device(id,tenant_id,serial_digest,model,credential_ref) VALUES($1::uuid,$2::uuid,$3,'direct-story-test','synthetic')`, f.device, directStoryTenant, strings.ReplaceAll(uuid.NewString()+uuid.NewString(), "-", ""))
	f.exec(`INSERT INTO memory_user_state(subject_id,tenant_id) VALUES($1::uuid,$2::uuid)`, f.subject, directStoryTenant)
	f.exec(`INSERT INTO memory_access_grant(subject_id,user_id,can_read,can_write,sensitive_read) VALUES($1::uuid,$2::uuid,true,true,true)`, f.subject, directStoryOwner)
	f.exec(`INSERT INTO consent_record(tenant_id,subject_id,consent_type,policy_version,notice_digest,actor_user_id,granted_at)
 SELECT $1::uuid,$2::uuid,kind,'direct-story-test','synthetic',$3::uuid,now()-interval '1 hour'
 FROM unnest(ARRAY['story_extraction','transcript_storage','personalized_memory']) kind`, directStoryTenant, f.subject, directStoryOwner)
	f.exec(`INSERT INTO call_session(id,device_id,tenant_id,subject_id,boot_id,hook_event_id,call_generation,binding_generation,budget_snapshot,
 call_state,transcript_state,transcript_revision,created_at,ended_at)
 VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5::uuid,'synthetic',1,1,'{"protocol":"DIRECT-CHAT-1","story_extraction_allowed":true}',
 'ENDED','COMPLETE',1,now()-interval '10 minutes',now())`, f.session, f.device, directStoryTenant, f.subject, uuid.NewString())
	f.exec(`INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES
 ($1::uuid,'user-final',1,'USER','我年轻时在河边小学教书，每天走五里山路去上课。记得有一次下大雨，我们一起把课本搬到楼上，后来孩子们给我送来自己采的野花，我特别高兴。',false,now()-interval '2 minutes'),
 ($1::uuid,'assistant-final',1,'ASSISTANT','助手虚构内容：我曾经驾驶飞船登上月球。',false,now()-interval '1 minute'),
 ($1::uuid,'unknown-final',1,'UNKNOWN','未知说话人内容',false,now())`, f.session)
	return f
}

func (f *directStoryFixture) exec(sql string, args ...any) {
	f.t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if _, err := f.pool.Exec(ctx, sql, args...); err != nil {
		f.t.Fatal(err)
	}
}

func (f *directStoryFixture) event(revision int) outbox.Event {
	return outbox.Event{AggregateID: f.session, EventType: DirectStoryEvent, Payload: []byte(fmt.Sprintf(`{"generation_version":%d,"privacy_epoch":0}`, revision))}
}

func (f *directStoryFixture) processor(g Generator) *Processor {
	f.t.Helper()
	p, err := NewProcessor(f.pool, g)
	if err != nil {
		f.t.Fatal(err)
	}
	return p
}

func (f *directStoryFixture) handle(p *Processor, revision int) {
	f.t.Helper()
	if err := p.HandleDirectStory(context.Background(), f.event(revision)); err != nil {
		f.t.Fatal(err)
	}
}

func (f *directStoryFixture) story(revision int) saas.Story {
	f.t.Helper()
	var publicID string
	if err := f.pool.QueryRow(context.Background(), `SELECT public_id::text FROM story_draft WHERE call_session_id=$1::uuid AND generation_version=$2`, f.session, revision).Scan(&publicID); err != nil {
		f.t.Fatal(err)
	}
	story, err := f.service.GetStory(context.Background(), f.owner, publicID)
	if err != nil {
		f.t.Fatal(err)
	}
	return story
}

func (f *directStoryFixture) assertCounts(stories, extractions int) {
	f.t.Helper()
	var actualStories, actualExtractions int
	if err := f.pool.QueryRow(context.Background(), `SELECT (SELECT count(*) FROM story_draft WHERE call_session_id=$1::uuid),
 (SELECT count(*) FROM call_story_extraction WHERE session_id=$1::uuid)`, f.session).Scan(&actualStories, &actualExtractions); err != nil {
		f.t.Fatal(err)
	}
	if actualStories != stories || actualExtractions != extractions {
		f.t.Fatalf("stories=%d extractions=%d; want %d/%d", actualStories, actualExtractions, stories, extractions)
	}
}

func (f *directStoryFixture) appendFinal(revision int) {
	f.exec(`INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at)
 VALUES($1::uuid,$2,$3,'USER','后来我们给学校修好了屋顶，孩子们再也不用淋雨。',false,now())`, f.session, fmt.Sprintf("late-%d", revision), revision)
	f.exec(`UPDATE call_session SET transcript_revision=$2 WHERE id=$1::uuid`, f.session, revision)
}

func (f *directStoryFixture) cleanup() {
	defer f.pool.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	for _, sql := range []string{
		`DELETE FROM outbox_event WHERE aggregate_id IN (SELECT id FROM curated_memory WHERE subject_id=$1::uuid) OR aggregate_id IN (SELECT id FROM call_session WHERE subject_id=$1::uuid)`,
		`DELETE FROM audit_event WHERE resource_id IN (SELECT public_id FROM story_draft WHERE subject_id=$1::uuid)`,
		`DELETE FROM content_review_signal WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_index_entry WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_projection_job WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_projection WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_source_suppression WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_deletion_ledger WHERE subject_id=$1::uuid`,
		`UPDATE curated_memory SET current_version_id=NULL WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_version WHERE memory_id IN (SELECT id FROM curated_memory WHERE subject_id=$1::uuid)`,
		`DELETE FROM curated_memory WHERE subject_id=$1::uuid`,
		`DELETE FROM fact_candidate WHERE subject_id=$1::uuid`,
		`DELETE FROM story_source WHERE story_revision_id IN (SELECT sr.id FROM story_revision sr JOIN story_draft sd ON sd.id=sr.story_id WHERE sd.subject_id=$1::uuid)`,
		`UPDATE story_draft SET current_revision_id=NULL WHERE subject_id=$1::uuid`,
		`DELETE FROM story_revision WHERE story_id IN (SELECT id FROM story_draft WHERE subject_id=$1::uuid)`,
		`DELETE FROM story_draft WHERE subject_id=$1::uuid`,
		`DELETE FROM call_story_extraction WHERE session_id IN (SELECT id FROM call_session WHERE subject_id=$1::uuid)`,
		`DELETE FROM call_command WHERE session_id IN (SELECT id FROM call_session WHERE subject_id=$1::uuid)`,
		`DELETE FROM call_utterance WHERE session_id IN (SELECT id FROM call_session WHERE subject_id=$1::uuid)`,
		`DELETE FROM call_session WHERE subject_id=$1::uuid`,
		`DELETE FROM privacy_request WHERE subject_id=$1::uuid`,
		`DELETE FROM consent_record WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_access_grant WHERE subject_id=$1::uuid`,
		`DELETE FROM memory_user_state WHERE subject_id=$1::uuid`,
		`DELETE FROM subject_profile WHERE id=$1::uuid`,
	} {
		if _, err := f.pool.Exec(ctx, sql, f.subject); err != nil {
			f.t.Errorf("fixture cleanup: %v", err)
		}
	}
	if _, err := f.pool.Exec(ctx, `DELETE FROM device WHERE id=$1::uuid`, f.device); err != nil {
		f.t.Errorf("fixture device cleanup: %v", err)
	}
}

func TestDirectStoryDraftReviewAndSave(t *testing.T) {
	f := newDirectStoryFixture(t)
	calls := 0
	p := f.processor(directStoryGeneratorFunc(func(ctx context.Context, input GenerationInput) (GenerationResult, error) {
		calls++
		if strings.Contains(input.Content, "助手") || strings.Contains(input.Content, "未知") || !strings.Contains(input.Content, "河边小学") {
			t.Fatalf("unexpected model input: %q", input.Content)
		}
		return (DeterministicGenerator{}).Generate(ctx, input)
	}))
	f.handle(p, 1)
	f.handle(p, 1)
	if calls != 1 {
		t.Fatalf("duplicate event called model %d times", calls)
	}
	f.assertCounts(1, 1)
	story := f.story(1)
	if story.Status != "CANDIDATE" || story.ConversationID != "" || story.CallSessionID != f.session || len(story.Sources) != 1 || len(story.Facts) != 1 {
		t.Fatalf("unexpected story: %+v", story)
	}
	source := story.Sources[0]
	if source.CallSessionID != f.session || source.UtteranceID != "user-final" || source.SourceRevision == nil || *source.SourceRevision != 1 || source.Speaker != "subject" {
		t.Fatalf("unexpected DIRECT source: %+v", source)
	}
	stories, err := f.service.ListSubjectStories(context.Background(), f.owner, f.publicSubject, "CANDIDATE")
	if err != nil || len(stories) != 1 || stories[0].ID != story.ID {
		t.Fatalf("story inbox=%+v err=%v", stories, err)
	}
	const edited = "我年轻时在河边小学教书，每天走五里山路，给孩子们上语文课。"
	revised, err := f.service.ReviseStory(context.Background(), f.owner, story.ID, edited, "核对原文", uuid.NewString(), story.Version, "")
	if err != nil {
		t.Fatal(err)
	}
	if revised.Status != "IN_REVIEW" || revised.RevisionNo != 2 || len(revised.Sources) != 1 || revised.Sources[0].UtteranceID != "user-final" {
		t.Fatalf("revised story lost source: %+v", revised)
	}
	confirmed, memory, err := f.service.ConfirmStory(context.Background(), f.owner, revised.ID, "已审核", uuid.NewString(), revised.Version)
	if err != nil {
		t.Fatal(err)
	}
	if confirmed.Status != "CONFIRMED" || memory.Status != "ACTIVE" || memory.Content != edited {
		t.Fatalf("save result: story=%+v memory=%+v", confirmed, memory)
	}
	refs, _ := json.Marshal(memory.SourceRefs)
	if !strings.Contains(string(refs), f.session) || !strings.Contains(string(refs), "user-final") || strings.Contains(string(refs), "assistant-final") {
		t.Fatalf("saved source refs=%s", refs)
	}
	f.handle(p, 1)
	if calls != 1 {
		t.Fatalf("confirmed replay called model %d times", calls)
	}
}

func TestDirectStoryAuthorizationGates(t *testing.T) {
	for _, gate := range []string{"snapshot_denied", "no_story_consent", "no_transcript_consent", "late_consent", "old_privacy_epoch", "delete_request", "old_event_revision", "old_event_epoch"} {
		t.Run(gate, func(t *testing.T) {
			f := newDirectStoryFixture(t)
			event := f.event(1)
			switch gate {
			case "snapshot_denied":
				f.exec(`UPDATE call_session SET budget_snapshot=budget_snapshot||'{"story_extraction_allowed":false}'::jsonb WHERE id=$1::uuid`, f.session)
			case "no_story_consent":
				f.exec(`DELETE FROM consent_record WHERE subject_id=$1::uuid AND consent_type='story_extraction'`, f.subject)
			case "no_transcript_consent":
				f.exec(`DELETE FROM consent_record WHERE subject_id=$1::uuid AND consent_type='transcript_storage'`, f.subject)
			case "late_consent":
				f.exec(`UPDATE consent_record SET granted_at=now() WHERE subject_id=$1::uuid AND consent_type='story_extraction'`, f.subject)
			case "old_privacy_epoch":
				f.exec(`UPDATE memory_user_state SET privacy_epoch=1 WHERE subject_id=$1::uuid`, f.subject)
			case "delete_request":
				f.exec(`INSERT INTO privacy_request(tenant_id,subject_id,request_type,requested_by,idempotency_key,request_digest) VALUES($1::uuid,$2::uuid,'DELETE',$3::uuid,$4,$5)`, directStoryTenant, f.subject, directStoryOwner, uuid.NewString(), strings.Repeat("a", 64))
				// Match epochs to prove DELETE is independently disqualifying.
				f.exec(`UPDATE call_session SET privacy_epoch=(SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid) WHERE id=$2::uuid`, f.subject, f.session)
				event.Payload = []byte(`{"generation_version":1,"privacy_epoch":1}`)
			case "old_event_revision":
				f.exec(`UPDATE call_session SET transcript_revision=2 WHERE id=$1::uuid`, f.session)
			case "old_event_epoch":
				event.Payload = []byte(`{"generation_version":1,"privacy_epoch":1}`)
			}
			p := f.processor(directStoryGeneratorFunc(func(context.Context, GenerationInput) (GenerationResult, error) {
				t.Fatal("unauthorized or stale event reached model")
				return GenerationResult{}, nil
			}))
			if err := p.HandleDirectStory(context.Background(), event); err != nil {
				t.Fatal(err)
			}
			f.assertCounts(0, 0)
		})
	}
}

func TestDirectStoryLateFinalPreservesHumanReview(t *testing.T) {
	f := newDirectStoryFixture(t)
	calls := 0
	p := f.processor(directStoryGeneratorFunc(func(ctx context.Context, input GenerationInput) (GenerationResult, error) {
		calls++
		return (DeterministicGenerator{}).Generate(ctx, input)
	}))
	f.handle(p, 1)
	f.appendFinal(2)
	f.handle(p, 2)
	f.assertCounts(2, 2)
	var oldStatus, factStatus string
	if err := f.pool.QueryRow(context.Background(), `SELECT sd.status,fc.status FROM story_draft sd JOIN fact_candidate fc ON fc.story_id=sd.id WHERE sd.call_session_id=$1::uuid AND sd.generation_version=1`, f.session).Scan(&oldStatus, &factStatus); err != nil {
		t.Fatal(err)
	}
	if oldStatus != "SUPERSEDED" || factStatus != "SUPERSEDED" {
		t.Fatalf("untouched candidate not replaced: %s/%s", oldStatus, factStatus)
	}
	story := f.story(2)
	if len(story.Sources) != 2 {
		t.Fatalf("late final source count=%d", len(story.Sources))
	}
	const human = "人工审核后的河边小学故事。"
	revised, err := f.service.ReviseStory(context.Background(), f.owner, story.ID, human, "核对", uuid.NewString(), story.Version, "")
	if err != nil {
		t.Fatal(err)
	}
	f.appendFinal(3)
	f.handle(p, 3)
	f.handle(p, 3)
	f.assertCounts(2, 3)
	if calls != 2 {
		t.Fatalf("late final after human review called model: %d", calls)
	}
	after := f.story(2)
	if after.Status != "IN_REVIEW" || after.Content != human || after.Version != revised.Version {
		t.Fatalf("human review overwritten: %+v", after)
	}
}

func TestDirectStoryModelResultRevalidated(t *testing.T) {
	for _, change := range []string{"revoke_story_consent", "revise_transcript", "human_review"} {
		t.Run(change, func(t *testing.T) {
			f := newDirectStoryFixture(t)
			revision := 1
			var old saas.Story
			if change == "human_review" {
				f.handle(f.processor(DeterministicGenerator{}), 1)
				old = f.story(1)
				f.appendFinal(2)
				revision = 2
			}
			p := f.processor(directStoryGeneratorFunc(func(ctx context.Context, input GenerationInput) (GenerationResult, error) {
				// A synchronous callback represents state changing while the model is
				// in flight and also proves no database transaction locks span it.
				switch change {
				case "revoke_story_consent":
					f.exec(`UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type='story_extraction'`, f.subject)
				case "revise_transcript":
					f.exec(`UPDATE call_utterance SET text_content='更正后的回忆',source_revision=2 WHERE session_id=$1::uuid AND utterance_id='user-final'`, f.session)
					f.exec(`UPDATE call_session SET transcript_revision=2 WHERE id=$1::uuid`, f.session)
				case "human_review":
					reviewCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
					defer cancel()
					if _, err := f.service.ReviseStory(reviewCtx, f.owner, old.ID, "保留人工修订内容", "审核", uuid.NewString(), old.Version, ""); err != nil {
						return GenerationResult{}, err
					}
				}
				return (DeterministicGenerator{}).Generate(ctx, input)
			}))
			f.handle(p, revision)
			if change == "human_review" {
				f.assertCounts(1, 2)
				if story := f.story(1); story.Status != "IN_REVIEW" || story.Content != "保留人工修订内容" {
					t.Fatalf("model overwrote review: %+v", story)
				}
			} else {
				f.assertCounts(0, 0)
			}
		})
	}
}

func TestDirectStoryReviewRejectsChangedSources(t *testing.T) {
	for _, change := range []string{"revision_changed", "source_deleted", "speaker_changed", "consent_withdrawn"} {
		t.Run(change, func(t *testing.T) {
			f := newDirectStoryFixture(t)
			f.handle(f.processor(DeterministicGenerator{}), 1)
			story := f.story(1)
			switch change {
			case "revision_changed":
				f.exec(`UPDATE call_utterance SET source_revision=2,text_content='已更正的转写' WHERE session_id=$1::uuid AND utterance_id='user-final'`, f.session)
			case "source_deleted":
				f.exec(`UPDATE call_utterance SET text_content='[deleted]' WHERE session_id=$1::uuid AND utterance_id='user-final'`, f.session)
			case "speaker_changed":
				f.exec(`UPDATE call_utterance SET speaker='ASSISTANT' WHERE session_id=$1::uuid AND utterance_id='user-final'`, f.session)
			case "consent_withdrawn":
				f.exec(`UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type='story_extraction'`, f.subject)
			}
			ctx := context.Background()
			if _, err := f.service.GetStory(ctx, f.owner, story.ID); !errors.Is(err, saas.ErrAccessDenied) {
				t.Fatalf("stale source remains readable: %v", err)
			}
			if _, _, err := f.service.ConfirmStory(ctx, f.owner, story.ID, "审核", uuid.NewString(), story.Version); !errors.Is(err, saas.ErrAccessDenied) {
				t.Fatalf("stale source was confirmable: %v", err)
			}
			var saved int
			if err := f.pool.QueryRow(ctx, `SELECT count(*) FROM curated_memory WHERE subject_id=$1::uuid`, f.subject).Scan(&saved); err != nil {
				t.Fatal(err)
			}
			if saved != 0 {
				t.Fatalf("unavailable source saved %d memories", saved)
			}
		})
	}
}