story/internal/saas/service_integration_test.go

package saas

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

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/pagination"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

const (
	testTenantPublicID = "10000000-0000-0000-0000-000000000010"
	testIssuer         = "http://localhost:58180/realms/voice-local"
	testOwnerSubject   = "00000000-0000-0000-0000-000000000101"
	testEditorSubject  = "00000000-0000-0000-0000-000000000102"
	testOutsideSubject = "00000000-0000-0000-0000-000000000103"
)

func TestHouseholdSubjectConsentLifecycleAndIsolation(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	service := NewService(pool, audit.NewRepository(), nil, nil)
	owner := identity.Principal{Issuer: testIssuer, Subject: testOwnerSubject}
	editor := identity.Principal{Issuer: testIssuer, Subject: testEditorSubject}
	outsider := identity.Principal{Issuer: testIssuer, Subject: testOutsideSubject}

	household, err := service.CreateHousehold(ctx, owner, testTenantPublicID, "集成测试家庭", "saas-integration-create")
	if err != nil {
		t.Fatalf("create household: %v", err)
	}
	var subjectID string
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE request_id LIKE 'saas-integration-%'`)
		if subjectID != "" {
			_, _ = pool.Exec(ctx, `DELETE FROM outbox_event WHERE aggregate_id IN (SELECT id FROM privacy_request WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid) UNION SELECT id FROM export_task WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid))`, subjectID)
			_, _ = pool.Exec(ctx, `DELETE FROM privacy_request WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid)`, subjectID)
			_, _ = pool.Exec(ctx, `DELETE FROM export_task WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid)`, subjectID)
			_, _ = pool.Exec(ctx, `DELETE FROM consent_snapshot WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid)`, subjectID)
			_, _ = pool.Exec(ctx, `DELETE FROM consent_record WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid)`, subjectID)
			_, _ = pool.Exec(ctx, `DELETE FROM subject_fact WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid)`, subjectID)
			_, _ = pool.Exec(ctx, `DELETE FROM subject_profile WHERE public_id=$1::uuid`, subjectID)
		}
		_, _ = pool.Exec(ctx, `DELETE FROM household_member WHERE household_id=(SELECT id FROM household WHERE public_id=$1::uuid)`, household.ID)
		_, _ = pool.Exec(ctx, `DELETE FROM household WHERE public_id=$1::uuid`, household.ID)
	}()

	if household.Version != 1 || household.Role != identity.RoleOwner {
		t.Fatalf("unexpected household: %+v", household)
	}
	items, err := service.ListHouseholds(ctx, owner, testTenantPublicID)
	if err != nil || !containsHousehold(items, household.ID) {
		t.Fatalf("list household: items=%+v err=%v", items, err)
	}
	if _, err := service.GetHousehold(ctx, outsider, household.ID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider get error=%v", err)
	}

	if _, err := pool.Exec(ctx, `
		INSERT INTO household_member (tenant_id,household_id,user_id,role,status)
		SELECT h.tenant_id,h.id,ua.id,'editor','active'
		FROM household h JOIN user_account ua ON ua.tenant_id=h.tenant_id
		WHERE h.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, household.ID, testIssuer, testEditorSubject); err != nil {
		t.Fatalf("add editor: %v", err)
	}
	if _, err := service.PatchHousehold(ctx, editor, household.ID, "禁止修改", 1, "saas-integration-editor-denied"); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("editor manage error=%v", err)
	}
	household, err = service.PatchHousehold(ctx, owner, household.ID, "集成测试家庭-已修改", 1, "saas-integration-household-patch")
	if err != nil || household.Version != 2 {
		t.Fatalf("patch household: %+v err=%v", household, err)
	}
	if _, err := service.PatchHousehold(ctx, owner, household.ID, "过期写入", 1, "saas-integration-stale"); !errors.Is(err, ErrVersionConflict) {
		t.Fatalf("stale household patch error=%v", err)
	}

	subject, err := service.CreateSubject(ctx, editor, household.ID, "saas-integration-subject-create", CreateSubjectInput{
		InitialConsents: InitialConsentSelection{Types: []string{ConsentTranscriptStorage, ConsentPersonalMemory}, Confirmed: true},
		PreferredName:   "集成长者", Language: "zh-CN", Timezone: "Asia/Shanghai",
		ProfileData: map[string]any{"synthetic": true}, ConversationPreferences: map[string]any{"pace": "slow"},
	})
	if err != nil {
		t.Fatalf("create subject: %v", err)
	}
	subjectID = subject.ID
	initialConsents, err := service.ListConsents(ctx, editor, subject.ID)
	if err != nil || len(initialConsents) != 2 {
		t.Fatalf("initial memory consents: %+v err=%v", initialConsents, err)
	}
	for _, c := range initialConsents {
		if !c.Effective || c.PolicyVersion != "local-v1" || c.ActorUserID == "" || c.NoticeDigest == "" || (c.Type != ConsentTranscriptStorage && c.Type != ConsentPersonalMemory) {
			t.Fatalf("invalid initial consent: %+v", c)
		}
	}
	var snapshotOK bool
	if err := pool.QueryRow(ctx, `SELECT effective_consents @> '{"voice_chat":true,"transcript_storage":true,"personalized_memory":true,"raw_audio_storage":false,"story_extraction":false,"family_sharing":false}'::jsonb AND jsonb_array_length(source_consent_ids)=2 FROM consent_snapshot WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid) ORDER BY snapshot_version DESC LIMIT 1`, subject.ID).Scan(&snapshotOK); err != nil || !snapshotOK {
		t.Fatalf("initial memory snapshot=%v err=%v", snapshotOK, err)
	}
	var grantAuditCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE request_id='saas-integration-subject-create' AND action='consent.grant'`).Scan(&grantAuditCount); err != nil || grantAuditCount != 2 {
		t.Fatalf("initial memory audit=%d err=%v", grantAuditCount, err)
	}
	subjects, err := service.ListSubjects(ctx, editor, household.ID)
	if err != nil || len(subjects) != 1 || subjects[0].ID != subject.ID {
		t.Fatalf("list subjects: %+v err=%v", subjects, err)
	}
	if _, err := service.GetSubject(ctx, outsider, subject.ID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider subject error=%v", err)
	}
	newName := "集成长者-已修改"
	subject, err = service.PatchSubject(ctx, editor, subject.ID, "saas-integration-subject-patch", 1, PatchSubjectInput{PreferredName: &newName})
	if err != nil || subject.Version != 2 || subject.PreferredName != newName {
		t.Fatalf("patch subject: %+v err=%v", subject, err)
	}
	fact, err := service.CreateSubjectFact(ctx, editor, subject.ID, "saas-integration-fact-create", CreateFactInput{
		FactType: "hometown", Content: "合成城市", SourceType: "model_candidate", SourceRef: "synthetic://turn/1",
		ConfirmationStatus: "pending", Sensitivity: "normal",
	})
	if err != nil || fact.Version != 1 {
		t.Fatalf("create fact: %+v err=%v", fact, err)
	}
	confirmed := "confirmed"
	fact, err = service.PatchSubjectFact(ctx, editor, subject.ID, fact.ID, "saas-integration-fact-confirm", 1, PatchFactInput{ConfirmationStatus: &confirmed})
	if err != nil || fact.Version != 2 || fact.ConfirmationStatus != "confirmed" {
		t.Fatalf("confirm fact: %+v err=%v", fact, err)
	}
	if _, err := service.PatchSubjectFact(ctx, editor, subject.ID, fact.ID, "saas-integration-fact-stale", 1, PatchFactInput{ConfirmationStatus: &confirmed}); !errors.Is(err, ErrVersionConflict) {
		t.Fatalf("stale fact patch error=%v", err)
	}
	facts, err := service.ListSubjectFacts(ctx, editor, subject.ID)
	if err != nil || len(facts) != 1 {
		t.Fatalf("list facts: %+v err=%v", facts, err)
	}

	grantInput := GrantConsentInput{
		Type: ConsentStoryExtraction, Scope: map[string]any{"scope": "future_conversations"},
		PolicyVersion: "local-v1", NoticeDigest: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		EvidenceRef: "synthetic://integration", GrantedAt: time.Now().UTC(),
	}
	consent, err := service.GrantConsent(ctx, editor, subject.ID, "saas-integration-consent-grant", grantInput)
	if err != nil || !consent.Effective {
		t.Fatalf("grant consent: %+v err=%v", consent, err)
	}
	if _, err := service.GrantConsent(ctx, editor, subject.ID, "saas-integration-consent-duplicate", grantInput); !errors.Is(err, ErrEffectiveConsentExist) {
		t.Fatalf("duplicate grant error=%v", err)
	}
	consent, err = service.WithdrawConsent(ctx, editor, subject.ID, consent.ID, "saas-integration-consent-withdraw")
	if err != nil || consent.WithdrawnAt == nil || consent.Effective {
		t.Fatalf("withdraw consent: %+v err=%v", consent, err)
	}
	replayed, err := service.WithdrawConsent(ctx, editor, subject.ID, consent.ID, "saas-integration-consent-withdraw-replay")
	if err != nil || replayed.Version != consent.Version {
		t.Fatalf("replay withdrawal: %+v err=%v", replayed, err)
	}
	consents, err := service.ListConsents(ctx, editor, subject.ID)
	if err != nil || len(consents) != 3 {
		t.Fatalf("list consents: %+v err=%v", consents, err)
	}
	privacyInput := CreatePrivacyRequestInput{RequestType: " access ", Scope: map[string]any{"categories": []any{"profile", "consents"}}}
	privacyRequest, created, err := service.CreatePrivacyRequest(ctx, owner, subject.ID, "saas-integration-privacy-key", "saas-integration-privacy-create", privacyInput)
	if err != nil || !created || privacyRequest.Status != "SUBMITTED" || len(privacyRequest.Subtasks) != 1 {
		t.Fatalf("privacy request=%+v created=%v err=%v", privacyRequest, created, err)
	}
	replayedPrivacy, created, err := service.CreatePrivacyRequest(ctx, owner, subject.ID, "saas-integration-privacy-key", "saas-integration-privacy-replay", privacyInput)
	if err != nil || created || replayedPrivacy.ID != privacyRequest.ID {
		t.Fatalf("privacy replay=%+v created=%v err=%v", replayedPrivacy, created, err)
	}
	conflictingPrivacy := privacyInput
	conflictingPrivacy.Scope = map[string]any{"categories": []any{"profile"}}
	if _, _, err := service.CreatePrivacyRequest(ctx, owner, subject.ID, "saas-integration-privacy-key", "saas-integration-privacy-conflict", conflictingPrivacy); !errors.Is(err, ErrIdempotencyConflict) {
		t.Fatalf("privacy idempotency conflict=%v", err)
	}
	correctedName := "集成长者-隐私更正"
	correction, created, err := service.CreatePrivacyRequest(ctx, owner, subject.ID, "saas-integration-correction-key", "saas-integration-correction", CreatePrivacyRequestInput{
		RequestType: "CORRECTION", Scope: map[string]any{"corrections": map[string]any{"preferred_name": correctedName}},
	})
	if err != nil || !created || correction.Status != "COMPLETED" || len(correction.Subtasks) != 1 || correction.Subtasks[0].Status != "COMPLETED" {
		t.Fatalf("correction request=%+v created=%v err=%v", correction, created, err)
	}
	correctedSubject, err := service.GetSubject(ctx, owner, subject.ID)
	if err != nil || correctedSubject.PreferredName != correctedName || correctedSubject.Version != subject.Version+1 {
		t.Fatalf("corrected subject=%+v err=%v", correctedSubject, err)
	}
	regranted, err := service.GrantConsent(ctx, editor, subject.ID, "saas-integration-consent-regrant", grantInput)
	if err != nil || !regranted.Effective {
		t.Fatalf("regrant consent=%+v err=%v", regranted, err)
	}
	withdrawal, created, err := service.CreatePrivacyRequest(ctx, owner, subject.ID, "saas-integration-withdrawal-key", "saas-integration-withdrawal", CreatePrivacyRequestInput{
		RequestType: "CONSENT_WITHDRAWAL", Scope: map[string]any{"consent_types": []string{ConsentStoryExtraction}},
	})
	if err != nil || !created || withdrawal.Status != "COMPLETED" || len(withdrawal.Subtasks) != 1 || withdrawal.Subtasks[0].Status != "COMPLETED" {
		t.Fatalf("withdrawal request=%+v created=%v err=%v", withdrawal, created, err)
	}
	consents, err = service.ListConsents(ctx, editor, subject.ID)
	if err != nil || len(consents) != 4 {
		t.Fatalf("consents after batch withdrawal=%+v err=%v", consents, err)
	}
	for _, c := range consents {
		if c.Type == ConsentStoryExtraction && c.Effective {
			t.Fatalf("story consent remains active after withdrawal: %+v", c)
		}
	}
	if _, err := service.ListPrivacyRequests(ctx, editor, subject.ID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("editor privacy list error=%v", err)
	}
	privacyRequests, err := service.ListPrivacyRequests(ctx, owner, subject.ID)
	if err != nil || len(privacyRequests) != 3 {
		t.Fatalf("privacy requests=%+v err=%v", privacyRequests, err)
	}
	if _, err := service.GetPrivacyRequest(ctx, outsider, privacyRequest.ID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider privacy request error=%v", err)
	}
	exportScope := map[string]any{"categories": []any{"profile", "consents"}}
	exportTask, created, err := service.CreateExportTask(ctx, owner, subject.ID, "saas-integration-export-key", "saas-integration-export-create", exportScope)
	if err != nil || !created || exportTask.Status != "QUEUED" {
		t.Fatalf("export task=%+v created=%v err=%v", exportTask, created, err)
	}
	replayedExport, created, err := service.CreateExportTask(ctx, owner, subject.ID, "saas-integration-export-key", "saas-integration-export-replay", exportScope)
	if err != nil || created || replayedExport.ID != exportTask.ID {
		t.Fatalf("export replay=%+v created=%v err=%v", replayedExport, created, err)
	}
	if _, _, err := service.CreateExportTask(ctx, owner, subject.ID, "saas-integration-export-key", "saas-integration-export-conflict", map[string]any{"categories": []any{"profile"}}); !errors.Is(err, ErrIdempotencyConflict) {
		t.Fatalf("export idempotency conflict=%v", err)
	}
	if _, err := service.ListExportTasks(ctx, editor, subject.ID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("editor export list error=%v", err)
	}
	exports, err := service.ListExportTasks(ctx, owner, subject.ID)
	if err != nil || len(exports) != 1 || exports[0].ID != exportTask.ID {
		t.Fatalf("export tasks=%+v err=%v", exports, err)
	}
	if _, err := service.GetExportTask(ctx, outsider, exportTask.ID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider export task error=%v", err)
	}

	var snapshotCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM consent_snapshot WHERE subject_id=(SELECT id FROM subject_profile WHERE public_id=$1::uuid)`, subject.ID).Scan(&snapshotCount); err != nil {
		t.Fatal(err)
	}
	if snapshotCount != 5 {
		t.Fatalf("snapshot count=%d, want 5", snapshotCount)
	}
	var auditCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE request_id LIKE 'saas-integration-%'`).Scan(&auditCount); err != nil {
		t.Fatal(err)
	}
	if auditCount != 15 {
		t.Fatalf("audit count=%d, want 15", auditCount)
	}

	_ = uuid.MustParse(household.ID)
	_ = uuid.MustParse(subject.ID)
	_ = uuid.MustParse(consent.ID)
}

func containsHousehold(items []Household, id string) bool {
	for _, item := range items {
		if item.ID == id {
			return true
		}
	}
	return false
}

func TestDeviceReadTwinUpdateAndIsolation(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	service := NewService(pool, audit.NewRepository(), nil, nil)
	owner := identity.Principal{Issuer: testIssuer, Subject: testOwnerSubject}
	editor := identity.Principal{Issuer: testIssuer, Subject: testEditorSubject}
	outsider := identity.Principal{Issuer: testIssuer, Subject: testOutsideSubject}
	deviceID := "10000000-0000-0000-0000-000000000511"
	var previousReported []byte
	var previousErrorCode, previousErrorSeverity *string
	var previousErrorAt *time.Time
	var previousUpdatedAt time.Time
	if err := pool.QueryRow(ctx, `SELECT reported_config,last_error_code,last_error_severity,last_error_at,updated_at FROM device_twin WHERE device_id=(SELECT id FROM device WHERE public_id=$1::uuid)`, deviceID).Scan(&previousReported, &previousErrorCode, &previousErrorSeverity, &previousErrorAt, &previousUpdatedAt); err != nil {
		t.Fatalf("read prior device report: %v", err)
	}
	failureAt := time.Now().UTC().Truncate(time.Second)
	if _, err := pool.Exec(ctx, `UPDATE device_twin SET reported_config=jsonb_build_object('telemetry',jsonb_build_object('occurred_at',$2::timestamptz,'status',jsonb_build_object('recording_queue',jsonb_build_object('pending_segments',3,'failed_segments',1,'last_failure_code','MEDIA_UPLOAD_TIMEOUT','last_failure_at',$2::timestamptz)))),last_error_code='MEDIA_UPLOAD_TIMEOUT',last_error_severity='error',last_error_at=$2,updated_at=$2 WHERE device_id=(SELECT id FROM device WHERE public_id=$1::uuid)`, deviceID, failureAt); err != nil {
		t.Fatalf("seed device upload queue: %v", err)
	}

	original, err := service.GetDeviceTwin(ctx, owner, deviceID)
	if err != nil {
		t.Fatalf("get original twin: %v", err)
	}
	originalJSON, _ := json.Marshal(original.DesiredConfig)
	defer func() {
		_, _ = pool.Exec(ctx, `
			UPDATE device_twin SET desired_revision=$1,desired_config=$2::jsonb,reported_config=$4::jsonb,
				last_error_code=$5,last_error_severity=$6,last_error_at=$7,updated_at=$8
			WHERE device_id=(SELECT id FROM device WHERE public_id=$3::uuid)`,
			original.DesiredRevision, originalJSON, deviceID, previousReported,
			previousErrorCode, previousErrorSeverity, previousErrorAt, previousUpdatedAt)
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE request_id='saas-integration-device-twin'`)
	}()
	if original.LastErrorCode == nil || *original.LastErrorCode != "MEDIA_UPLOAD_TIMEOUT" || original.LastErrorSeverity == nil || *original.LastErrorSeverity != "error" || original.LastErrorAt == nil {
		t.Fatalf("device twin did not expose the latest safe error: %+v", original)
	}
	telemetry, ok := original.ReportedConfig["telemetry"].(map[string]any)
	if !ok || telemetry["status"] == nil {
		t.Fatalf("device twin did not expose the upload queue report: %+v", original.ReportedConfig)
	}

	devices, err := service.ListDevices(ctx, owner)
	if err != nil || !containsDevice(devices, deviceID) {
		t.Fatalf("list devices: %+v err=%v", devices, err)
	}
	if _, err := service.GetDevice(ctx, outsider, deviceID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider device error=%v", err)
	}
	presence, err := service.GetDevicePresence(ctx, owner, deviceID)
	if err != nil || presence.DeviceID != deviceID || presence.Source != "database" {
		t.Fatalf("database presence: %+v err=%v", presence, err)
	}
	updated, err := service.PatchDeviceTwin(ctx, editor, deviceID, "saas-integration-device-twin", original.DesiredRevision, map[string]any{
		"volume": map[string]any{"speaker": 65},
		"upload": map[string]any{"policy": "wifi_preferred"},
	})
	if err != nil || updated.DesiredRevision != original.DesiredRevision+1 || updated.SyncState != "pending" {
		t.Fatalf("patch device twin: %+v err=%v", updated, err)
	}
	if updated.DesiredConfig["codec"] != original.DesiredConfig["codec"] || updated.DesiredConfig["sample_rate"] != original.DesiredConfig["sample_rate"] || updated.DesiredConfig["heartbeat_interval_seconds"] != original.DesiredConfig["heartbeat_interval_seconds"] {
		t.Fatalf("patch device twin discarded existing device configuration: before=%+v after=%+v", original.DesiredConfig, updated.DesiredConfig)
	}
	if _, err := service.PatchDeviceTwin(ctx, editor, deviceID, "saas-integration-device-stale", original.DesiredRevision, map[string]any{}); !errors.Is(err, ErrVersionConflict) {
		t.Fatalf("stale twin patch error=%v", err)
	}
	power := map[string]any{"power": map[string]any{"mode": "standby"}}
	if _, err := service.PatchDeviceTwin(ctx, editor, deviceID, "saas-integration-power-unsupported", updated.DesiredRevision, power); !errors.Is(err, ErrDevicePowerUnsupported) {
		t.Fatalf("unsupported power: %v", err)
	}
	if _, err := service.PatchDeviceTwin(ctx, outsider, deviceID, "saas-integration-power-foreign", updated.DesiredRevision, power); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("foreign power: %v", err)
	}
	if _, err := pool.Exec(ctx, `UPDATE device_twin SET reported_config=reported_config || '{"power":{"mode":"active","microphone_paused":false,"backlight_percent":80,"microphone_reads":12}}'::jsonb WHERE device_id=(SELECT id FROM device WHERE public_id=$1::uuid)`, deviceID); err != nil {
		t.Fatal(err)
	}
	sleep, err := service.PatchDeviceTwin(ctx, editor, deviceID, "saas-integration-power-supported", updated.DesiredRevision, power)
	if err != nil || sleep.DesiredRevision != updated.DesiredRevision+1 || sleep.ReportedConfig["power"].(map[string]any)["mode"] != "active" {
		t.Fatalf("power must stay pending until actual report: %+v %v", sleep, err)
	}
}

