package orchestration

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

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/embeddingprovider"
	"data-terminal-cloud/internal/idempotency"
	platformidentity "data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/opsconfig"
	"data-terminal-cloud/internal/outbox"
	"data-terminal-cloud/internal/ratelimit"
	"data-terminal-cloud/internal/rtc"
	"data-terminal-cloud/internal/voiceprovider"

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

const (
	testDeviceID              = "90000000-0000-0000-0000-000000000501"
	testDevicePublicID        = "90000000-0000-0000-0000-000000000511"
	testSubjectID             = "90000000-0000-0000-0000-000000000401"
	testSubjectPublicID       = "90000000-0000-0000-0000-000000000411"
	testPriceBookID           = "90000000-0000-0000-0000-000000000601"
	testPersonaConfigID       = "90000000-0000-0000-0000-000000000701"
	testPersonaConfigPublicID = "90000000-0000-0000-0000-000000000711"
	testVoiceConfigID         = "90000000-0000-0000-0000-000000000702"
	testVoiceConfigPublicID   = "90000000-0000-0000-0000-000000000712"
	testModelConfigID         = "90000000-0000-0000-0000-000000000703"
	testModelConfigPublicID   = "90000000-0000-0000-0000-000000000713"
	testRoutingConfigID       = "90000000-0000-0000-0000-000000000704"
	testRoutingConfigPublicID = "90000000-0000-0000-0000-000000000714"
	testConsentSnapshotID     = "90000000-0000-0000-0000-000000000801"
	testPriorContextID        = "90000000-0000-0000-0000-000000000802"
	testPriorConversationID   = "90000000-0000-0000-0000-000000000803"
	testBundleConfigKey       = "orchestration-test-agent-bundle"
	testBundleCreateKey       = "orchestration-agent-bundle-create"
	testBundleEvaluationKey   = "orchestration-agent-bundle-evaluation"
)

type testLeaseCache struct{}

func (testLeaseCache) Refresh(context.Context, LeaseRecord) error           { return nil }
func (testLeaseCache) End(context.Context, string, string, time.Time) error { return nil }

type failingTokenIssuer struct{}

func (failingTokenIssuer) Issue(string, string, time.Time) (string, error) {
	return "", errors.New("simulated token signing failure")
}

type countingEmbedder struct{ calls atomic.Int32 }

func (e *countingEmbedder) Embed(ctx context.Context, content string) (embeddingprovider.Result, error) {
	e.calls.Add(1)
	return (embeddingprovider.Local{}).Embed(ctx, content)
}

type testVoiceProvider struct {
	startRequest voiceprovider.StartRequest
	stopRequest  voiceprovider.StopRequest
	startStatus  string
	stopStatus   string
	queryStatus  string
	startErr     error
	stopErr      error
	queryErr     error
}

func (p *testVoiceProvider) Start(_ context.Context, request voiceprovider.StartRequest) (voiceprovider.StartResponse, error) {
	p.startRequest = request
	if p.startErr != nil {
		return voiceprovider.StartResponse{}, p.startErr
	}
	status := p.startStatus
	if status == "" {
		status = "STARTING"
	}
	return voiceprovider.StartResponse{TaskRef: request.TaskRef, Status: status}, nil
}

func (p *testVoiceProvider) Query(_ context.Context, request voiceprovider.QueryRequest) (voiceprovider.QueryResponse, error) {
	if p.queryErr != nil {
		return voiceprovider.QueryResponse{}, p.queryErr
	}
	status := p.queryStatus
	if status == "" {
		status = "STARTING"
	}
	return voiceprovider.QueryResponse{TaskRef: request.TaskRef, Status: status}, nil
}

func (p *testVoiceProvider) Stop(_ context.Context, request voiceprovider.StopRequest) (voiceprovider.StopResponse, error) {
	p.stopRequest = request
	if p.stopErr != nil {
		return voiceprovider.StopResponse{}, p.stopErr
	}
	status := p.stopStatus
	if status == "" {
		status = "STOPPING"
	}
	return voiceprovider.StopResponse{TaskRef: request.TaskRef, Status: status}, nil
}

type allowProviderStart struct{}

func (allowProviderStart) Allow(context.Context, string, string) (ratelimit.Result, error) {
	return ratelimit.Result{Allowed: true}, nil
}

