story/internal/orchestration/provider_processing.go

package orchestration

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"data-terminal-cloud/internal/billing"
	"data-terminal-cloud/internal/conversation"
	"data-terminal-cloud/internal/outbox"
	"data-terminal-cloud/internal/ratelimit"
	"data-terminal-cloud/internal/voiceprovider"

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

var (
	ErrProviderTaskNotFound  = errors.New("provider task was not found")
	ErrProviderEventConflict = errors.New("provider event key was reused with different content")
	ErrProviderRateLimited   = errors.New("provider start rate limit reached")
	ErrProviderCapacity      = errors.New("provider active task capacity reached")
)

type providerStartRateLimiter interface {
	Allow(context.Context, string, string) (ratelimit.Result, error)
}

type StartProcessor struct {
	pool            *pgxpool.Pool
	provider        voiceprovider.Client
	health          *ProviderHealth
	initialLeaseTTL time.Duration
	startLimiter    providerStartRateLimiter
	maxActiveTasks  int
	now             func() time.Time
}

// ConfigureStartProtection enables the provider-wide RPS bucket and durable
// active-task capacity reservation used by every worker replica.
func (p *StartProcessor) ConfigureStartProtection(limiter providerStartRateLimiter, maxActiveTasks int) error {
	if limiter == nil || maxActiveTasks <= 0 {
		return errors.New("provider start limiter and positive active task capacity are required")
	}
	p.startLimiter = limiter
	p.maxActiveTasks = maxActiveTasks
	return nil
}

func NewStartProcessor(pool *pgxpool.Pool, provider voiceprovider.Client, health *ProviderHealth, initialLeaseTTL time.Duration) (*StartProcessor, error) {
	if pool == nil || provider == nil || health == nil || initialLeaseTTL <= 0 {
		return nil, errors.New("database pool, voice provider client, provider health and positive lease TTL are required")
	}
	return &StartProcessor{pool: pool, provider: provider, health: health, initialLeaseTTL: initialLeaseTTL, now: time.Now}, nil
}