func containsDevice(items []ManagedDevice, id string) bool {
	for _, item := range items {
		if item.ID == id {
			return true
		}
	}
	return false
}

func TestDeviceClaimIsOneTimeAndRetrySafe(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	service := NewService(pool, audit.NewRepository(), nil, nil)
	owner := identity.Principal{Issuer: testIssuer, Subject: testOwnerSubject}
	outsider := identity.Principal{Issuer: testIssuer, Subject: testOutsideSubject}
	householdID := "10000000-0000-0000-0000-000000000211"
	subjectID := "10000000-0000-0000-0000-000000000411"
	claimCode := "LOCAL-CLAIM-0001"
	var bindingID string
	defer func() {
		_, _ = pool.Exec(ctx, `UPDATE device_claim SET used_at=NULL,used_binding_id=NULL WHERE public_id='10000000-0000-0000-0000-000000000911'::uuid`)
		if bindingID != "" {
			_, _ = pool.Exec(ctx, `DELETE FROM device_binding WHERE public_id=$1::uuid`, bindingID)
		}
		_, _ = pool.Exec(ctx, `UPDATE device SET activation_state='CLAIMABLE' WHERE public_id='10000000-0000-0000-0000-000000000512'::uuid AND NOT EXISTS(SELECT 1 FROM device_binding WHERE device_id=device.id AND unbound_at IS NULL)`)
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE request_id='saas-integration-device-claim'`)
	}()
	if _, _, err := service.ClaimDevice(ctx, outsider, claimCode, householdID, subjectID, "saas-integration-outsider-claim"); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider claim error=%v", err)
	}
	claimed, replayed, err := service.ClaimDevice(ctx, owner, claimCode, householdID, subjectID, "saas-integration-device-claim")
	if err != nil || replayed {
		t.Fatalf("claim device: %+v replayed=%v err=%v", claimed, replayed, err)
	}
	bindingID = claimed.BindingID
	again, replayed, err := service.ClaimDevice(ctx, owner, claimCode, householdID, subjectID, "saas-integration-device-claim-replay")
	if err != nil || !replayed || again.BindingID != bindingID {
		t.Fatalf("replay claim: %+v replayed=%v err=%v", again, replayed, err)
	}
	var activeCount int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM device_binding WHERE device_id=(SELECT id FROM device WHERE public_id='10000000-0000-0000-0000-000000000512'::uuid) AND unbound_at IS NULL`).Scan(&activeCount); err != nil {
		t.Fatal(err)
	}
	if activeCount != 1 {
		t.Fatalf("active binding count=%d", activeCount)
	}
}