func TestServiceIdempotentlyCreatesConversationWithoutPersistingToken(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := database.Open(ctx, databaseURL, "orchestration-service-test")
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	cleanupOrchestrationFixture(t, ctx, pool)
	defer cleanupOrchestrationFixture(t, context.Background(), pool)
	fixtureStatements := []struct {
		query string
		args  []any
	}{
		{`INSERT INTO subject_profile (id, public_id, tenant_id, household_id, preferred_name, profile_data, conversation_preferences, status)
		  VALUES ($1::uuid, '90000000-0000-0000-0000-000000000411', '10000000-0000-0000-0000-000000000001', '10000000-0000-0000-0000-000000000201', '服务测试长者',
		  '{"background":"年轻时从事铁路工作","interests":["老照片","家乡变化"],"internal_note":"不得进入语音上下文"}',
		  '{"conversation_goal":"聊聊第一次参加工作的经历","pace":"slow","preferred_topics":["工作经历"],"avoided_topics":["收入"]}', 'active')`, []any{testSubjectID}},
		{`INSERT INTO device (id, public_id, tenant_id, serial_digest, model, status, credential_ref)
		  VALUES ($1::uuid, $2::uuid, '10000000-0000-0000-0000-000000000001', repeat('9',64), 'service-test', 'registered', 'test://service')`, []any{testDeviceID, testDevicePublicID}},
		{`INSERT INTO device_binding (tenant_id, device_id, household_id, subject_id, reason)
		  VALUES ('10000000-0000-0000-0000-000000000001', $1::uuid, '10000000-0000-0000-0000-000000000201', $2::uuid, 'integration test')`, []any{testDeviceID, testSubjectID}},
		{`INSERT INTO device_twin (device_id, tenant_id) VALUES ($1::uuid, '10000000-0000-0000-0000-000000000001')`, []any{testDeviceID}},
		{`INSERT INTO consent_snapshot (id,tenant_id,subject_id,snapshot_version,effective_consents)
		  VALUES ($1::uuid,'10000000-0000-0000-0000-000000000001',$2::uuid,1,'{"voice_chat":true,"transcript":true,"audio_recording":false,"story_extraction":true,"personalized_memory":true}'::jsonb)`, []any{testConsentSnapshotID, testSubjectID}},
		{`INSERT INTO context_snapshot (id,tenant_id,subject_id,profile_version,consent_snapshot_id,memory_versions,prompt_bundle_version,engine_mode,content_digest,sanitized_summary)
		  VALUES ($1::uuid,'10000000-0000-0000-0000-000000000001',$2::uuid,1,$3::uuid,'[]'::jsonb,'prior-test','cascade',repeat('b',64),'此前通话上下文')`, []any{testPriorContextID, testSubjectID, testConsentSnapshotID}},
		{`INSERT INTO conversation (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,created_at,updated_at)
		  VALUES ($1::uuid,'10000000-0000-0000-0000-000000000001','10000000-0000-0000-0000-000000000201',$2::uuid,$3::uuid,$4::uuid,$5::uuid,'prior-context-task','cascade','transcript_only','COMPLETED','2026-08-23T10:30:00Z','2026-08-23T10:00:00Z','2026-08-23T10:20:00Z','2026-08-23T10:00:00Z','2026-08-23T10:20:00Z')`, []any{testPriorConversationID, testSubjectID, testDeviceID, testConsentSnapshotID, testPriorContextID}},
		{`INSERT INTO conversation_turn (tenant_id,subject_id,conversation_id,turn_key,sequence_no,speaker,start_at,end_at,final_text,source_type)
		  VALUES ('10000000-0000-0000-0000-000000000001',$1::uuid,$2::uuid,'prior-subject-1',1,'subject','2026-08-23T10:01:00Z','2026-08-23T10:01:05Z','我注册了一个个体工商户。','provider')`, []any{testSubjectID, testPriorConversationID}},
		{`INSERT INTO conversation_summary (tenant_id,subject_id,conversation_id,summary_type,content,source_turn_ids,model_ref,prompt_version,generation_version,status,created_at)
		  VALUES ('10000000-0000-0000-0000-000000000001',$1::uuid,$2::uuid,'conversation_summary','上次聊到刚注册了一个个体工商户，准备开始新的工作。','["prior-subject-1"]'::jsonb,'mock-content-v1','prompt-v1',1,'READY','2026-08-23T10:21:00Z')`, []any{testSubjectID, testPriorConversationID}},
		{`INSERT INTO consent_record (tenant_id,subject_id,consent_type,policy_version,notice_digest,actor_user_id)
		  VALUES ('10000000-0000-0000-0000-000000000001',$1::uuid,'personalized_memory','orchestration-service-test',repeat('a',64),'10000000-0000-0000-0000-000000000101')`, []any{testSubjectID}},
		{`INSERT INTO runtime_config_version (
			id,public_id,tenant_id,config_key,display_name,config_kind,version_no,status,payload,change_summary,
			created_by,submitted_by,submitted_at,approved_by,approved_at,published_by,published_at
		  ) VALUES (
			$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','orchestration-test-persona','编排测试人设','PERSONA',1,'PUBLISHED',
			'{"system_prompt":"你是已审核发布的陪伴助手；一次只问一个问题；不要编造长者事实。","tone":"warm"}'::jsonb,'编排集成测试',
			'10000000-0000-0000-0000-000000000102','10000000-0000-0000-0000-000000000102','2026-08-23T11:00:00Z',
			'10000000-0000-0000-0000-000000000102','2026-08-23T11:01:00Z','10000000-0000-0000-0000-000000000102','2026-08-23T11:02:00Z'
		  )`, []any{testPersonaConfigID, testPersonaConfigPublicID}},
		{`INSERT INTO runtime_config_version (
			id,public_id,tenant_id,config_key,display_name,config_kind,version_no,status,payload,change_summary,
			created_by,submitted_by,submitted_at,approved_by,approved_at,published_by,published_at
		  ) VALUES (
			$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','orchestration-test-voice','编排测试音色','VOICE',1,'PUBLISHED',
			'{"provider":"mock","voice_resource_ref":"voice://orchestration/approved"}'::jsonb,'编排集成测试',
			'10000000-0000-0000-0000-000000000102','10000000-0000-0000-0000-000000000102','2026-08-23T11:00:00Z',
			'10000000-0000-0000-0000-000000000102','2026-08-23T11:01:00Z','10000000-0000-0000-0000-000000000102','2026-08-23T11:02:00Z'
		  )`, []any{testVoiceConfigID, testVoiceConfigPublicID}},
		{`INSERT INTO runtime_config_version (
			id,public_id,tenant_id,config_key,display_name,config_kind,version_no,status,payload,change_summary,
			created_by,submitted_by,submitted_at,approved_by,approved_at,published_by,published_at
		  ) VALUES (
			$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','orchestration-test-model','编排测试模型','MODEL',1,'PUBLISHED',
			'{"provider":"mock","model_id":"model://orchestration/approved"}'::jsonb,'编排集成测试',
			'10000000-0000-0000-0000-000000000102','10000000-0000-0000-0000-000000000102','2026-08-23T11:00:00Z',
			'10000000-0000-0000-0000-000000000102','2026-08-23T11:01:00Z','10000000-0000-0000-0000-000000000102','2026-08-23T11:02:00Z'
		  )`, []any{testModelConfigID, testModelConfigPublicID}},
		{`INSERT INTO runtime_config_version (
			id,public_id,tenant_id,config_key,display_name,config_kind,version_no,status,payload,change_summary,
			created_by,submitted_by,submitted_at,approved_by,approved_at,published_by,published_at
		  ) VALUES (
			$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001','orchestration-test-routing','编排测试路由','ROUTING',1,'PUBLISHED',
			'{"default_engine_mode":"cascade","routes":[{"priority":1,"provider":"mock"}]}'::jsonb,'编排集成测试',
			'10000000-0000-0000-0000-000000000102','10000000-0000-0000-0000-000000000102','2026-08-23T11:00:00Z',
			'10000000-0000-0000-0000-000000000102','2026-08-23T11:01:00Z','10000000-0000-0000-0000-000000000102','2026-08-23T11:02:00Z'
		  )`, []any{testRoutingConfigID, testRoutingConfigPublicID}},
		{`INSERT INTO billing_price_book (
			id,public_id,tenant_id,provider,currency,version_no,status,source_ref,change_summary,
			effective_from,created_by,published_by,published_at
		  ) VALUES (
			$1::uuid,'90000000-0000-0000-0000-000000000611','10000000-0000-0000-0000-000000000001',
			'mock','CNY',9001,'PUBLISHED','integration-test','provider usage integration test',
			'2026-01-01T00:00:00Z','10000000-0000-0000-0000-000000000102','10000000-0000-0000-0000-000000000102','2026-01-01T00:00:00Z'
		  )`, []any{testPriceBookID}},
		{`INSERT INTO billing_price_rate (price_book_id,usage_category,metric_code,unit,unit_price_micros)
		  VALUES ($1::uuid,'LLM','input_tokens','token',2)`, []any{testPriceBookID}},
	}
	for _, statement := range fixtureStatements {
		if _, err := pool.Exec(ctx, statement.query, statement.args...); err != nil {
			t.Fatal(err)
		}
	}
	configService, err := opsconfig.NewService(pool, audit.NewRepository())
	if err != nil {
		t.Fatal(err)
	}
	configAdmin := platformidentity.Principal{Issuer: "http://localhost:58180/realms/voice-local", Subject: "00000000-0000-0000-0000-000000000104"}
	configApprover := platformidentity.Principal{Issuer: configAdmin.Issuer, Subject: "00000000-0000-0000-0000-000000000105"}
	bundle, created, err := configService.Create(ctx, configAdmin, testBundleCreateKey, "orchestration-agent-bundle-create", opsconfig.CreateInput{
		TenantID: "10000000-0000-0000-0000-000000000010", ConfigKey: testBundleConfigKey,
		DisplayName: "编排测试智能体包", ConfigKind: "AGENT_BUNDLE", ChangeSummary: "验证运营发布的自定义版本族进入通话运行时",
		Payload: map[string]any{
			"persona_version_id": testPersonaConfigPublicID, "voice_version_id": testVoiceConfigPublicID,
			"model_version_id": testModelConfigPublicID, "routing_version_id": testRoutingConfigPublicID,
		}, SecretRefs: map[string]string{},
	})
	if err != nil || !created {
		t.Fatalf("create Agent bundle=%+v created=%v err=%v", bundle, created, err)
	}
	bundle, err = configService.Submit(ctx, configAdmin, "10000000-0000-0000-0000-000000000010", bundle.ID, "orchestration-agent-bundle-submit", bundle.RowVersion)
	if err != nil {
		t.Fatal(err)
	}
	qualityMetrics := map[string]any{
		"report_schema": "agent-eval-report-v1", "suite_sha256": "b5cd3b798a01a48ddf0081f2fcc2a917f93681a3ee2fa55bbf4190466de3bef8",
		"agent_version": "orchestration-integration-v1", "model_config_ref": "runtime-config-version:" + testModelConfigPublicID,
		"config_version_id": bundle.ID, "environment": "mock", "executed_at": "2026-08-24T10:00:00Z",
		"cases_total": 10, "cases_passed": 10, "failed_check_count": 0, "conversation_success_rate": 1.0,
		"first_response_p95_ms": 1200, "interruption_recovery_rate": 1.0, "average_response_chars": 36.5,
		"salutation_accuracy": 1.0, "memory_accuracy": 1.0, "fabricated_facts": 0,
		"story_source_completeness": 1.0, "sensitive_information_handling_rate": 1.0,
	}
	if _, created, err = configService.CreateEvaluation(ctx, configAdmin, "10000000-0000-0000-0000-000000000010", bundle.ID,
		testBundleEvaluationKey, "orchestration-agent-bundle-evaluation", opsconfig.EvidenceTypeAgentQualityV1,
		"orchestration-agent-quality-v1", "PASSED", "artifact://orchestration-agent-quality.json", qualityMetrics); err != nil || !created {
		t.Fatalf("register Agent quality evidence created=%v err=%v", created, err)
	}
	bundle, err = configService.Publish(ctx, configApprover, "10000000-0000-0000-0000-000000000010", bundle.ID,
		"orchestration-agent-bundle-publish", opsconfig.PublishInput{Reason: "标准质量证据通过后灰度到测试长者", Rollout: opsconfig.RolloutInput{
			ScopeType: "SUBJECT", ScopeID: testSubjectPublicID, RolloutPercent: 100, MaxConcurrentCalls: 100,
		}}, bundle.RowVersion)
	if err != nil || bundle.Status != "PUBLISHED" {
		t.Fatalf("publish Agent bundle=%+v err=%v", bundle, err)
	}
	issuer, _ := rtc.NewLocalTokenIssuer([]byte("01234567890123456789012345678901"))
	service, err := NewService(
		idempotency.NewExecutor(pool), NewRepository(), audit.NewRepository(), outbox.NewStore(pool), issuer,
		testLeaseCache{},
		ServiceOptions{RTCAppID: "test-app", EngineMode: "s2s_hybrid", AllocationTTL: 30 * time.Second, TokenTTL: 5 * time.Minute, HeartbeatInterval: 15},
	)
	if err != nil {
		t.Fatal(err)
	}
	embedder := &countingEmbedder{}
	if err = service.ConfigureMemoryEmbedding(pool, embedder); err != nil {
		t.Fatal(err)
	}
	service.now = func() time.Time { return time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC) }
	identity := deviceauth.Identity{ID: testDeviceID, PublicID: testDevicePublicID, TenantID: "10000000-0000-0000-0000-000000000001", Status: "registered"}
	resume := true
	rtcCredentialRenewal := true
	input := CreateInput{DeviceTime: service.now(), Capabilities: Capabilities{RTC: true, AudioTracks: []string{"near"}, UploadResume: &resume, RTCCredentialRenewal: &rtcCredentialRenewal}, Network: Network{Type: "test"}, RequestedGoal: "open_conversation"}
	body := []byte(`{"device_time":"2026-08-23T12:00:00Z","capabilities":{"rtc":true,"audio_tracks":["near"],"upload_resume":true,"rtc_credential_renewal":true},"network":{"type":"test"},"requested_goal":"open_conversation"}`)
	first, replayed, err := service.Create(ctx, identity, "idempotency-test-0001", "request-1", body, input)
	if err != nil || replayed || first.RTC.Token == "" {
		t.Fatalf("first response=%#v replayed=%v err=%v", first, replayed, err)
	}
	var frozenContext, frozenBundleVersion string
	if err := pool.QueryRow(ctx, `SELECT cs.sanitized_summary,cs.prompt_bundle_version FROM context_snapshot cs JOIN conversation c ON c.context_snapshot_id=cs.id WHERE c.public_id=$1::uuid`, first.ConversationID).Scan(&frozenContext, &frozenBundleVersion); err != nil {
		t.Fatal(err)
	}
	if frozenBundleVersion != "cfg:"+bundle.ID || first.Session.EngineMode != "cascade" {
		t.Fatalf("frozen bundle=%q engine=%q", frozenBundleVersion, first.Session.EngineMode)
	}
	for _, expected := range []string{"服务测试长者", "年轻时从事铁路工作", "老照片", "聊聊第一次参加工作的经历", "放慢语速", "工作经历", "交流边界", "收入", "最近第1次通话摘要", "个体工商户"} {
		if !strings.Contains(frozenContext, expected) {
			t.Fatalf("frozen context %q does not contain %q", frozenContext, expected)
		}
	}
	if strings.Contains(frozenContext, "不得进入语音上下文") || strings.Contains(frozenContext, "open_conversation") {
		t.Fatalf("frozen context exposed non-whitelisted or technical fields: %q", frozenContext)
	}
	second, replayed, err := service.Create(ctx, identity, "idempotency-test-0001", "request-2", body, input)
	if err != nil || !replayed || second.ConversationID != first.ConversationID || second.RTC.Token != first.RTC.Token {
		t.Fatalf("second response=%#v replayed=%v err=%v", second, replayed, err)
	}
	if _, _, err := service.Create(ctx, identity, "idempotency-test-0001", "request-3", append(body, ' '), input); !errors.Is(err, idempotency.ErrKeyConflict) {
		t.Fatalf("conflicting body error = %v", err)
	}
	if embedder.calls.Load() != 1 {
		t.Fatalf("embedding calls=%d want=1; sequential replay and body conflict must not repeat provider billing", embedder.calls.Load())
	}
	var conversations, histories, audits, events int
	var tokenPersisted bool
	if err := pool.QueryRow(ctx, "SELECT count(*) FROM conversation WHERE device_id=$1::uuid", testDeviceID).Scan(&conversations); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM conversation_state_history h JOIN conversation c ON c.id=h.conversation_id WHERE c.device_id=$1::uuid`, testDeviceID).Scan(&histories); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, "SELECT count(*) FROM audit_event WHERE actor_public_id=$1::uuid AND action='conversation.create'", testDevicePublicID).Scan(&audits); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM outbox_event o JOIN conversation c ON c.id=o.aggregate_id WHERE c.device_id=$1::uuid AND o.event_type='conversation.room_issued'`, testDeviceID).Scan(&events); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM idempotency_record WHERE actor_key=$1 AND response_body::text LIKE '%dev1.%')`, "device:"+testDevicePublicID).Scan(&tokenPersisted); err != nil {
		t.Fatal(err)
	}
	if conversations != 2 || histories != 2 || audits != 1 || events != 1 || tokenPersisted {
		t.Fatalf("conversation=%d history=%d audit=%d event=%d tokenPersisted=%v", conversations, histories, audits, events, tokenPersisted)
	}
	activateInput := ActivateInput{
		JoinedAt: service.now().Add(time.Second), RoomID: first.RTC.RoomID, RTCUserID: first.RTC.UserID,
		RTCJoinResult: "success", ConfigRevision: 1,
	}
	activateBody, _ := json.Marshal(activateInput)
	activated, replayed, err := service.Activate(ctx, identity, first.ConversationID, "activate-idempotency-0001", "activation-request", activateBody, activateInput)
	if err != nil || replayed || activated.Status != "STARTING_AGENT" {
		t.Fatalf("activation=%#v replayed=%v err=%v", activated, replayed, err)
	}
	replayedActivation, replayed, err := service.Activate(ctx, identity, first.ConversationID, "activate-idempotency-0001", "activation-request-replay", activateBody, activateInput)
	if err != nil || !replayed || replayedActivation != activated {
		t.Fatalf("replayed activation=%#v replayed=%v err=%v", replayedActivation, replayed, err)
	}
	var conversationInternalID, providerTaskRef string
	if err := pool.QueryRow(ctx, `SELECT id::text, provider_task_ref FROM conversation WHERE public_id=$1::uuid`, first.ConversationID).Scan(&conversationInternalID, &providerTaskRef); err != nil {
		t.Fatal(err)
	}
	providerClient := &testVoiceProvider{}
	providerHealth, err := NewProviderHealth(pool, 3, time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	startProcessor, err := NewStartProcessor(pool, providerClient, providerHealth, 75*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	startProcessor.now = func() time.Time { return service.now().Add(2 * time.Second) }
	if err := startProcessor.Handle(ctx, outbox.Event{AggregateID: conversationInternalID}); err != nil {
		t.Fatal(err)
	}
	if providerClient.startRequest.TaskRef != providerTaskRef || providerClient.startRequest.ConversationID != first.ConversationID || providerClient.startRequest.RoomID != first.RTC.RoomID || providerClient.startRequest.EngineMode != first.Session.EngineMode {
		t.Fatalf("provider start request=%#v", providerClient.startRequest)
	}
	if providerClient.startRequest.ContextSummary != frozenContext {
		t.Fatalf("provider context=%q want frozen context=%q", providerClient.startRequest.ContextSummary, frozenContext)
	}
	if providerClient.startRequest.PromptBundleVersion != frozenBundleVersion ||
		providerClient.startRequest.AgentBundleVersion != bundle.ID ||
		providerClient.startRequest.PersonaVersion != testPersonaConfigPublicID ||
		providerClient.startRequest.VoiceVersion != testVoiceConfigPublicID ||
		providerClient.startRequest.ModelVersion != testModelConfigPublicID ||
		providerClient.startRequest.RoutingVersion != testRoutingConfigPublicID ||
		providerClient.startRequest.SystemPrompt != "你是已审核发布的陪伴助手；一次只问一个问题；不要编造长者事实。" ||
		providerClient.startRequest.PersonaTone != "warm" ||
		providerClient.startRequest.VoiceResourceRef != "voice://orchestration/approved" ||
		providerClient.startRequest.ModelResourceRef != "model://orchestration/approved" {
		t.Fatalf("provider bundle request=%#v", providerClient.startRequest)
	}
	processor, _ := NewProviderEventProcessor(pool, 75*time.Second)
	processor.now = func() time.Time { return service.now().Add(3 * time.Second) }
	callback := voiceprovider.CallbackEvent{
		Provider: "mock", EventID: "provider-event-started-1", TaskRef: providerTaskRef,
		ConversationID: first.ConversationID, EventType: "voice_chat.started", Status: "ACTIVE",
		OccurredAt: service.now().Add(2500 * time.Millisecond),
	}
	rawCallback, _ := json.Marshal(callback)
	processed, err := processor.Process(ctx, callback, rawCallback)
	if err != nil || processed.Status != "ACTIVE" || processed.Duplicate || processed.Ignored {
		t.Fatalf("processed=%#v err=%v", processed, err)
	}
	duplicateCallback, err := processor.Process(ctx, callback, rawCallback)
	if err != nil || !duplicateCallback.Duplicate || duplicateCallback.Status != "ACTIVE" {
		t.Fatalf("duplicate=%#v err=%v", duplicateCallback, err)
	}
	conflictingCallback := callback
	conflictingCallback.Status = "FAILED"
	conflictingBody, _ := json.Marshal(conflictingCallback)
	if _, err := processor.Process(ctx, conflictingCallback, conflictingBody); !errors.Is(err, ErrProviderEventConflict) {
		t.Fatalf("conflicting callback error=%v", err)
	}
	callback.EventID = "provider-event-started-late"
	lateBody, _ := json.Marshal(callback)
	lateCallback, err := processor.Process(ctx, callback, lateBody)
	if err != nil || !lateCallback.Ignored || lateCallback.Status != "ACTIVE" {
		t.Fatalf("late=%#v err=%v", lateCallback, err)
	}
	var finalStatus, finalTaskStatus string
	var providerEvents int
	if err := pool.QueryRow(ctx, `SELECT c.status, r.status FROM conversation c JOIN rtc_task r ON r.conversation_id=c.id WHERE c.public_id=$1::uuid`, first.ConversationID).Scan(&finalStatus, &finalTaskStatus); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM provider_event e JOIN conversation c ON c.id=e.conversation_id WHERE c.public_id=$1::uuid`, first.ConversationID).Scan(&providerEvents); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM conversation_state_history h JOIN conversation c ON c.id=h.conversation_id WHERE c.public_id=$1::uuid`, first.ConversationID).Scan(&histories); err != nil {
		t.Fatal(err)
	}
	if finalStatus != "ACTIVE" || finalTaskStatus != "ACTIVE" || providerEvents != 2 || histories != 4 {
		t.Fatalf("conversation=%s task=%s events=%d history=%d", finalStatus, finalTaskStatus, providerEvents, histories)
	}
	heartbeatOptions := HeartbeatServiceOptions{
		RTCAppID: "test-app", LeaseTTL: time.Minute, TokenTTL: 5 * time.Minute, CredentialRenewalWindow: 30 * time.Second,
	}
	heartbeatService, err := NewHeartbeatService(pool, NewRepository(), testLeaseCache{}, issuer, heartbeatOptions)
	if err != nil {
		t.Fatal(err)
	}
	heartbeatService.now = func() time.Time { return time.Date(2026, 8, 23, 12, 0, 3, 500*1e6, time.UTC) }
	oldCredentialExpiry := heartbeatService.now().Add(20 * time.Second)
	if _, err := pool.Exec(ctx, `UPDATE rtc_task SET credential_issued_at=$2,credential_expires_at=$3 WHERE conversation_id=$1::uuid`, conversationInternalID, service.now(), oldCredentialExpiry); err != nil {
		t.Fatal(err)
	}
	monotonicMS := int64(12345)
	heartbeatInput := HeartbeatInput{SequenceNo: 1, DeviceTime: heartbeatService.now(), MonotonicMS: &monotonicMS, ConversationStatus: "ACTIVE", RTC: RTCHeartbeat{Joined: true, Uplink: true, Downlink: true}, Network: Network{Type: "wifi"}, FirmwareVersion: "1.0.0", ReportedConfigRevision: 1, RecordingQueue: RecordingQueue{}, RTCCredentialExpiresAt: &oldCredentialExpiry, CommandResults: []json.RawMessage{}}
	failingHeartbeatService, err := NewHeartbeatService(pool, NewRepository(), testLeaseCache{}, failingTokenIssuer{}, heartbeatOptions)
	if err != nil {
		t.Fatal(err)
	}
	failingHeartbeatService.now = heartbeatService.now
	if _, err = pool.Exec(ctx, `UPDATE conversation SET rtc_credential_renewal_supported=false WHERE id=$1::uuid`, conversationInternalID); err != nil {
		t.Fatal(err)
	}
	legacyHeartbeat, err := failingHeartbeatService.Heartbeat(ctx, identity, first.ConversationID, heartbeatInput)
	if err != nil || len(legacyHeartbeat.Actions) != 1 || legacyHeartbeat.Actions[0].Type != "continue" {
		t.Fatalf("legacy heartbeat=%#v err=%v", legacyHeartbeat, err)
	}
	var legacyExpiry time.Time
	if err = pool.QueryRow(ctx, `SELECT credential_expires_at FROM rtc_task WHERE conversation_id=$1::uuid`, conversationInternalID).Scan(&legacyExpiry); err != nil {
		t.Fatal(err)
	}
	if !legacyExpiry.Equal(oldCredentialExpiry) {
		t.Fatalf("legacy client rotated credential metadata: expiry=%s", legacyExpiry)
	}
	if _, err = pool.Exec(ctx, `UPDATE conversation SET rtc_credential_renewal_supported=true,last_heartbeat_seq=0 WHERE id=$1::uuid`, conversationInternalID); err != nil {
		t.Fatal(err)
	}
	if _, err = failingHeartbeatService.Heartbeat(ctx, identity, first.ConversationID, heartbeatInput); err == nil {
		t.Fatal("credential signing failure did not abort heartbeat")
	}
	var expiryAfterFailure time.Time
	var sequenceAfterFailure int64
	if err = pool.QueryRow(ctx, `SELECT r.credential_expires_at,c.last_heartbeat_seq FROM rtc_task r JOIN conversation c ON c.id=r.conversation_id WHERE c.id=$1::uuid`, conversationInternalID).Scan(&expiryAfterFailure, &sequenceAfterFailure); err != nil {
		t.Fatal(err)
	}
	if !expiryAfterFailure.Equal(oldCredentialExpiry) || sequenceAfterFailure != 0 {
		t.Fatalf("failed renewal committed metadata expiry=%s sequence=%d", expiryAfterFailure, sequenceAfterFailure)
	}
	heartbeat, err := heartbeatService.Heartbeat(ctx, identity, first.ConversationID, heartbeatInput)
	if err != nil || heartbeat.LeaseExpiresAt == nil || len(heartbeat.Actions) != 2 || heartbeat.Actions[0].Type != "continue" || heartbeat.Actions[1].Type != "renew_rtc_credential" || heartbeat.Actions[1].RTC == nil || heartbeat.Actions[1].RTC.Token == "" || !heartbeat.Actions[1].RTC.ExpiresAt.Equal(heartbeatService.now().Add(5*time.Minute)) {
		t.Fatalf("heartbeat=%#v err=%v", heartbeat, err)
	}
	staleHeartbeat, err := heartbeatService.Heartbeat(ctx, identity, first.ConversationID, heartbeatInput)
	if err != nil || staleHeartbeat.LeaseExpiresAt == nil || !staleHeartbeat.LeaseExpiresAt.Equal(*heartbeat.LeaseExpiresAt) || len(staleHeartbeat.Actions) != 2 || staleHeartbeat.Actions[1].RTC == nil || staleHeartbeat.Actions[1].RTC.Token != heartbeat.Actions[1].RTC.Token {
		t.Fatalf("stale heartbeat=%#v err=%v", staleHeartbeat, err)
	}
	heartbeatInput.SequenceNo = 2
	heartbeatInput.RTCCredentialExpiresAt = &heartbeat.Actions[1].RTC.ExpiresAt
	heartbeatInput.ReportedConfigRevision = 2
	if _, err := heartbeatService.Heartbeat(ctx, identity, first.ConversationID, heartbeatInput); !errors.Is(err, ErrConfigVersionConflict) {
		t.Fatalf("future heartbeat configuration error=%v", err)
	}
	service.now = func() time.Time { return time.Date(2026, 8, 23, 12, 0, 4, 0, time.UTC) }
	stopInput := StopInput{
		Reason: "handset_on_hook", DeviceTime: service.now(), LastHeartbeatSequence: 2,
		RTCLeft: true, Recording: StopRecording{NearSegments: 1, FarSegments: 1},
	}
	stopBody := []byte(`{"reason":"handset_on_hook","device_time":"2026-08-23T12:00:04Z","last_heartbeat_sequence":2,"rtc_left":true,"recording":{"near_segments":1,"far_segments":1}}`)
	stopped, replayed, err := service.Stop(ctx, identity, first.ConversationID, "stop-idempotency-0001", "stop-request-1", stopBody, stopInput)
	if err != nil || replayed || stopped.Status != "STOPPING" {
		t.Fatalf("stop=%#v replayed=%v err=%v", stopped, replayed, err)
	}
	replayedStop, replayed, err := service.Stop(ctx, identity, first.ConversationID, "stop-idempotency-0001", "stop-request-2", stopBody, stopInput)
	if err != nil || !replayed || replayedStop.Status != "STOPPING" {
		t.Fatalf("replayed stop=%#v replayed=%v err=%v", replayedStop, replayed, err)
	}
	var stopEvents, stopAudits int
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM outbox_event o JOIN conversation c ON c.id=o.aggregate_id
		WHERE c.public_id=$1::uuid AND o.event_type='conversation.agent_stop_requested'`, first.ConversationID).Scan(&stopEvents); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event a JOIN conversation c ON c.id=a.resource_id
		WHERE c.public_id=$1::uuid AND a.action='conversation.stop'`, first.ConversationID).Scan(&stopAudits); err != nil {
		t.Fatal(err)
	}
	if stopEvents != 1 || stopAudits != 1 {
		t.Fatalf("stop events=%d audits=%d", stopEvents, stopAudits)
	}
	stopProcessor, err := NewStopProcessor(pool, providerClient, providerHealth)
	if err != nil {
		t.Fatal(err)
	}
	stopProcessor.now = func() time.Time { return service.now().Add(500 * time.Millisecond) }
	if err := stopProcessor.Handle(ctx, outbox.Event{AggregateID: conversationInternalID}); err != nil {
		t.Fatal(err)
	}
	if providerClient.stopRequest.TaskRef != providerTaskRef || providerClient.stopRequest.ConversationID != first.ConversationID || providerClient.stopRequest.Reason != stopInput.Reason {
		t.Fatalf("provider stop request=%#v", providerClient.stopRequest)
	}
	processor.now = func() time.Time { return time.Date(2026, 8, 23, 12, 0, 5, 0, time.UTC) }
	stopCallback := voiceprovider.CallbackEvent{
		Provider: "mock", EventID: "provider-event-stopped-1", TaskRef: providerTaskRef,
		ConversationID: first.ConversationID, EventType: "voice_chat.stopped", Status: "COMPLETED",
		OccurredAt: service.now().Add(time.Second),
	}
	rawStopCallback, _ := json.Marshal(stopCallback)
	completed, err := processor.Process(ctx, stopCallback, rawStopCallback)
	if err != nil || completed.Status != "COMPLETED" || completed.Ignored {
		t.Fatalf("completed=%#v err=%v", completed, err)
	}
	duplicateStop, err := processor.Process(ctx, stopCallback, rawStopCallback)
	if err != nil || !duplicateStop.Duplicate || duplicateStop.Status != "COMPLETED" {
		t.Fatalf("duplicate stop=%#v err=%v", duplicateStop, err)
	}
	if err := pool.QueryRow(ctx, `SELECT c.status, r.status FROM conversation c JOIN rtc_task r ON r.conversation_id=c.id
		WHERE c.public_id=$1::uuid`, first.ConversationID).Scan(&finalStatus, &finalTaskStatus); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM conversation_state_history h JOIN conversation c ON c.id=h.conversation_id
		WHERE c.public_id=$1::uuid`, first.ConversationID).Scan(&histories); err != nil {
		t.Fatal(err)
	}
	if finalStatus != "COMPLETED" || finalTaskStatus != "COMPLETED" || histories != 6 {
		t.Fatalf("final conversation=%s task=%s history=%d", finalStatus, finalTaskStatus, histories)
	}

	// Exercise the synchronous ACTIVE/COMPLETED provider responses in addition
	// to the asynchronous callback path above. This is the other supported
	// provider contract and must preserve the same durable state transitions.
	secondConversation, replayed, err := service.Create(ctx, identity, "idempotency-test-0002", "request-4", body, input)
	if err != nil || replayed {
		t.Fatalf("second conversation=%#v replayed=%v err=%v", secondConversation, replayed, err)
	}
	secondActivateInput := activateInput
	secondActivateInput.RoomID = secondConversation.RTC.RoomID
	secondActivateInput.RTCUserID = secondConversation.RTC.UserID
	secondActivateBody, _ := json.Marshal(secondActivateInput)
	if _, replayed, err := service.Activate(ctx, identity, secondConversation.ConversationID, "activate-idempotency-0002", "activation-request-2", secondActivateBody, secondActivateInput); err != nil || replayed {
		t.Fatalf("second activation replayed=%v err=%v", replayed, err)
	}
	var secondInternalID string
	if err := pool.QueryRow(ctx, `SELECT id::text FROM conversation WHERE public_id=$1::uuid`, secondConversation.ConversationID).Scan(&secondInternalID); err != nil {
		t.Fatal(err)
	}
	providerClient.startStatus = "ACTIVE"
	secondStartProcessor, err := NewStartProcessor(pool, providerClient, providerHealth, 75*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	if err := secondStartProcessor.ConfigureStartProtection(allowProviderStart{}, 100); err != nil {
		t.Fatal(err)
	}
	secondStartProcessor.now = func() time.Time { return service.now().Add(2 * time.Minute) }
	if err := secondStartProcessor.Handle(ctx, outbox.Event{AggregateID: secondInternalID}); err != nil {
		t.Fatal(err)
	}
	secondStopInput := stopInput
	secondStopInput.DeviceTime = service.now().Add(2*time.Minute + time.Second)
	secondStopBody, _ := json.Marshal(secondStopInput)
	if _, replayed, err := service.Stop(ctx, identity, secondConversation.ConversationID, "stop-idempotency-0002", "stop-request-3", secondStopBody, secondStopInput); err != nil || replayed {
		t.Fatalf("second stop replayed=%v err=%v", replayed, err)
	}
	providerClient.stopStatus = "COMPLETED"
	secondStopProcessor, err := NewStopProcessor(pool, providerClient, providerHealth)
	if err != nil {
		t.Fatal(err)
	}
	secondStopProcessor.now = func() time.Time { return service.now().Add(2*time.Minute + 2*time.Second) }
	if err := secondStopProcessor.Handle(ctx, outbox.Event{AggregateID: secondInternalID}); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT c.status, r.status FROM conversation c JOIN rtc_task r ON r.conversation_id=c.id WHERE c.id=$1::uuid`, secondInternalID).Scan(&finalStatus, &finalTaskStatus); err != nil {
		t.Fatal(err)
	}
	if finalStatus != "COMPLETED" || finalTaskStatus != "COMPLETED" {
		t.Fatalf("second final conversation=%s task=%s", finalStatus, finalTaskStatus)
	}

	usageCallback := voiceprovider.CallbackEvent{
		Provider: "mock", EventID: "provider-event-usage-1", TaskRef: providerTaskRef,
		ConversationID: first.ConversationID, EventType: "voice_chat.usage", Status: "FINAL",
		OccurredAt: service.now().Add(3 * time.Minute), Usage: []voiceprovider.UsageMetric{{
			UsageCategory: "LLM", MetricCode: "input_tokens", Quantity: "12", Unit: "token",
		}},
	}
	usageBody, _ := json.Marshal(usageCallback)
	usageResult, err := processor.Process(ctx, usageCallback, usageBody)
	if err != nil || usageResult.UsageAccepted != 1 {
		t.Fatalf("provider usage result=%#v err=%v", usageResult, err)
	}

	// A provider can end an active call before a device stop request arrives.
	// Verify that callback-only termination creates both durable transitions and
	// schedules content extraction.
	providerEndedConversation, replayed, err := service.Create(ctx, identity, "idempotency-test-0003", "request-5", body, input)
	if err != nil || replayed {
		t.Fatalf("provider-ended conversation=%#v replayed=%v err=%v", providerEndedConversation, replayed, err)
	}
	providerEndedActivate := activateInput
	providerEndedActivate.RoomID = providerEndedConversation.RTC.RoomID
	providerEndedActivate.RTCUserID = providerEndedConversation.RTC.UserID
	providerEndedActivateBody, _ := json.Marshal(providerEndedActivate)
	if _, replayed, err := service.Activate(ctx, identity, providerEndedConversation.ConversationID, "activate-idempotency-0003", "activation-request-3", providerEndedActivateBody, providerEndedActivate); err != nil || replayed {
		t.Fatalf("provider-ended activation replayed=%v err=%v", replayed, err)
	}
	var providerEndedInternalID, providerEndedTaskRef string
	if err := pool.QueryRow(ctx, `SELECT id::text,provider_task_ref FROM conversation WHERE public_id=$1::uuid`, providerEndedConversation.ConversationID).Scan(&providerEndedInternalID, &providerEndedTaskRef); err != nil {
		t.Fatal(err)
	}
	if err := secondStartProcessor.Handle(ctx, outbox.Event{AggregateID: providerEndedInternalID}); err != nil {
		t.Fatal(err)
	}
	providerEndedCallback := voiceprovider.CallbackEvent{
		Provider: "mock", EventID: "provider-event-ended-active", TaskRef: providerEndedTaskRef,
		ConversationID: providerEndedConversation.ConversationID, EventType: "voice_chat.stopped", Status: "COMPLETED",
		OccurredAt: service.now().Add(4 * time.Minute),
	}
	providerEndedBody, _ := json.Marshal(providerEndedCallback)
	providerEndedResult, err := processor.Process(ctx, providerEndedCallback, providerEndedBody)
	if err != nil || providerEndedResult.Status != "COMPLETED" || providerEndedResult.Ignored {
		t.Fatalf("provider-ended result=%#v err=%v", providerEndedResult, err)
	}

	// A runtime failure moves an active conversation to STOPPING and repeated
	// failure delivery remains safe while retaining the first failure cause.
	failedConversation, replayed, err := service.Create(ctx, identity, "idempotency-test-0004", "request-6", body, input)
	if err != nil || replayed {
		t.Fatalf("failed conversation=%#v replayed=%v err=%v", failedConversation, replayed, err)
	}
	failedActivate := activateInput
	failedActivate.RoomID = failedConversation.RTC.RoomID
	failedActivate.RTCUserID = failedConversation.RTC.UserID
	failedActivateBody, _ := json.Marshal(failedActivate)
	if _, replayed, err := service.Activate(ctx, identity, failedConversation.ConversationID, "activate-idempotency-0004", "activation-request-4", failedActivateBody, failedActivate); err != nil || replayed {
		t.Fatalf("failed activation replayed=%v err=%v", replayed, err)
	}
	var failedInternalID, failedTaskRef string
	if err := pool.QueryRow(ctx, `SELECT id::text,provider_task_ref FROM conversation WHERE public_id=$1::uuid`, failedConversation.ConversationID).Scan(&failedInternalID, &failedTaskRef); err != nil {
		t.Fatal(err)
	}
	if err := secondStartProcessor.Handle(ctx, outbox.Event{AggregateID: failedInternalID}); err != nil {
		t.Fatal(err)
	}
	failedCallback := voiceprovider.CallbackEvent{
		Provider: "mock", EventID: "provider-event-runtime-failed-1", TaskRef: failedTaskRef,
		ConversationID: failedConversation.ConversationID, EventType: "voice_chat.failed", Status: "FAILED",
		OccurredAt: service.now().Add(5 * time.Minute),
	}
	failedBody, _ := json.Marshal(failedCallback)
	failedResult, err := processor.Process(ctx, failedCallback, failedBody)
	if err != nil || failedResult.Status != "STOPPING" || failedResult.Ignored {
		t.Fatalf("runtime failure result=%#v err=%v", failedResult, err)
	}
	failedCallback.EventID = "provider-event-runtime-failed-2"
	repeatedFailedBody, _ := json.Marshal(failedCallback)
	repeatedFailedResult, err := processor.Process(ctx, failedCallback, repeatedFailedBody)
	if err != nil || repeatedFailedResult.Status != "STOPPING" || repeatedFailedResult.Ignored {
		t.Fatalf("repeated runtime failure result=%#v err=%v", repeatedFailedResult, err)
	}
	failedStopProcessor, err := NewStopProcessor(pool, providerClient, providerHealth)
	if err != nil {
		t.Fatal(err)
	}
	providerClient.stopErr = &voiceprovider.Error{Operation: "stop", Temporary: true, Cause: errors.New("temporary test failure")}
	if err := failedStopProcessor.Handle(ctx, outbox.Event{AggregateID: failedInternalID}); err == nil {
		t.Fatal("temporary provider stop failure was not retried")
	}
	providerClient.stopErr = nil
	providerClient.stopStatus = "COMPLETED"
	if err := failedStopProcessor.Handle(ctx, outbox.Event{AggregateID: failedInternalID}); err != nil {
		t.Fatal(err)
	}

	// Reconciliation after an ambiguous start must query first. A still-starting
	// response is retried, a confirmed 404 permits the same stable task to be
	// started again, and a later ACTIVE query closes the ambiguity safely.
	reconciledConversation, replayed, err := service.Create(ctx, identity, "idempotency-test-0005", "request-7", body, input)
	if err != nil || replayed {
		t.Fatalf("reconciled conversation=%#v replayed=%v err=%v", reconciledConversation, replayed, err)
	}
	reconciledActivate := activateInput
	reconciledActivate.RoomID = reconciledConversation.RTC.RoomID
	reconciledActivate.RTCUserID = reconciledConversation.RTC.UserID
	reconciledActivateBody, _ := json.Marshal(reconciledActivate)
	if _, replayed, err := service.Activate(ctx, identity, reconciledConversation.ConversationID, "activate-idempotency-0005", "activation-request-5", reconciledActivateBody, reconciledActivate); err != nil || replayed {
		t.Fatalf("reconciled activation replayed=%v err=%v", replayed, err)
	}
	var reconciledInternalID string
	if err := pool.QueryRow(ctx, `SELECT id::text FROM conversation WHERE public_id=$1::uuid`, reconciledConversation.ConversationID).Scan(&reconciledInternalID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `UPDATE rtc_task SET status='UNKNOWN' WHERE conversation_id=$1::uuid`, reconciledInternalID); err != nil {
		t.Fatal(err)
	}
	providerClient.queryStatus = "STARTING"
	if err := secondStartProcessor.Handle(ctx, outbox.Event{AggregateID: reconciledInternalID}); err == nil {
		t.Fatal("still-starting reconciliation was not retried")
	}
	if _, err := pool.Exec(ctx, `UPDATE rtc_task SET status='UNKNOWN' WHERE conversation_id=$1::uuid`, reconciledInternalID); err != nil {
		t.Fatal(err)
	}
	providerClient.queryErr = &voiceprovider.Error{Operation: "query", StatusCode: 404}
	providerClient.startErr = &voiceprovider.Error{Operation: "start", Temporary: true, Cause: errors.New("ambiguous start")}
	if err := secondStartProcessor.Handle(ctx, outbox.Event{AggregateID: reconciledInternalID}); err == nil {
		t.Fatal("temporary provider start failure was not retried")
	}
	providerClient.queryErr = nil
	providerClient.startErr = nil
	providerClient.queryStatus = "ACTIVE"
	if err := secondStartProcessor.Handle(ctx, outbox.Event{AggregateID: reconciledInternalID}); err != nil {
		t.Fatal(err)
	}
	reconciledStopInput := stopInput
	reconciledStopInput.DeviceTime = service.now().Add(6*time.Minute + time.Second)
	reconciledStopBody, _ := json.Marshal(reconciledStopInput)
	if _, replayed, err := service.Stop(ctx, identity, reconciledConversation.ConversationID, "stop-idempotency-0005", "stop-request-5", reconciledStopBody, reconciledStopInput); err != nil || replayed {
		t.Fatalf("reconciled stop replayed=%v err=%v", replayed, err)
	}
	if err := failedStopProcessor.Handle(ctx, outbox.Event{AggregateID: reconciledInternalID}); err != nil {
		t.Fatal(err)
	}

	// A permanent provider start rejection is not retryable. It must close the
	// authoritative state immediately so the device can leave the RTC room with
	// a stable, non-sensitive failure code instead of timing out in STARTING_AGENT.
	rejectedConversation, replayed, err := service.Create(ctx, identity, "idempotency-test-0006", "request-8", body, input)
	if err != nil || replayed {
		t.Fatalf("rejected conversation=%#v replayed=%v err=%v", rejectedConversation, replayed, err)
	}
	rejectedActivate := activateInput
	rejectedActivate.RoomID = rejectedConversation.RTC.RoomID
	rejectedActivate.RTCUserID = rejectedConversation.RTC.UserID
	rejectedActivateBody, _ := json.Marshal(rejectedActivate)
	if _, replayed, err = service.Activate(ctx, identity, rejectedConversation.ConversationID, "activate-idempotency-0006", "activation-request-6", rejectedActivateBody, rejectedActivate); err != nil || replayed {
		t.Fatalf("rejected activation replayed=%v err=%v", replayed, err)
	}
	var rejectedInternalID string
	if err = pool.QueryRow(ctx, `SELECT id::text FROM conversation WHERE public_id=$1::uuid`, rejectedConversation.ConversationID).Scan(&rejectedInternalID); err != nil {
		t.Fatal(err)
	}
	providerClient.startErr = &voiceprovider.Error{Operation: "start", StatusCode: 400, Cause: errors.New("invalid provider request")}
	if err = secondStartProcessor.Handle(ctx, outbox.Event{AggregateID: rejectedInternalID}); err != nil {
		t.Fatalf("permanent provider rejection was not durably consumed: %v", err)
	}
	providerClient.startErr = nil
	rejectedState, err := service.GetDeviceState(ctx, identity, rejectedConversation.ConversationID)
	if err != nil || rejectedState.Status != "FAILED" || rejectedState.Action != "leave_room" || rejectedState.FailureCode != "provider_start_rejected" {
		t.Fatalf("rejected device state=%#v err=%v", rejectedState, err)
	}
	var rejectedTaskStatus string
	var rejectedHistory, rejectedAudits int
	if err = pool.QueryRow(ctx, `SELECT status FROM rtc_task WHERE conversation_id=$1::uuid`, rejectedInternalID).Scan(&rejectedTaskStatus); err != nil {
		t.Fatal(err)
	}
	if err = pool.QueryRow(ctx, `SELECT count(*) FROM conversation_state_history WHERE conversation_id=$1::uuid AND from_status='STARTING_AGENT' AND to_status='FAILED'`, rejectedInternalID).Scan(&rejectedHistory); err != nil {
		t.Fatal(err)
	}
	if err = pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE resource_id=$1::uuid AND action='conversation.agent_start_failed'`, rejectedInternalID).Scan(&rejectedAudits); err != nil {
		t.Fatal(err)
	}
	if rejectedTaskStatus != "FAILED" || rejectedHistory != 1 || rejectedAudits != 1 {
		t.Fatalf("rejected task=%s history=%d audits=%d", rejectedTaskStatus, rejectedHistory, rejectedAudits)
	}

	transcriptService, err := NewTranscriptBatchService(pool, idempotency.NewExecutor(pool), audit.NewRepository(), outbox.NewStore(pool))
	if err != nil {
		t.Fatal(err)
	}
	providerTurn := TranscriptTurnInput{
		TurnKey: "provider-final-1", SequenceNo: 1, Speaker: "subject",
		StartAt: time.Date(2026, 8, 23, 12, 0, 1, 0, time.UTC), EndAt: time.Date(2026, 8, 23, 12, 0, 2, 0, time.UTC),
		FinalText: "供应商已经返回的终稿", Interrupted: false,
	}
	lateTranscript := voiceprovider.CallbackEvent{
		Provider: "mock", EventID: "provider-event-transcript-late-1", TaskRef: providerTaskRef,
		ConversationID: first.ConversationID, EventType: "voice_chat.transcript.final", Status: "FINAL",
		OccurredAt: service.now().Add(2 * time.Second), Turns: []voiceprovider.FinalTurn{{
			TurnKey: providerTurn.TurnKey, SequenceNo: providerTurn.SequenceNo, Speaker: providerTurn.Speaker,
			StartAt: providerTurn.StartAt, EndAt: providerTurn.EndAt, FinalText: providerTurn.FinalText,
			Interrupted: providerTurn.Interrupted, ProviderRef: "provider-turn-1",
		}},
	}
	lateTranscriptBody, _ := json.Marshal(lateTranscript)
	lateTranscriptResult, err := processor.Process(ctx, lateTranscript, lateTranscriptBody)
	if err != nil || lateTranscriptResult.TurnsAccepted != 1 || lateTranscriptResult.Status != "COMPLETED" {
		t.Fatalf("late transcript=%#v err=%v", lateTranscriptResult, err)
	}
	deviceTurn := TranscriptTurnInput{
		TurnKey: "device-final-2", SequenceNo: 2, Speaker: "agent",
		StartAt: time.Date(2026, 8, 23, 12, 0, 2, 0, time.UTC), EndAt: time.Date(2026, 8, 23, 12, 0, 3, 0, time.UTC),
		FinalText: "终端补传的缺失终稿", Interrupted: true,
	}
	transcriptInput := TranscriptBatchInput{Turns: []TranscriptTurnInput{providerTurn, deviceTurn}}
	transcriptBody, _ := json.Marshal(transcriptInput)
	transcriptResponse, replayed, err := transcriptService.Submit(ctx, identity, first.ConversationID, "transcript-idempotency-0001", "transcript-request-1", transcriptBody, transcriptInput)
	if err != nil || replayed || transcriptResponse.AcceptedCount != 1 || transcriptResponse.DuplicateCount != 1 {
		t.Fatalf("transcript response=%#v replayed=%v err=%v", transcriptResponse, replayed, err)
	}
	replayedTranscript, replayed, err := transcriptService.Submit(ctx, identity, first.ConversationID, "transcript-idempotency-0001", "transcript-request-2", transcriptBody, transcriptInput)
	if err != nil || !replayed || replayedTranscript != transcriptResponse {
		t.Fatalf("replayed transcript=%#v replayed=%v err=%v", replayedTranscript, replayed, err)
	}
	duplicateTranscript, replayed, err := transcriptService.Submit(ctx, identity, first.ConversationID, "transcript-idempotency-0002", "transcript-request-3", transcriptBody, transcriptInput)
	if err != nil || replayed || duplicateTranscript.AcceptedCount != 0 || duplicateTranscript.DuplicateCount != 2 {
		t.Fatalf("duplicate transcript=%#v replayed=%v err=%v", duplicateTranscript, replayed, err)
	}
	conflictingInput := TranscriptBatchInput{Turns: []TranscriptTurnInput{deviceTurn}}
	conflictingInput.Turns[0].FinalText = "不一致的终稿"
	conflictingBody, _ = json.Marshal(conflictingInput)
	if _, _, err := transcriptService.Submit(ctx, identity, first.ConversationID, "transcript-idempotency-0003", "transcript-request-4", conflictingBody, conflictingInput); err == nil {
		t.Fatal("conflicting transcript was accepted")
	} else {
		var conflict *TranscriptConflictError
		if !errors.As(err, &conflict) || len(conflict.Conflicts) != 1 {
			t.Fatalf("conflicting transcript error=%v", err)
		}
	}
	wrongDevice := identity
	wrongDevice.ID = "90000000-0000-0000-0000-000000000599"
	if _, _, err := transcriptService.Submit(ctx, wrongDevice, first.ConversationID, "transcript-idempotency-0004", "transcript-request-5", transcriptBody, transcriptInput); !errors.Is(err, ErrConversationNotFound) {
		t.Fatalf("wrong device transcript error=%v", err)
	}
	if _, err := pool.Exec(ctx, `UPDATE conversation SET retention_mode='none' WHERE public_id=$1::uuid`, first.ConversationID); err != nil {
		t.Fatal(err)
	}
	if _, _, err := transcriptService.Submit(ctx, identity, first.ConversationID, "transcript-idempotency-0005", "transcript-request-6", transcriptBody, transcriptInput); !errors.Is(err, ErrTranscriptStorageNotAllowed) {
		t.Fatalf("retention-disabled transcript error=%v", err)
	}
	var turnCount, deviceTurnCount, transcriptAudits, extractionEvents, providerExtractionEvents int
	var extractionGeneration int64
	if err := pool.QueryRow(ctx, `SELECT count(*),count(*) FILTER (WHERE source_type='device') FROM conversation_turn
		WHERE conversation_id=(SELECT id FROM conversation WHERE public_id=$1::uuid)`, first.ConversationID).Scan(&turnCount, &deviceTurnCount); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE actor_public_id=$1::uuid AND action='conversation.transcript_batch.submit'`, testDevicePublicID).Scan(&transcriptAudits); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*),max((payload->>'generation_version')::bigint) FROM outbox_event WHERE aggregate_id=(SELECT id FROM conversation WHERE public_id=$1::uuid)
		AND event_type='conversation.content_extraction_requested'
		AND deduplication_key LIKE 'content-extraction-v1:%:device:%'`, first.ConversationID).Scan(&extractionEvents, &extractionGeneration); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM outbox_event WHERE aggregate_id=(SELECT id FROM conversation WHERE public_id=$1::uuid)
		AND event_type='conversation.content_extraction_requested'
		AND deduplication_key LIKE 'content-extraction-v1:%:provider:%'`, first.ConversationID).Scan(&providerExtractionEvents); err != nil {
		t.Fatal(err)
	}
	if turnCount != 2 || deviceTurnCount != 1 || transcriptAudits != 2 || extractionEvents != 1 || providerExtractionEvents != 1 || extractionGeneration != 2 {
		t.Fatalf("turns=%d device=%d audits=%d device_extraction=%d provider_extraction=%d generation=%d", turnCount, deviceTurnCount, transcriptAudits, extractionEvents, providerExtractionEvents, extractionGeneration)
	}
}

func cleanupOrchestrationFixture(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
	t.Helper()
	statements := []struct {
		query string
		arg   string
	}{
		{`DELETE FROM runtime_config_rollout WHERE config_key=$1`, testBundleConfigKey},
		{`DELETE FROM ops_config_evaluation WHERE config_version_id IN (SELECT id FROM runtime_config_version WHERE config_key=$1)`, testBundleConfigKey},
		{`DELETE FROM runtime_config_version WHERE config_key=$1`, testBundleConfigKey},
		{`DELETE FROM audit_event WHERE request_id LIKE $1`, "orchestration-agent-bundle-%"},
		{`DELETE FROM idempotency_record WHERE idempotency_key=$1`, testBundleCreateKey},
		{`DELETE FROM idempotency_record WHERE idempotency_key=$1`, testBundleEvaluationKey},
		{`DELETE FROM runtime_config_version WHERE id=$1::uuid`, testPersonaConfigID},
		{`DELETE FROM runtime_config_version WHERE id=$1::uuid`, testVoiceConfigID},
		{`DELETE FROM runtime_config_version WHERE id=$1::uuid`, testModelConfigID},
		{`DELETE FROM runtime_config_version WHERE id=$1::uuid`, testRoutingConfigID},
		{`DELETE FROM conversation_summary WHERE conversation_id IN (SELECT id FROM conversation WHERE device_id=$1::uuid)`, testDeviceID},
		{`DELETE FROM conversation_turn WHERE conversation_id IN (SELECT id FROM conversation WHERE device_id=$1::uuid)`, testDeviceID},
		{`DELETE FROM provider_event WHERE conversation_id IN (SELECT id FROM conversation WHERE device_id=$1::uuid)`, testDeviceID},
		{`DELETE FROM conversation_state_history WHERE conversation_id IN (SELECT id FROM conversation WHERE device_id=$1::uuid)`, testDeviceID},
		{`DELETE FROM rtc_task WHERE conversation_id IN (SELECT id FROM conversation WHERE device_id=$1::uuid)`, testDeviceID},
		{`DELETE FROM outbox_event WHERE aggregate_id IN (SELECT id FROM conversation WHERE device_id=$1::uuid)`, testDeviceID},
		{`DELETE FROM audit_event WHERE resource_id IN (SELECT id FROM conversation WHERE device_id=$1::uuid)`, testDeviceID},
		{`DELETE FROM audit_event WHERE actor_public_id=$1::uuid`, testDevicePublicID},
		{`DELETE FROM idempotency_record WHERE actor_key=$1`, "device:" + testDevicePublicID},
		{`DELETE FROM conversation WHERE device_id=$1::uuid`, testDeviceID},
		{`DELETE FROM context_snapshot WHERE subject_id=$1::uuid`, testSubjectID},
		{`DELETE FROM consent_snapshot WHERE subject_id=$1::uuid`, testSubjectID},
		{`DELETE FROM consent_record WHERE subject_id=$1::uuid`, testSubjectID},
		{`DELETE FROM device_binding WHERE device_id=$1::uuid`, testDeviceID},
		{`DELETE FROM device_presence_state WHERE device_id=$1::uuid`, testDeviceID},
		{`DELETE FROM device_twin WHERE device_id=$1::uuid`, testDeviceID},
		{`DELETE FROM device WHERE id=$1::uuid`, testDeviceID},
		{`DELETE FROM subject_profile WHERE id=$1::uuid`, testSubjectID},
		{`DELETE FROM billing_price_book WHERE id=$1::uuid`, testPriceBookID},
	}
	for _, statement := range statements {
		if _, err := pool.Exec(ctx, statement.query, statement.arg); err != nil {
			t.Fatalf("cleanup fixture: %v", err)
		}
	}
}