func (p *StartProcessor) Handle(ctx context.Context, event outbox.Event) error {
	var request voiceprovider.StartRequest
	var conversationStatus conversation.Status
	var taskStatus, tenantID, providerName string
	err := p.pool.QueryRow(ctx, `
		SELECT c.public_id::text, c.status, r.provider_task_ref, r.room_id,
			r.device_rtc_user_id, r.agent_rtc_user_id, r.status,cs.sanitized_summary,
			c.engine_mode,c.tenant_id::text,r.provider,cs.prompt_bundle_version,
			COALESCE(bundle.public_id::text,''),COALESCE(bundle.payload->>'persona_version_id',''),
			COALESCE(bundle.payload->>'voice_version_id',''),COALESCE(bundle.payload->>'model_version_id',''),
			COALESCE(bundle.payload->>'routing_version_id',''),COALESCE(persona.payload->>'system_prompt',''),
			COALESCE(persona.payload->>'tone',''),COALESCE(voice.payload->>'voice_resource_ref',''),
			COALESCE(model.payload->>'model_id','')
		FROM conversation c JOIN rtc_task r ON r.conversation_id=c.id
		JOIN context_snapshot cs ON cs.id=c.context_snapshot_id AND cs.tenant_id=c.tenant_id
		LEFT JOIN runtime_config_version bundle
		  ON bundle.tenant_id=c.tenant_id AND bundle.config_kind='AGENT_BUNDLE'
		 AND cs.prompt_bundle_version='cfg:' || bundle.public_id::text
		LEFT JOIN runtime_config_version persona
		  ON persona.tenant_id=bundle.tenant_id AND persona.config_kind='PERSONA'
		 AND persona.public_id::text=bundle.payload->>'persona_version_id'
		LEFT JOIN runtime_config_version voice
		  ON voice.tenant_id=bundle.tenant_id AND voice.config_kind='VOICE'
		 AND voice.public_id::text=bundle.payload->>'voice_version_id'
		LEFT JOIN runtime_config_version model
		  ON model.tenant_id=bundle.tenant_id AND model.config_kind='MODEL'
		 AND model.public_id::text=bundle.payload->>'model_version_id'
		WHERE c.id=$1::uuid`, event.AggregateID,
	).Scan(&request.ConversationID, &conversationStatus, &request.TaskRef, &request.RoomID,
		&request.DeviceUserID, &request.AgentUserID, &taskStatus, &request.ContextSummary, &request.EngineMode, &tenantID, &providerName,
		&request.PromptBundleVersion, &request.AgentBundleVersion, &request.PersonaVersion, &request.VoiceVersion,
		&request.ModelVersion, &request.RoutingVersion, &request.SystemPrompt, &request.PersonaTone,
		&request.VoiceResourceRef, &request.ModelResourceRef)
	if errors.Is(err, pgx.ErrNoRows) {
		return ErrProviderTaskNotFound
	}
	if err != nil {
		return fmt.Errorf("read provider start task: %w", err)
	}
	if conversationStatus != conversation.StatusStartingAgent {
		return nil
	}
	if taskStatus != "STARTING" && taskStatus != "UNKNOWN" {
		return fmt.Errorf("provider task status %s cannot be started", taskStatus)
	}
	allowed, err := p.health.AllowStart(ctx, tenantID, providerName)
	if err != nil {
		return fmt.Errorf("read provider health: %w", err)
	}
	if !allowed {
		return outbox.Retryable(ErrProviderUnavailable)
	}
	var response voiceprovider.StartResponse
	if taskStatus == "UNKNOWN" {
		queryResult, queryErr := p.provider.Query(ctx, voiceprovider.QueryRequest{
			TaskRef: request.TaskRef, ConversationID: request.ConversationID,
		})
		switch {
		case queryErr == nil && queryResult.Status == "ACTIVE":
			response = voiceprovider.StartResponse{TaskRef: queryResult.TaskRef, Status: queryResult.Status}
		case queryErr == nil && queryResult.Status == "STARTING":
			_, _ = p.pool.Exec(ctx, `UPDATE rtc_task SET status='STARTING',last_checked_at=$2,updated_at=$2 WHERE conversation_id=$1::uuid`, event.AggregateID, p.now().UTC())
			return outbox.Retryable(errors.New("provider start is still pending confirmation"))
		case queryErr == nil:
			return fmt.Errorf("provider query returned unsupported start reconciliation status %s", queryResult.Status)
		case !voiceprovider.IsNotFound(queryErr):
			if voiceprovider.IsTemporary(queryErr) {
				return outbox.Retryable(fmt.Errorf("query unknown provider start before retry: %w", queryErr))
			}
			return fmt.Errorf("query unknown provider start before retry: %w", queryErr)
		}
		// A provider-confirmed 404 is the only UNKNOWN case in which Start may
		// safely be sent again. The same stable task reference is retained.
	}
	newReservation := false
	if response.Status == "" && p.maxActiveTasks > 0 {
		newReservation, err = p.reserveProviderCapacity(ctx, event.AggregateID, providerName)
		if err != nil {
			return err
		}
	}
	if response.Status == "" && p.startLimiter != nil {
		limitResult, limitErr := p.startLimiter.Allow(ctx, "provider_start", providerName)
		if limitErr != nil {
			return fmt.Errorf("apply provider start rate limit: %w", limitErr)
		}
		if !limitResult.Allowed {
			return outbox.Retryable(fmt.Errorf("%w: retry after %s", ErrProviderRateLimited, limitResult.RetryAfter))
		}
	}
	if response.Status == "" && !newReservation {
		_, err = p.pool.Exec(ctx, `
			UPDATE rtc_task SET start_attempts=start_attempts+1, last_checked_at=$2, updated_at=$2
			WHERE conversation_id=$1::uuid`, event.AggregateID, p.now().UTC())
		if err != nil {
			return fmt.Errorf("record provider start attempt: %w", err)
		}
	}
	if response.Status == "" {
		response, err = p.provider.Start(ctx, request)
		if err != nil {
			if voiceprovider.IsTemporary(err) {
				_, _ = p.pool.Exec(ctx, `UPDATE rtc_task SET status='UNKNOWN', updated_at=$2 WHERE conversation_id=$1::uuid AND status IN ('STARTING','UNKNOWN')`, event.AggregateID, p.now().UTC())
				if healthErr := p.health.RecordFailure(ctx, tenantID, providerName, "PROVIDER_START_TEMPORARY"); healthErr != nil {
					return fmt.Errorf("record provider start failure: %v; provider error: %w", healthErr, err)
				}
				return outbox.Retryable(err)
			}
			slog.ErrorContext(ctx, "provider voice chat start rejected",
				"provider", providerName,
				"error", err,
			)
			return p.failRejectedStart(ctx, event.AggregateID, tenantID, "provider_start_rejected", "provider permanently rejected voice chat start")
		}
	}
	if err := p.health.RecordSuccess(ctx, tenantID, providerName); err != nil {
		return fmt.Errorf("record provider start success: %w", err)
	}
	if response.Status != "STARTING" && response.Status != "ACTIVE" {
		return p.failRejectedStart(ctx, event.AggregateID, tenantID, "provider_start_invalid_response", "provider returned an unsupported voice chat start status")
	}
	now := p.now().UTC()
	if response.Status == "STARTING" {
		_, err = p.pool.Exec(ctx, `
			UPDATE rtc_task SET status=CASE WHEN status='ACTIVE' THEN 'ACTIVE' ELSE 'STARTING' END,
				last_checked_at=$2, updated_at=$2 WHERE conversation_id=$1::uuid`, event.AggregateID, now)
		if err != nil {
			return fmt.Errorf("record provider start response: %w", err)
		}
		return nil
	}
	tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer tx.Rollback(context.Background())
	leaseExpiresAt := now.Add(p.initialLeaseTTL)
	if _, err := tx.Exec(ctx, `UPDATE rtc_task SET status='ACTIVE', lease_expires_at=$2,
		last_checked_at=$3, updated_at=$3 WHERE conversation_id=$1::uuid`, event.AggregateID, leaseExpiresAt, now); err != nil {
		return fmt.Errorf("record provider active response: %w", err)
	}
	tag, err := tx.Exec(ctx, `UPDATE conversation SET status='ACTIVE', started_at=COALESCE(started_at,$2),
		lease_expires_at=$3, version=version+1, updated_at=$2
		WHERE id=$1::uuid AND status='STARTING_AGENT'`, event.AggregateID, now, leaseExpiresAt)
	if err != nil {
		return fmt.Errorf("reconcile active conversation: %w", err)
	}
	if tag.RowsAffected() == 1 {
		if _, err := tx.Exec(ctx, `INSERT INTO conversation_state_history (
			conversation_id, from_status, to_status, reason, source
		) VALUES ($1::uuid, 'STARTING_AGENT', 'ACTIVE', 'provider idempotent start response confirmed active task', 'worker')`, event.AggregateID); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (p *StartProcessor) failRejectedStart(ctx context.Context, conversationID, tenantID, failureCode, reason string) error {
	tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer tx.Rollback(context.Background())
	now := p.now().UTC()
	tag, err := tx.Exec(ctx, `
		UPDATE conversation SET status='FAILED',ended_at=$2,lease_expires_at=NULL,
			failure_code=$3,version=version+1,updated_at=$2
		WHERE id=$1::uuid AND status='STARTING_AGENT'`, conversationID, now, failureCode)
	if err != nil {
		return fmt.Errorf("fail permanently rejected provider start: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return tx.Commit(ctx)
	}
	if _, err = tx.Exec(ctx, `UPDATE rtc_task SET status='FAILED',lease_expires_at=NULL,
		last_checked_at=$2,updated_at=$2 WHERE conversation_id=$1::uuid`, conversationID, now); err != nil {
		return fmt.Errorf("fail rejected provider RTC task: %w", err)
	}
	if _, err = tx.Exec(ctx, `INSERT INTO conversation_state_history (
		conversation_id,from_status,to_status,reason,source
	) VALUES ($1::uuid,'STARTING_AGENT','FAILED',$2,'worker')`, conversationID, reason); err != nil {
		return fmt.Errorf("record rejected provider start state: %w", err)
	}
	detail, _ := json.Marshal(map[string]string{"failure_code": failureCode})
	if _, err = tx.Exec(ctx, `INSERT INTO audit_event (
		tenant_id,actor_type,action,resource_type,resource_id,purpose,result,detail
	) VALUES ($1::uuid,'system','conversation.agent_start_failed','conversation',$2::uuid,
		'provider_start','failure',$3::jsonb)`, tenantID, conversationID, detail); err != nil {
		return fmt.Errorf("audit rejected provider start: %w", err)
	}
	return tx.Commit(ctx)
}

func (p *StartProcessor) reserveProviderCapacity(ctx context.Context, conversationID, providerName string) (bool, error) {
	tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return false, err
	}
	defer tx.Rollback(context.Background())
	if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "provider-capacity:"+providerName); err != nil {
		return false, fmt.Errorf("lock provider capacity: %w", err)
	}
	var attempts int
	if err = tx.QueryRow(ctx, `SELECT start_attempts FROM rtc_task WHERE conversation_id=$1::uuid FOR UPDATE`, conversationID).Scan(&attempts); err != nil {
		return false, fmt.Errorf("read provider capacity reservation: %w", err)
	}
	if attempts > 0 {
		return false, tx.Commit(ctx)
	}
	var active int
	if err = tx.QueryRow(ctx, `
		SELECT count(*) FROM rtc_task
		WHERE provider=$1 AND (
			status IN ('ACTIVE','STOPPING') OR
			(status IN ('STARTING','UNKNOWN') AND start_attempts>0)
		)`, providerName).Scan(&active); err != nil {
		return false, fmt.Errorf("count provider capacity: %w", err)
	}
	if active >= p.maxActiveTasks {
		return false, outbox.Retryable(fmt.Errorf("%w: provider=%s active=%d maximum=%d", ErrProviderCapacity, providerName, active, p.maxActiveTasks))
	}
	if _, err = tx.Exec(ctx, `UPDATE rtc_task SET start_attempts=1,last_checked_at=$2,updated_at=$2 WHERE conversation_id=$1::uuid`, conversationID, p.now().UTC()); err != nil {
		return false, fmt.Errorf("reserve provider capacity: %w", err)
	}
	if err = tx.Commit(ctx); err != nil {
		return false, err
	}
	return true, nil
}

type StopProcessor struct {
	pool     *pgxpool.Pool
	provider voiceprovider.Client
	health   *ProviderHealth
	outbox   *outbox.Store
	now      func() time.Time
}

func NewStopProcessor(pool *pgxpool.Pool, provider voiceprovider.Client, health *ProviderHealth) (*StopProcessor, error) {
	if pool == nil || provider == nil || health == nil {
		return nil, errors.New("database pool, voice provider client and provider health are required")
	}
	return &StopProcessor{pool: pool, provider: provider, health: health, outbox: outbox.NewStore(pool), now: time.Now}, nil
}

func (p *StopProcessor) Handle(ctx context.Context, event outbox.Event) error {
	var request voiceprovider.StopRequest
	var conversationStatus conversation.Status
	var taskStatus, tenantID, providerName string
	err := p.pool.QueryRow(ctx, `
		SELECT c.public_id::text, c.status, COALESCE(c.stop_reason,'provider_failure'), r.provider_task_ref, r.room_id, r.status, c.tenant_id::text, r.provider
		FROM conversation c JOIN rtc_task r ON r.conversation_id=c.id
		WHERE c.id=$1::uuid`, event.AggregateID,
	).Scan(&request.ConversationID, &conversationStatus, &request.Reason, &request.TaskRef, &request.RoomID, &taskStatus, &tenantID, &providerName)
	if errors.Is(err, pgx.ErrNoRows) {
		return ErrProviderTaskNotFound
	}
	if err != nil {
		return fmt.Errorf("read provider stop task: %w", err)
	}
	if conversationStatus != conversation.StatusStopping || taskStatus == "COMPLETED" {
		return nil
	}
	if taskStatus != "STOPPING" && taskStatus != "UNKNOWN" && taskStatus != "FAILED" {
		return fmt.Errorf("provider task status %s cannot be stopped", taskStatus)
	}
	now := p.now().UTC()
	_, err = p.pool.Exec(ctx, `
		UPDATE rtc_task SET stop_attempts=stop_attempts+1, last_checked_at=$2, updated_at=$2
		WHERE conversation_id=$1::uuid`, event.AggregateID, now)
	if err != nil {
		return fmt.Errorf("record provider stop attempt: %w", err)
	}
	response, err := p.provider.Stop(ctx, request)
	if err != nil {
		_, _ = p.pool.Exec(ctx, `UPDATE rtc_task SET status='UNKNOWN', updated_at=$2
			WHERE conversation_id=$1::uuid AND status IN ('STOPPING','UNKNOWN','FAILED')`, event.AggregateID, p.now().UTC())
		if voiceprovider.IsTemporary(err) {
			if healthErr := p.health.RecordFailure(ctx, tenantID, providerName, "PROVIDER_STOP_TEMPORARY"); healthErr != nil {
				return fmt.Errorf("record provider stop failure: %v; provider error: %w", healthErr, err)
			}
		}
		if voiceprovider.IsTemporary(err) {
			return outbox.Retryable(err)
		}
		return err
	}
	if err := p.health.RecordSuccess(ctx, tenantID, providerName); err != nil {
		return fmt.Errorf("record provider stop success: %w", err)
	}
	if response.Status != "STOPPING" && response.Status != "COMPLETED" {
		return fmt.Errorf("provider returned unsupported stop status %s", response.Status)
	}
	now = p.now().UTC()
	if response.Status == "STOPPING" {
		_, err = p.pool.Exec(ctx, `UPDATE rtc_task SET status=CASE WHEN status='COMPLETED' THEN 'COMPLETED' ELSE 'STOPPING' END,
			last_checked_at=$2, updated_at=$2 WHERE conversation_id=$1::uuid`, event.AggregateID, now)
		if err != nil {
			return fmt.Errorf("record provider stop response: %w", err)
		}
		return nil
	}
	tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return err
	}
	defer tx.Rollback(context.Background())
	if _, err := tx.Exec(ctx, `UPDATE rtc_task SET status='COMPLETED', lease_expires_at=NULL,
		last_checked_at=$2, updated_at=$2 WHERE conversation_id=$1::uuid`, event.AggregateID, now); err != nil {
		return err
	}
	tag, err := tx.Exec(ctx, `UPDATE conversation SET status='COMPLETED', ended_at=COALESCE(ended_at,$2),
		lease_expires_at=NULL, version=version+1, updated_at=$2 WHERE id=$1::uuid AND status='STOPPING'`, event.AggregateID, now)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 1 {
		if _, err := tx.Exec(ctx, `INSERT INTO conversation_state_history (
			conversation_id, from_status, to_status, reason, source
		) VALUES ($1::uuid, 'STOPPING', 'COMPLETED', 'provider idempotent stop response confirmed completed task', 'worker')`, event.AggregateID); err != nil {
			return err
		}
	}
	deduplicationKey := "content-extraction-v1:" + event.AggregateID
	if _, err := p.outbox.Enqueue(ctx, tx, outbox.NewEvent{
		TenantID: &tenantID, AggregateType: "conversation", AggregateID: event.AggregateID,
		EventType: "conversation.content_extraction_requested", DeduplicationKey: &deduplicationKey,
		Payload: json.RawMessage(`{"generation_version":1}`),
	}); err != nil {
		return fmt.Errorf("enqueue conversation content extraction: %w", err)
	}
	return tx.Commit(ctx)
}