func TestConversationQueryPaginationHistoryAndIsolation(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	codec, err := pagination.NewCodec([]byte("integration-pagination-signing-key-32-bytes"))
	if err != nil {
		t.Fatal(err)
	}
	service := NewService(pool, audit.NewRepository(), nil, codec)
	owner := identity.Principal{Issuer: testIssuer, Subject: testOwnerSubject}
	outsider := identity.Principal{Issuer: testIssuer, Subject: testOutsideSubject}
	subjectID := "10000000-0000-0000-0000-000000000411"
	contextID, contextPublicID := uuid.NewString(), uuid.NewString()
	conversationIDs := []string{uuid.NewString(), uuid.NewString(), uuid.NewString()}
	conversationPublicIDs := []string{uuid.NewString(), uuid.NewString(), uuid.NewString()}
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM conversation_turn WHERE conversation_id=ANY($1::uuid[])`, conversationIDs)
		_, _ = pool.Exec(ctx, `DELETE FROM conversation_state_history WHERE conversation_id=ANY($1::uuid[])`, conversationIDs)
		_, _ = pool.Exec(ctx, `DELETE FROM rtc_task WHERE conversation_id=ANY($1::uuid[])`, conversationIDs)
		_, _ = pool.Exec(ctx, `DELETE FROM conversation WHERE id=ANY($1::uuid[])`, conversationIDs)
		_, _ = pool.Exec(ctx, `DELETE FROM context_snapshot WHERE id=$1::uuid`, contextID)
	}()
	_, err = pool.Exec(ctx, `
		INSERT INTO context_snapshot(id,public_id,tenant_id,subject_id,profile_version,consent_snapshot_id,memory_versions,prompt_bundle_version,engine_mode,content_digest,sanitized_summary)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001'::uuid,'10000000-0000-0000-0000-000000000401'::uuid,1,'10000000-0000-0000-0000-000000000701'::uuid,'[]'::jsonb,'integration-v1','s2s_hybrid',$3,'合成上下文')`, contextID, contextPublicID, "f"+strings.Repeat("0", 63))
	if err != nil {
		t.Fatalf("insert context snapshot: %v", err)
	}
	base := time.Now().UTC().Add(-time.Hour)
	for index := range conversationIDs {
		status := "COMPLETED"
		failureCode := any(nil)
		rtcStatus := "COMPLETED"
		stopReason := any("user_hangup")
		if index == 2 {
			status = "FAILED"
			failureCode = "SYNTHETIC_FAILURE"
			rtcStatus = "FAILED"
			stopReason = nil
		}
		created := base.Add(time.Duration(index) * time.Minute)
		started := created.Add(time.Second)
		ended := started.Add(30 * time.Second)
		_, err = pool.Exec(ctx, `
			INSERT INTO conversation(id,public_id,tenant_id,household_id,subject_id,device_id,consent_snapshot_id,context_snapshot_id,
				provider_task_ref,engine_mode,retention_mode,status,allocation_expires_at,started_at,ended_at,stop_reason,failure_code,created_at,updated_at)
			VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001'::uuid,'10000000-0000-0000-0000-000000000201'::uuid,
				'10000000-0000-0000-0000-000000000401'::uuid,'10000000-0000-0000-0000-000000000501'::uuid,
				'10000000-0000-0000-0000-000000000701'::uuid,$3::uuid,$4,'s2s_hybrid','transcript_only',$5,$6,$7,$8,$9,$10,$11,$11)
		`, conversationIDs[index], conversationPublicIDs[index], contextID, "integration-task-"+conversationIDs[index], status, created.Add(time.Minute), started, ended, stopReason, failureCode, created)
		if err != nil {
			t.Fatalf("insert conversation %d: %v", index, err)
		}
		_, err = pool.Exec(ctx, `INSERT INTO rtc_task(conversation_id,provider,provider_task_ref,room_id,device_rtc_user_id,agent_rtc_user_id,status) VALUES($1::uuid,'mock',$2,$3,$4,$5,$6)`, conversationIDs[index], "integration-task-"+conversationIDs[index], "room-"+conversationIDs[index], "device-"+conversationIDs[index], "agent-"+conversationIDs[index], rtcStatus)
		if err != nil {
			t.Fatalf("insert rtc task: %v", err)
		}
		_, err = pool.Exec(ctx, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source,occurred_at) VALUES($1::uuid,NULL,'PREPARING','integration','platform',$2),($1::uuid,'PREPARING',$3,'integration','provider',$4)`, conversationIDs[index], created, status, ended)
		if err != nil {
			t.Fatalf("insert history: %v", err)
		}
	}
	for sequence := 1; sequence <= 3; sequence++ {
		start := base.Add(time.Duration(sequence) * time.Second)
		if _, err := pool.Exec(ctx, `INSERT INTO conversation_turn(
			tenant_id,subject_id,conversation_id,turn_key,sequence_no,speaker,start_at,end_at,final_text,interrupted,source_type
		) VALUES (
			'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$1::uuid,$2,$3::bigint,
			CASE WHEN $3::bigint%2=1 THEN 'subject' ELSE 'agent' END,$4,$5,$6,false,'provider'
		)`, conversationIDs[0], fmt.Sprintf("saas-turn-%d", sequence), sequence, start, start.Add(time.Second), fmt.Sprintf("合成话轮%d", sequence)); err != nil {
			t.Fatalf("insert conversation turn: %v", err)
		}
	}
	turns, err := service.ListConversationTurns(ctx, owner, conversationPublicIDs[0], 2, "")
	if err != nil || len(turns.Items) != 2 || turns.NextCursor == "" {
		t.Fatalf("first turn page=%+v err=%v", turns, err)
	}
	remainingTurns, err := service.ListConversationTurns(ctx, owner, conversationPublicIDs[0], 2, turns.NextCursor)
	if err != nil || len(remainingTurns.Items) != 1 || remainingTurns.NextCursor != "" {
		t.Fatalf("second turn page=%+v err=%v", remainingTurns, err)
	}
	if _, err := service.ListConversationTurns(ctx, outsider, conversationPublicIDs[0], 2, ""); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider conversation turns error=%v", err)
	}
	if _, err := service.ListConversationTurns(ctx, owner, conversationPublicIDs[0], 2, turns.NextCursor+"x"); !errors.Is(err, pagination.ErrInvalidCursor) {
		t.Fatalf("tampered turn cursor error=%v", err)
	}
	first, err := service.ListSubjectConversations(ctx, owner, subjectID, ConversationFilter{Limit: 2})
	if err != nil {
		t.Fatalf("first page: %v", err)
	}
	if len(first.Items) != 2 || first.NextCursor == "" {
		t.Fatalf("first page=%+v", first)
	}
	second, err := service.ListSubjectConversations(ctx, owner, subjectID, ConversationFilter{Limit: 2, Cursor: first.NextCursor})
	if err != nil || len(second.Items) != 1 || second.NextCursor != "" {
		t.Fatalf("second page=%+v err=%v", second, err)
	}
	failed, err := service.ListSubjectConversations(ctx, owner, subjectID, ConversationFilter{Limit: 20, Status: "FAILED"})
	if err != nil || len(failed.Items) != 1 {
		t.Fatalf("failed filter=%+v err=%v", failed, err)
	}
	detail, err := service.GetConversation(ctx, owner, conversationPublicIDs[2])
	if err != nil || len(detail.History) != 2 || detail.Snapshots.ContextSnapshotID != contextPublicID {
		t.Fatalf("detail=%+v err=%v", detail, err)
	}
	if _, err := service.GetConversation(ctx, outsider, conversationPublicIDs[2]); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider conversation error=%v", err)
	}
	if _, err := service.ListSubjectConversations(ctx, owner, subjectID, ConversationFilter{Limit: 2, Cursor: first.NextCursor + "x"}); !errors.Is(err, pagination.ErrInvalidCursor) {
		t.Fatalf("tampered cursor error=%v", err)
	}
}

func TestContentReviewAndMemoryLifecycle(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	service := NewService(pool, audit.NewRepository(), nil, nil)
	owner := identity.Principal{Issuer: testIssuer, Subject: testOwnerSubject}
	outsider := identity.Principal{Issuer: testIssuer, Subject: testOutsideSubject}

	contextID, contextPublicID := uuid.NewString(), uuid.NewString()
	conversationID, conversationPublicID := uuid.NewString(), uuid.NewString()
	turnID, turnPublicID := uuid.NewString(), uuid.NewString()
	storyID, storyPublicID := uuid.NewString(), uuid.NewString()
	revisionID := uuid.NewString()
	var factPublicID string
	var memoryID string
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE request_id LIKE 'content-integration-%'`)
		_, _ = pool.Exec(ctx, `DELETE FROM content_review_signal WHERE resource_public_id=$1::uuid OR resource_public_id=$2::uuid`, storyPublicID, factPublicID)
		if memoryID != "" {
			_, _ = pool.Exec(ctx, `DELETE FROM outbox_event WHERE aggregate_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, memoryID)
			_, _ = pool.Exec(ctx, `DELETE FROM memory_index_entry WHERE memory_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, memoryID)
			_, _ = pool.Exec(ctx, `UPDATE curated_memory SET current_version_id=NULL WHERE public_id=$1::uuid`, memoryID)
			_, _ = pool.Exec(ctx, `DELETE FROM memory_version WHERE memory_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, memoryID)
			_, _ = pool.Exec(ctx, `DELETE FROM curated_memory WHERE public_id=$1::uuid`, memoryID)
		}
		_, _ = pool.Exec(ctx, `DELETE FROM fact_candidate_revision WHERE fact_candidate_id IN (SELECT id FROM fact_candidate WHERE story_id=$1::uuid)`, storyID)
		_, _ = pool.Exec(ctx, `DELETE FROM fact_candidate WHERE story_id=$1::uuid`, storyID)
		_, _ = pool.Exec(ctx, `DELETE FROM story_source WHERE story_revision_id IN (SELECT id FROM story_revision WHERE story_id=$1::uuid)`, storyID)
		_, _ = pool.Exec(ctx, `UPDATE story_draft SET current_revision_id=NULL WHERE id=$1::uuid`, storyID)
		_, _ = pool.Exec(ctx, `DELETE FROM story_revision WHERE story_id=$1::uuid`, storyID)
		_, _ = pool.Exec(ctx, `DELETE FROM story_draft WHERE id=$1::uuid`, storyID)
		_, _ = pool.Exec(ctx, `DELETE FROM conversation_summary WHERE conversation_id=$1::uuid`, conversationID)
		_, _ = pool.Exec(ctx, `DELETE FROM conversation_turn WHERE conversation_id=$1::uuid`, conversationID)
		_, _ = pool.Exec(ctx, `DELETE FROM conversation WHERE id=$1::uuid`, conversationID)
		_, _ = pool.Exec(ctx, `DELETE FROM context_snapshot WHERE id=$1::uuid`, contextID)
	}()

	if _, err := pool.Exec(ctx, `
		INSERT INTO context_snapshot(id,public_id,tenant_id,subject_id,profile_version,consent_snapshot_id,memory_versions,prompt_bundle_version,engine_mode,content_digest,sanitized_summary)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',1,
			'10000000-0000-0000-0000-000000000701','[]','content-integration-v1','s2s_hybrid',$3,'合成内容审核上下文')`,
		contextID, contextPublicID, "e"+strings.Repeat("0", 63)); err != nil {
		t.Fatalf("insert context snapshot: %v", err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO conversation(id,public_id,tenant_id,household_id,subject_id,device_id,consent_snapshot_id,context_snapshot_id,
			provider_task_ref,engine_mode,retention_mode,status,allocation_expires_at,started_at,ended_at,stop_reason)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000201',
			'10000000-0000-0000-0000-000000000401','10000000-0000-0000-0000-000000000501',
			'10000000-0000-0000-0000-000000000701',$3::uuid,$4,'s2s_hybrid','transcript_only','COMPLETED',now(),now()-interval '1 minute',now(),'user_hangup')`,
		conversationID, conversationPublicID, contextID, "content-task-"+conversationID); err != nil {
		t.Fatalf("insert conversation: %v", err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO conversation_turn(id,public_id,tenant_id,subject_id,conversation_id,turn_key,sequence_no,speaker,start_at,end_at,final_text,source_type)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$3::uuid,$4,1,'subject',now()-interval '50 seconds',now()-interval '45 seconds',$5,'provider')`,
		turnID, turnPublicID, conversationID, "content-turn-"+turnID, "我年轻时在合成城市做过教师。"); err != nil {
		t.Fatalf("insert turn: %v", err)
	}
	sourceIDs, _ := json.Marshal([]string{turnPublicID})
	if _, err := pool.Exec(ctx, `
		INSERT INTO conversation_summary(tenant_id,subject_id,conversation_id,summary_type,content,source_turn_ids,model_ref,prompt_version,generation_version)
		VALUES('10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$1::uuid,'conversation_summary',$2,$3::jsonb,'mock-content-v1','prompt-v1',1)`,
		conversationID, "受试者回忆了教师经历。", sourceIDs); err != nil {
		t.Fatalf("insert summary: %v", err)
	}
	summaries, err := service.ListConversationContentSummaries(ctx, owner, conversationPublicID)
	if err != nil || len(summaries) != 1 || len(summaries[0].SourceTurnIDs) != 1 || summaries[0].SourceTurnIDs[0] != turnPublicID {
		t.Fatalf("summaries=%+v err=%v", summaries, err)
	}
	if _, err := service.ListConversationContentSummaries(ctx, outsider, conversationPublicID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider summary error=%v", err)
	}

	if _, err := pool.Exec(ctx, `
		INSERT INTO story_draft(id,public_id,tenant_id,subject_id,conversation_id,title,status,confidence,sensitivity,generation_version)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$3::uuid,$4,'CANDIDATE',0.91,'normal',1)`,
		storyID, storyPublicID, conversationID, "教师生涯"); err != nil {
		t.Fatalf("insert story: %v", err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO story_revision(id,story_id,revision_no,content,author_type,model_ref,prompt_version)
		VALUES($1::uuid,$2::uuid,1,$3,'model','mock-content-v1','prompt-v1')`, revisionID, storyID, "受试者年轻时在合成城市从事教师工作。"); err != nil {
		t.Fatalf("insert story revision: %v", err)
	}
	if _, err := pool.Exec(ctx, `UPDATE story_draft SET current_revision_id=$1::uuid WHERE id=$2::uuid`, revisionID, storyID); err != nil {
		t.Fatalf("select current story revision: %v", err)
	}
	if _, err := pool.Exec(ctx, `
		INSERT INTO story_source(story_revision_id,conversation_turn_id,source_start_offset,source_end_offset,source_role)
		VALUES($1::uuid,$2::uuid,0,15,'subject')`, revisionID, turnID); err != nil {
		t.Fatalf("insert story source: %v", err)
	}
	if err := pool.QueryRow(ctx, `
		INSERT INTO fact_candidate(tenant_id,subject_id,story_id,fact_type,content,statement_basis,source_turn_ids,status,generation_version)
		VALUES('10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000401',$1::uuid,'occupation',$2,'subject_statement',$3::jsonb,'CANDIDATE',1)
		RETURNING public_id::text`, storyID, "曾从事教师工作", sourceIDs).Scan(&factPublicID); err != nil {
		t.Fatalf("insert fact candidate: %v", err)
	}
	stories, err := service.ListSubjectStories(ctx, owner, "10000000-0000-0000-0000-000000000411", "CANDIDATE")
	if err != nil || len(stories) != 1 || stories[0].ID != storyPublicID {
		t.Fatalf("stories=%+v err=%v", stories, err)
	}
	story, err := service.GetStory(ctx, owner, storyPublicID)
	if err != nil || len(story.Sources) != 1 || story.Sources[0].TurnID != turnPublicID || len(story.Facts) != 1 {
		t.Fatalf("story=%+v err=%v", story, err)
	}
	correctedFact := "曾在合成城市从事教师工作"
	confirmedFactStatus := "CONFIRMED"
	factErrorType := "PLACE_ERROR"
	fact, err := service.UpdateFactCandidate(ctx, owner, storyPublicID, factPublicID, "content-integration-fact-confirm", 1, UpdateFactCandidateInput{
		Content: &correctedFact, Status: &confirmedFactStatus, ErrorType: &factErrorType, Reason: "逐字核对长者原话",
	})
	if err != nil || fact.Content != correctedFact || fact.Status != "CONFIRMED" || fact.Version != 2 {
		t.Fatalf("confirmed fact=%+v err=%v", fact, err)
	}
	var factRevisionCount, factSignalCount int
	if err := pool.QueryRow(ctx, `SELECT (SELECT count(*) FROM fact_candidate_revision WHERE fact_candidate_id=(SELECT id FROM fact_candidate WHERE public_id=$1::uuid)),(SELECT count(*) FROM content_review_signal WHERE resource_type='fact' AND resource_public_id=$1::uuid)`, factPublicID).Scan(&factRevisionCount, &factSignalCount); err != nil || factRevisionCount != 2 || factSignalCount != 1 {
		t.Fatalf("fact revisions=%d signals=%d err=%v", factRevisionCount, factSignalCount, err)
	}
	if _, err := service.UpdateFactCandidate(ctx, owner, storyPublicID, factPublicID, "content-integration-fact-stale", 1, UpdateFactCandidateInput{Reason: "旧页面提交"}); !errors.Is(err, ErrVersionConflict) {
		t.Fatalf("stale fact review error=%v", err)
	}
	if _, err := service.UpdateFactCandidate(ctx, outsider, storyPublicID, factPublicID, "content-integration-fact-outsider", 2, UpdateFactCandidateInput{Reason: "越权提交"}); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider fact review error=%v", err)
	}
	revisedContent := "受试者年轻时在合成城市任职教师，并明确这是本人陈述。"
	story, err = service.ReviseStory(ctx, owner, storyPublicID, revisedContent, "人工核对原话", "content-integration-revise", 1, "PLACE_ERROR")
	if err != nil || story.Status != "IN_REVIEW" || story.Version != 2 || story.RevisionNo != 2 || len(story.Sources) != 1 {
		t.Fatalf("revised story=%+v err=%v", story, err)
	}
	if _, err := service.ReviseStory(ctx, owner, storyPublicID, "过期修改", "旧版本", "content-integration-stale", 1, ""); !errors.Is(err, ErrVersionConflict) {
		t.Fatalf("stale story revision error=%v", err)
	}
	story, memory, err := service.ConfirmStory(ctx, owner, storyPublicID, "来源与本人同意均已核验", "content-integration-confirm", 2)
	if err != nil || story.Status != "CONFIRMED" || story.Version != 3 || memory.Status != "ACTIVE" || memory.Content != revisedContent || memory.IndexStatus != "PENDING" {
		t.Fatalf("confirmed story=%+v memory=%+v err=%v", story, memory, err)
	}
	memoryID = memory.ID
	if _, _, err := service.ConfirmStory(ctx, owner, storyPublicID, "重复确认", "content-integration-confirm-again", 3); !errors.Is(err, ErrReviewStateConflict) {
		t.Fatalf("duplicate story confirmation error=%v", err)
	}
	memories, err := service.ListSubjectMemories(ctx, owner, "10000000-0000-0000-0000-000000000411", "合成城市")
	if err != nil || len(memories) != 1 || memories[0].ID != memory.ID {
		t.Fatalf("memories=%+v err=%v", memories, err)
	}
	if _, err := service.GetMemory(ctx, outsider, memory.ID); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider memory error=%v", err)
	}
	updatedContent := revisedContent + "（家庭成员复核）"
	memory, err = service.UpdateMemory(ctx, owner, memory.ID, "content-integration-memory-update", 1, UpdateMemoryInput{Content: &updatedContent, Reason: "补充复核结果"})
	if err != nil || memory.Version != 2 || memory.VersionNo != 2 || memory.Content != updatedContent || memory.IndexStatus != "PENDING" {
		t.Fatalf("updated memory=%+v err=%v", memory, err)
	}
	if _, err := service.UpdateMemory(ctx, owner, memory.ID, "content-integration-memory-noop", 2, UpdateMemoryInput{Content: &updatedContent, Reason: "无变化"}); !errors.Is(err, ErrReviewStateConflict) {
		t.Fatalf("memory no-op error=%v", err)
	}
	disabled := "DISABLED"
	memory, err = service.UpdateMemory(ctx, owner, memory.ID, "content-integration-memory-disable", 2, UpdateMemoryInput{Status: &disabled, Reason: "暂停个性化使用"})
	if err != nil || memory.Status != "DISABLED" || memory.Version != 3 || memory.IndexStatus != "STALE" {
		t.Fatalf("disabled memory=%+v err=%v", memory, err)
	}
	memory, err = service.RequestMemoryDeletion(ctx, owner, memory.ID, "用户要求删除", "content-integration-memory-delete", 3)
	if err != nil || memory.Status != "PENDING_DELETE" || memory.Version != 4 || memory.IndexStatus != "STALE" {
		t.Fatalf("pending deletion memory=%+v err=%v", memory, err)
	}
	active := "ACTIVE"
	if _, err := service.UpdateMemory(ctx, owner, memory.ID, "content-integration-memory-reactivate", 4, UpdateMemoryInput{Status: &active, Reason: "尝试恢复删除中的记忆"}); !errors.Is(err, ErrReviewStateConflict) {
		t.Fatalf("pending deletion memory reactivation error=%v", err)
	}
	replayed, err := service.RequestMemoryDeletion(ctx, owner, memory.ID, "重复请求", "content-integration-memory-delete-replay", 4)
	if err != nil || replayed.Version != memory.Version || replayed.Status != "PENDING_DELETE" {
		t.Fatalf("replayed memory deletion=%+v err=%v", replayed, err)
	}
}

func TestHouseholdMemberInvitationRoleRemovalAndOwnership(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	service := NewService(pool, audit.NewRepository(), nil, nil)
	owner := identity.Principal{Issuer: testIssuer, Subject: testOwnerSubject}
	editor := identity.Principal{Issuer: testIssuer, Subject: testEditorSubject}
	outsider := identity.Principal{Issuer: testIssuer, Subject: testOutsideSubject}

	household, err := service.CreateHousehold(ctx, owner, testTenantPublicID, "成员管理集成家庭", "member-integration-household")
	if err != nil {
		t.Fatalf("create household: %v", err)
	}
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM audit_event WHERE request_id LIKE 'member-integration-%'`)
		_, _ = pool.Exec(ctx, `DELETE FROM member_invitation WHERE household_id=(SELECT id FROM household WHERE public_id=$1::uuid)`, household.ID)
		_, _ = pool.Exec(ctx, `DELETE FROM household_member WHERE household_id=(SELECT id FROM household WHERE public_id=$1::uuid)`, household.ID)
		_, _ = pool.Exec(ctx, `DELETE FROM household WHERE public_id=$1::uuid`, household.ID)
		_, _ = pool.Exec(ctx, `DELETE FROM user_account WHERE tenant_id=(SELECT id FROM tenant WHERE public_id=$1::uuid) AND oidc_issuer=$2 AND oidc_subject=$3`, testTenantPublicID, testIssuer, testOutsideSubject)
	}()

	if _, err := service.CreateMemberInvitation(ctx, outsider, household.ID, "禁止", identity.RoleViewer, time.Now().Add(time.Hour), "member-integration-outsider"); !errors.Is(err, ErrAccessDenied) {
		t.Fatalf("outsider invitation error=%v", err)
	}
	crossTenantInvitation, err := service.CreateMemberInvitation(ctx, owner, household.ID, "跨客户组织受邀账号", identity.RoleViewer, time.Now().Add(time.Hour), "member-integration-cross-tenant-invite")
	if err != nil {
		t.Fatal(err)
	}
	crossTenantMember, _, err := service.AcceptMemberInvitation(ctx, outsider, crossTenantInvitation.Token, "member-integration-cross-tenant-accept")
	if err != nil || crossTenantMember.Role != identity.RoleViewer {
		t.Fatalf("cross-tenant account invitation: %+v err=%v", crossTenantMember, err)
	}
	invitation, err := service.CreateMemberInvitation(ctx, owner, household.ID, "editor@example.invalid", identity.RoleEditor, time.Now().Add(time.Hour), "member-integration-invite")
	if err != nil || invitation.Token == "" || invitation.Status != "pending" {
		t.Fatalf("create invitation: %+v err=%v", invitation, err)
	}
	var storedToken string
	if err := pool.QueryRow(ctx, `SELECT token_digest FROM member_invitation WHERE public_id=$1::uuid`, invitation.ID).Scan(&storedToken); err != nil {
		t.Fatal(err)
	}
	if storedToken == invitation.Token || len(storedToken) != 64 {
		t.Fatalf("invitation token was not stored as SHA-256 digest: %q", storedToken)
	}
	invitations, err := service.ListMemberInvitations(ctx, owner, household.ID)
	if err != nil || len(invitations) != 2 {
		t.Fatalf("member invitations=%+v err=%v", invitations, err)
	}
	member, replayed, err := service.AcceptMemberInvitation(ctx, editor, invitation.Token, "member-integration-accept")
	if err != nil || replayed || member.Role != identity.RoleEditor {
		t.Fatalf("accept invitation: %+v replayed=%v err=%v", member, replayed, err)
	}
	if _, _, err := service.AcceptMemberInvitation(ctx, editor, invitation.Token, "member-integration-reuse"); !errors.Is(err, ErrInvitationInvalid) {
		t.Fatalf("reused invitation error=%v", err)
	}
	var failedAcceptanceAudits int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE request_id='member-integration-reuse' AND action='member.invitation.accept' AND result='failure'`).Scan(&failedAcceptanceAudits); err != nil || failedAcceptanceAudits != 1 {
		t.Fatalf("failed invitation audit count=%d err=%v", failedAcceptanceAudits, err)
	}
	members, err := service.ListHouseholdMembers(ctx, editor, household.ID)
	if err != nil || len(members) != 3 {
		t.Fatalf("list members: %+v err=%v", members, err)
	}
	var ownerMember HouseholdMember
	for _, item := range members {
		if item.IsOwner {
			ownerMember = item
		}
	}
	if _, err := service.RemoveHouseholdMember(ctx, owner, household.ID, ownerMember.ID, ownerMember.Version, "member-integration-owner-remove"); !errors.Is(err, ErrOwnerProtected) {
		t.Fatalf("owner removal error=%v", err)
	}
	member, err = service.PatchHouseholdMemberRole(ctx, owner, household.ID, member.ID, identity.RoleAdmin, member.Version, "member-integration-role")
	if err != nil || member.Role != identity.RoleAdmin || member.Version != 2 {
		t.Fatalf("patch member role: %+v err=%v", member, err)
	}
	if _, err := service.PatchHouseholdMemberRole(ctx, owner, household.ID, member.ID, identity.RoleViewer, 1, "member-integration-stale"); !errors.Is(err, ErrVersionConflict) {
		t.Fatalf("stale role patch error=%v", err)
	}
	household, err = service.TransferHouseholdOwnership(ctx, owner, household.ID, member.ID, household.Version, "member-integration-transfer")
	if err != nil || household.Version != 2 || household.Role != identity.RoleAdmin {
		t.Fatalf("transfer ownership: %+v err=%v", household, err)
	}
	members, err = service.ListHouseholdMembers(ctx, editor, household.ID)
	if err != nil {
		t.Fatal(err)
	}
	for _, item := range members {
		if item.UserID == ownerMember.UserID {
			ownerMember = item
		}
	}
	household, err = service.TransferHouseholdOwnership(ctx, editor, household.ID, ownerMember.ID, household.Version, "member-integration-transfer-back")
	if err != nil || household.Version != 3 {
		t.Fatalf("transfer ownership back: %+v err=%v", household, err)
	}
	members, err = service.ListHouseholdMembers(ctx, owner, household.ID)
	if err != nil {
		t.Fatal(err)
	}
	for _, item := range members {
		if item.UserID == member.UserID {
			member = item
		}
	}
	removed, err := service.RemoveHouseholdMember(ctx, owner, household.ID, member.ID, member.Version, "member-integration-remove")
	if err != nil || removed.Status != "removed" {
		t.Fatalf("remove member: %+v err=%v", removed, err)
	}

	revokable, err := service.CreateMemberInvitation(ctx, owner, household.ID, "待撤销", identity.RoleViewer, time.Now().Add(time.Hour), "member-integration-revokable")
	if err != nil {
		t.Fatal(err)
	}
	revoked, err := service.RevokeMemberInvitation(ctx, owner, household.ID, revokable.ID, "member-integration-revoke")
	if err != nil || revoked.Status != "revoked" {
		t.Fatalf("revoke invitation: %+v err=%v", revoked, err)
	}
	if _, _, err := service.AcceptMemberInvitation(ctx, editor, revokable.Token, "member-integration-revoked-accept"); !errors.Is(err, ErrInvitationInvalid) {
		t.Fatalf("revoked invitation error=%v", err)
	}
	invitations, err = service.ListMemberInvitations(ctx, owner, household.ID)
	if err != nil || len(invitations) != 3 {
		t.Fatalf("final member invitations=%+v err=%v", invitations, err)
	}
}