type ProviderEventResult struct {
	ConversationID string              `json:"conversation_id"`
	Status         conversation.Status `json:"status"`
	Duplicate      bool                `json:"duplicate"`
	Ignored        bool                `json:"ignored"`
	TurnsAccepted  int                 `json:"turns_accepted,omitempty"`
	UsageAccepted  int                 `json:"usage_accepted,omitempty"`
}

type ProviderEventProcessor struct {
	pool            *pgxpool.Pool
	outbox          *outbox.Store
	initialLeaseTTL time.Duration
	now             func() time.Time
}

func NewProviderEventProcessor(pool *pgxpool.Pool, initialLeaseTTL time.Duration) (*ProviderEventProcessor, error) {
	if pool == nil || initialLeaseTTL <= 0 {
		return nil, errors.New("database pool and positive initial lease TTL are required")
	}
	return &ProviderEventProcessor{pool: pool, outbox: outbox.NewStore(pool), initialLeaseTTL: initialLeaseTTL, now: time.Now}, nil
}

func (p *ProviderEventProcessor) Process(ctx context.Context, event voiceprovider.CallbackEvent, rawBody []byte) (ProviderEventResult, error) {
	tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return ProviderEventResult{}, err
	}
	defer tx.Rollback(context.Background())
	var internalID, tenantID, subjectID, publicID, taskProvider, retentionMode string
	var currentStatus conversation.Status
	err = tx.QueryRow(ctx, `
		SELECT c.id::text, c.tenant_id::text, c.subject_id::text, c.public_id::text, c.status, r.provider, c.retention_mode
		FROM conversation c JOIN rtc_task r ON r.conversation_id=c.id
		WHERE r.provider_task_ref=$1
		FOR UPDATE OF c, r`, event.TaskRef,
	).Scan(&internalID, &tenantID, &subjectID, &publicID, &currentStatus, &taskProvider, &retentionMode)
	if errors.Is(err, pgx.ErrNoRows) {
		return ProviderEventResult{}, ErrProviderTaskNotFound
	}
	if err != nil {
		return ProviderEventResult{}, fmt.Errorf("read callback task: %w", err)
	}
	if event.Provider != taskProvider || event.ConversationID != publicID {
		return ProviderEventResult{}, ErrProviderTaskNotFound
	}
	digest := sha256.Sum256(rawBody)
	digestText := hex.EncodeToString(digest[:])
	eventID := uuid.NewString()
	sanitized, _ := json.Marshal(map[string]any{
		"event_type": event.EventType, "status": event.Status, "occurred_at": event.OccurredAt, "turn_count": len(event.Turns), "usage_count": len(event.Usage),
	})
	tag, err := tx.Exec(ctx, `
		INSERT INTO provider_event (
			id, tenant_id, conversation_id, provider, provider_event_key,
			provider_task_ref, event_type, occurred_at, payload_digest, sanitized_payload
		) VALUES ($1::uuid, $2::uuid, $3::uuid, $4, $5, $6, $7, $8, $9, $10::jsonb)
		ON CONFLICT (provider, provider_event_key) DO NOTHING`,
		eventID, tenantID, internalID, event.Provider, event.EventID, event.TaskRef,
		event.EventType, event.OccurredAt, digestText, sanitized,
	)
	if err != nil {
		return ProviderEventResult{}, fmt.Errorf("store provider event: %w", err)
	}
	if tag.RowsAffected() == 0 {
		var storedDigest string
		if err := tx.QueryRow(ctx, `SELECT payload_digest FROM provider_event WHERE provider=$1 AND provider_event_key=$2`, event.Provider, event.EventID).Scan(&storedDigest); err != nil {
			return ProviderEventResult{}, err
		}
		if storedDigest != digestText {
			return ProviderEventResult{}, ErrProviderEventConflict
		}
		if err := tx.Commit(ctx); err != nil {
			return ProviderEventResult{}, err
		}
		return ProviderEventResult{ConversationID: publicID, Status: currentStatus, Duplicate: true}, nil
	}

	result := ProviderEventResult{ConversationID: publicID, Status: currentStatus}
	if event.EventType == "voice_chat.usage" && event.Status == "FINAL" {
		metrics := make([]billing.UsageMetricInput, 0, len(event.Usage))
		for _, metric := range event.Usage {
			metrics = append(metrics, billing.UsageMetricInput{UsageCategory: metric.UsageCategory, MetricCode: metric.MetricCode, Quantity: metric.Quantity, Unit: metric.Unit})
		}
		recorded, usageErr := billing.RecordProviderUsageTx(ctx, tx, tenantID, internalID, eventID, event.Provider, event.EventID, event.OccurredAt, metrics)
		if errors.Is(usageErr, billing.ErrUsageConflict) {
			return ProviderEventResult{}, ErrProviderEventConflict
		}
		if usageErr != nil {
			return ProviderEventResult{}, usageErr
		}
		if _, usageErr = tx.Exec(ctx, `UPDATE provider_event SET processing_status='processed',processed_at=$2 WHERE id=$1::uuid`, eventID, p.now().UTC()); usageErr != nil {
			return ProviderEventResult{}, usageErr
		}
		if usageErr = tx.Commit(ctx); usageErr != nil {
			return ProviderEventResult{}, usageErr
		}
		result.UsageAccepted = len(recorded)
		return result, nil
	}
	if event.EventType == "voice_chat.transcript.final" && event.Status == "FINAL" {
		return p.processTranscriptEvent(ctx, tx, eventID, internalID, tenantID, subjectID, retentionMode, event, result)
	}
	if event.EventType == "voice_chat.stopped" && event.Status == "COMPLETED" {
		return p.completeStoppedEvent(ctx, tx, eventID, internalID, tenantID, currentStatus, result)
	}
	if event.EventType == "voice_chat.failed" && event.Status == "FAILED" {
		return p.processFailedEvent(ctx, tx, eventID, internalID, currentStatus, result)
	}
	if event.EventType != "voice_chat.started" || event.Status != "ACTIVE" || currentStatus != conversation.StatusStartingAgent {
		result.Ignored = true
		if _, err := tx.Exec(ctx, `UPDATE provider_event SET processing_status='ignored', processed_at=$2 WHERE id=$1::uuid`, eventID, p.now().UTC()); err != nil {
			return ProviderEventResult{}, err
		}
		if err := tx.Commit(ctx); err != nil {
			return ProviderEventResult{}, err
		}
		return result, nil
	}
	if err := conversation.ValidateTransition(currentStatus, conversation.StatusActive); err != nil {
		return ProviderEventResult{}, err
	}
	now := p.now().UTC()
	_, err = tx.Exec(ctx, `
		UPDATE conversation SET status='ACTIVE', started_at=COALESCE(started_at,$2),
			lease_expires_at=$3, version=version+1, updated_at=$2
		WHERE id=$1::uuid AND status='STARTING_AGENT'`, internalID, now, now.Add(p.initialLeaseTTL))
	if err != nil {
		return ProviderEventResult{}, fmt.Errorf("activate conversation from provider event: %w", err)
	}
	_, err = tx.Exec(ctx, `
		UPDATE rtc_task SET status='ACTIVE', lease_expires_at=$2, last_checked_at=$3, updated_at=$3
		WHERE conversation_id=$1::uuid`, internalID, now.Add(p.initialLeaseTTL), now)
	if err != nil {
		return ProviderEventResult{}, fmt.Errorf("activate RTC task: %w", err)
	}
	_, err = tx.Exec(ctx, `
		INSERT INTO conversation_state_history (
			conversation_id, from_status, to_status, reason, source, provider_event_id
		) VALUES ($1::uuid, 'STARTING_AGENT', 'ACTIVE', 'provider confirmed voice chat start', 'provider', $2::uuid)`, internalID, eventID)
	if err != nil {
		return ProviderEventResult{}, fmt.Errorf("record provider active state: %w", err)
	}
	_, err = tx.Exec(ctx, `UPDATE provider_event SET processing_status='processed', processed_at=$2 WHERE id=$1::uuid`, eventID, now)
	if err != nil {
		return ProviderEventResult{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return ProviderEventResult{}, err
	}
	result.Status = conversation.StatusActive
	return result, nil
}

func (p *ProviderEventProcessor) processTranscriptEvent(
	ctx context.Context,
	tx pgx.Tx,
	eventID, conversationID, tenantID, subjectID, retentionMode string,
	event voiceprovider.CallbackEvent,
	result ProviderEventResult,
) (ProviderEventResult, error) {
	now := p.now().UTC()
	if retentionMode == "none" {
		result.Ignored = true
		if _, err := tx.Exec(ctx, `UPDATE provider_event SET processing_status='ignored', processed_at=$2 WHERE id=$1::uuid`, eventID, now); err != nil {
			return ProviderEventResult{}, err
		}
		if err := tx.Commit(ctx); err != nil {
			return ProviderEventResult{}, err
		}
		return result, nil
	}
	for _, turn := range event.Turns {
		providerRef := turn.ProviderRef
		if providerRef == "" {
			providerRef = turn.TurnKey
		}
		tag, err := tx.Exec(ctx, `
			INSERT INTO conversation_turn (
				tenant_id,subject_id,conversation_id,turn_key,sequence_no,speaker,start_at,end_at,
				final_text,interrupted,provider_ref,source_type
			) VALUES ($1::uuid,$2::uuid,$3::uuid,$4,$5,$6,$7,$8,$9,$10,$11,'provider')
			ON CONFLICT (conversation_id,turn_key) DO NOTHING`, tenantID, subjectID, conversationID,
			turn.TurnKey, turn.SequenceNo, turn.Speaker, turn.StartAt, turn.EndAt, turn.FinalText, turn.Interrupted, providerRef)
		if err != nil {
			return ProviderEventResult{}, fmt.Errorf("store final transcript turn: %w", err)
		}
		if tag.RowsAffected() == 0 {
			var sequenceNo int64
			var speaker, finalText string
			var startAt, endAt time.Time
			var interrupted bool
			if err := tx.QueryRow(ctx, `
				SELECT sequence_no,speaker,start_at,end_at,final_text,interrupted
				FROM conversation_turn WHERE conversation_id=$1::uuid AND turn_key=$2`, conversationID, turn.TurnKey).Scan(
				&sequenceNo, &speaker, &startAt, &endAt, &finalText, &interrupted); err != nil {
				return ProviderEventResult{}, err
			}
			if sequenceNo != turn.SequenceNo || speaker != turn.Speaker || !startAt.Equal(turn.StartAt) ||
				!endAt.Equal(turn.EndAt) || finalText != turn.FinalText || interrupted != turn.Interrupted {
				return ProviderEventResult{}, ErrProviderEventConflict
			}
			continue
		}
		result.TurnsAccepted++
	}
	// A provider may deliver its final transcript after the stopped callback.
	// In that ordering the original extraction job may already have run before
	// any turns existed, so accepted late turns must create a new idempotent
	// generation request rather than leaving the conversation without content.
	if result.TurnsAccepted > 0 && conversation.IsTerminal(result.Status) {
		var generationVersion int64
		if err := tx.QueryRow(ctx, `SELECT COALESCE(max(sequence_no),1) FROM conversation_turn WHERE conversation_id=$1::uuid`, conversationID).Scan(&generationVersion); err != nil {
			return ProviderEventResult{}, fmt.Errorf("read late transcript generation: %w", err)
		}
		digest := sha256.Sum256([]byte(event.Provider + ":" + event.EventID))
		deduplicationKey := "content-extraction-v1:" + conversationID + ":provider:" + hex.EncodeToString(digest[:16])
		payload, err := json.Marshal(map[string]int64{"generation_version": generationVersion})
		if err != nil {
			return ProviderEventResult{}, err
		}
		if _, err := p.outbox.Enqueue(ctx, tx, outbox.NewEvent{
			TenantID: &tenantID, AggregateType: "conversation", AggregateID: conversationID,
			EventType: "conversation.content_extraction_requested", DeduplicationKey: &deduplicationKey,
			Payload: payload,
		}); err != nil {
			return ProviderEventResult{}, fmt.Errorf("enqueue late provider transcript extraction: %w", err)
		}
	}
	if _, err := tx.Exec(ctx, `UPDATE provider_event SET processing_status='processed', processed_at=$2 WHERE id=$1::uuid`, eventID, now); err != nil {
		return ProviderEventResult{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return ProviderEventResult{}, err
	}
	return result, nil
}

func (p *ProviderEventProcessor) completeStoppedEvent(
	ctx context.Context,
	tx pgx.Tx,
	eventID, internalID, tenantID string,
	currentStatus conversation.Status,
	result ProviderEventResult,
) (ProviderEventResult, error) {
	if conversation.IsTerminal(currentStatus) || currentStatus == conversation.StatusRoomIssued || currentStatus == conversation.StatusPreparing {
		result.Ignored = true
		if _, err := tx.Exec(ctx, `UPDATE provider_event SET processing_status='ignored', processed_at=$2 WHERE id=$1::uuid`, eventID, p.now().UTC()); err != nil {
			return ProviderEventResult{}, err
		}
		if err := tx.Commit(ctx); err != nil {
			return ProviderEventResult{}, err
		}
		return result, nil
	}
	now := p.now().UTC()
	if currentStatus == conversation.StatusStartingAgent || currentStatus == conversation.StatusActive {
		if err := conversation.ValidateTransition(currentStatus, conversation.StatusStopping); err != nil {
			return ProviderEventResult{}, err
		}
		if _, err := tx.Exec(ctx, `UPDATE conversation SET status='STOPPING', lease_expires_at=NULL,
			version=version+1, updated_at=$2 WHERE id=$1::uuid`, internalID, now); err != nil {
			return ProviderEventResult{}, err
		}
		if _, err := tx.Exec(ctx, `
			INSERT INTO conversation_state_history (
				conversation_id, from_status, to_status, reason, source, provider_event_id
			) VALUES ($1::uuid, $2, 'STOPPING', 'provider ended task before platform stop completed', 'provider', $3::uuid)`,
			internalID, currentStatus, eventID); err != nil {
			return ProviderEventResult{}, err
		}
	}
	if _, err := tx.Exec(ctx, `UPDATE conversation SET status='COMPLETED', ended_at=COALESCE(ended_at,$2),
		lease_expires_at=NULL, stop_reason=COALESCE(stop_reason,'provider_ended'),
		version=version+1, updated_at=$2 WHERE id=$1::uuid`, internalID, now); err != nil {
		return ProviderEventResult{}, err
	}
	if _, err := tx.Exec(ctx, `UPDATE rtc_task SET status='COMPLETED', lease_expires_at=NULL,
		last_checked_at=$2, updated_at=$2 WHERE conversation_id=$1::uuid`, internalID, now); err != nil {
		return ProviderEventResult{}, err
	}
	if _, err := tx.Exec(ctx, `
		INSERT INTO conversation_state_history (
			conversation_id, from_status, to_status, reason, source, provider_event_id
		) VALUES ($1::uuid, 'STOPPING', 'COMPLETED', 'provider confirmed voice chat stop', 'provider', $2::uuid)`,
		internalID, eventID); err != nil {
		return ProviderEventResult{}, err
	}
	if _, err := tx.Exec(ctx, `UPDATE provider_event SET processing_status='processed', processed_at=$2 WHERE id=$1::uuid`, eventID, now); err != nil {
		return ProviderEventResult{}, err
	}
	deduplicationKey := "content-extraction-v1:" + internalID
	if _, err := p.outbox.Enqueue(ctx, tx, outbox.NewEvent{
		TenantID: &tenantID, AggregateType: "conversation", AggregateID: internalID,
		EventType: "conversation.content_extraction_requested", DeduplicationKey: &deduplicationKey,
		Payload: json.RawMessage(`{"generation_version":1}`),
	}); err != nil {
		return ProviderEventResult{}, fmt.Errorf("enqueue conversation content extraction: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return ProviderEventResult{}, err
	}
	result.Status = conversation.StatusCompleted
	return result, nil
}

func (p *ProviderEventProcessor) processFailedEvent(
	ctx context.Context,
	tx pgx.Tx,
	eventID, internalID string,
	currentStatus conversation.Status,
	result ProviderEventResult,
) (ProviderEventResult, error) {
	now := p.now().UTC()
	switch currentStatus {
	case conversation.StatusStartingAgent:
		if _, err := tx.Exec(ctx, `UPDATE conversation SET status='FAILED', ended_at=$2,
			lease_expires_at=NULL, failure_code='provider_start_failed', version=version+1, updated_at=$2
			WHERE id=$1::uuid`, internalID, now); err != nil {
			return ProviderEventResult{}, err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO conversation_state_history (
			conversation_id, from_status, to_status, reason, source, provider_event_id
		) VALUES ($1::uuid, 'STARTING_AGENT', 'FAILED', 'provider reported voice chat failure', 'provider', $2::uuid)`, internalID, eventID); err != nil {
			return ProviderEventResult{}, err
		}
		result.Status = conversation.StatusFailed
	case conversation.StatusActive:
		if _, err := tx.Exec(ctx, `UPDATE conversation SET status='STOPPING', lease_expires_at=NULL,
			failure_code='provider_runtime_failed', version=version+1, updated_at=$2 WHERE id=$1::uuid`, internalID, now); err != nil {
			return ProviderEventResult{}, err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO conversation_state_history (
			conversation_id, from_status, to_status, reason, source, provider_event_id
		) VALUES ($1::uuid, 'ACTIVE', 'STOPPING', 'provider reported runtime failure', 'provider', $2::uuid)`, internalID, eventID); err != nil {
			return ProviderEventResult{}, err
		}
		result.Status = conversation.StatusStopping
	case conversation.StatusStopping:
		if _, err := tx.Exec(ctx, `UPDATE conversation SET failure_code=COALESCE(failure_code,'provider_stop_failed'), updated_at=$2 WHERE id=$1::uuid`, internalID, now); err != nil {
			return ProviderEventResult{}, err
		}
	default:
		result.Ignored = true
	}
	if !result.Ignored {
		if _, err := tx.Exec(ctx, `UPDATE rtc_task SET status='FAILED', lease_expires_at=NULL,
			last_checked_at=$2, updated_at=$2 WHERE conversation_id=$1::uuid`, internalID, now); err != nil {
			return ProviderEventResult{}, err
		}
	}
	processingStatus := "processed"
	if result.Ignored {
		processingStatus = "ignored"
	}
	if _, err := tx.Exec(ctx, `UPDATE provider_event SET processing_status=$2, processed_at=$3 WHERE id=$1::uuid`, eventID, processingStatus, now); err != nil {
		return ProviderEventResult{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return ProviderEventResult{}, err
	}
	return result, nil
}