story/internal/diagnostics/service.go

package diagnostics

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/idempotency"
	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/outbox"

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

var (
	ErrInvalid              = errors.New("diagnostic request is invalid")
	ErrAccessDenied         = errors.New("diagnostic access is denied")
	ErrNotFound             = errors.New("diagnostic resource was not found")
	ErrStateConflict        = errors.New("diagnostic action conflicts with current state")
	requestIDPattern        = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.:/-]{2,127}$`)
	searchFieldAllowed      = map[string]bool{"device_id": true, "conversation_id": true, "room_id": true, "provider_task_ref": true, "request_id": true}
	retryableJobTypeAllowed = map[string]bool{
		"platform.noop":                                true,
		"conversation.room_issued":                     true,
		"conversation.agent_start_requested":           true,
		"conversation.agent_start_reconcile_requested": true,
		"conversation.agent_stop_requested":            true,
		"conversation.agent_stop_reconcile_requested":  true,
		"conversation.content_extraction_requested":    true,
		"call.story_extraction_requested":              true,
		"memory.index_requested":                       true,
		"memory.delete_requested":                      true,
		"privacy.access_requested":                     true,
		"privacy.delete_requested":                     true,
		"export.generate_requested":                    true,
		"media.verify_requested":                       true,
		"alert.webhook_test_requested":                 true,
	}
)

type Record struct {
	DeviceID                  string     `json:"device_id"`
	DeviceStatus              string     `json:"device_status"`
	DeviceOnline              bool       `json:"device_online"`
	FirmwareVersion           string     `json:"firmware_version,omitempty"`
	DesiredConfigRevision     int64      `json:"desired_config_revision"`
	ReportedConfigRevision    *int64     `json:"reported_config_revision,omitempty"`
	LastDeviceErrorCode       string     `json:"last_device_error_code,omitempty"`
	LastDeviceErrorAt         *time.Time `json:"last_device_error_at,omitempty"`
	ConversationID            string     `json:"conversation_id,omitempty"`
	ConversationStatus        string     `json:"conversation_status,omitempty"`
	ConversationFailureCode   string     `json:"conversation_failure_code,omitempty"`
	AllocationExpiresAt       *time.Time `json:"allocation_expires_at,omitempty"`
	CredentialIssuedAt        *time.Time `json:"credential_issued_at,omitempty"`
	CredentialExpiresAt       *time.Time `json:"credential_expires_at,omitempty"`
	RoomID                    string     `json:"room_id,omitempty"`
	Provider                  string     `json:"provider,omitempty"`
	ProviderTaskRef           string     `json:"provider_task_ref,omitempty"`
	ProviderTaskStatus        string     `json:"provider_task_status,omitempty"`
	ProviderStartAttempts     int        `json:"provider_start_attempts"`
	ProviderStopAttempts      int        `json:"provider_stop_attempts"`
	ProviderNotificationCount int64      `json:"provider_notification_count"`
	LastProviderEvent         string     `json:"last_provider_event,omitempty"`
	LastProviderEventStatus   string     `json:"last_provider_event_status,omitempty"`
	LastProviderEventAt       *time.Time `json:"last_provider_event_at,omitempty"`
	LastRequestID             string     `json:"last_request_id,omitempty"`
	Timings                   *Timings   `json:"timings,omitempty"`
}

type Timings struct {
	ConversationCreatedAt time.Time  `json:"conversation_created_at"`
	RoomIssuedAt          *time.Time `json:"room_issued_at,omitempty"`
	DeviceActivatedAt     *time.Time `json:"device_activated_at,omitempty"`
	AgentActiveAt         *time.Time `json:"agent_active_at,omitempty"`
	StopRequestedAt       *time.Time `json:"stop_requested_at,omitempty"`
	ConversationEndedAt   *time.Time `json:"conversation_ended_at,omitempty"`
	RoomIssueMS           *int64     `json:"room_issue_ms,omitempty"`
	DeviceJoinMS          *int64     `json:"device_join_ms,omitempty"`
	AgentStartMS          *int64     `json:"agent_start_ms,omitempty"`
	AgentStopMS           *int64     `json:"agent_stop_ms,omitempty"`
	TotalMS               *int64     `json:"total_ms,omitempty"`
}

type ActionResult struct {
	Action             string `json:"action"`
	DeviceID           string `json:"device_id"`
	ConversationID     string `json:"conversation_id,omitempty"`
	ConversationStatus string `json:"conversation_status,omitempty"`
	CommandID          string `json:"command_id,omitempty"`
	CommandStatus      string `json:"command_status,omitempty"`
	JobID              string `json:"job_id,omitempty"`
	JobType            string `json:"job_type,omitempty"`
	ManualRetryCount   int    `json:"manual_retry_count,omitempty"`
	Queued             bool   `json:"queued"`
}

type Job struct {
	ID                string     `json:"id"`
	EventType         string     `json:"event_type"`
	AggregateType     string     `json:"aggregate_type"`
	Status            string     `json:"status"`
	AttemptCount      int        `json:"attempt_count"`
	ManualRetryCount  int        `json:"manual_retry_count"`
	LastErrorCode     string     `json:"last_error_code,omitempty"`
	LastManualRetryAt *time.Time `json:"last_manual_retry_at,omitempty"`
	CreatedAt         time.Time  `json:"created_at"`
}

type Bundle struct {
	ID              string    `json:"id"`
	GeneratedAt     time.Time `json:"generated_at"`
	SearchField     string    `json:"search_field"`
	SearchValue     string    `json:"search_value"`
	RedactionPolicy string    `json:"redaction_policy"`
	Items           []Record  `json:"items"`
}

type ExternalDependency struct {
	ServiceName         string     `json:"service_name"`
	Dependency          string     `json:"dependency"`
	Status              string     `json:"status"`
	CheckedAt           time.Time  `json:"checked_at"`
	LastSuccessAt       *time.Time `json:"last_success_at,omitempty"`
	LastFailureAt       *time.Time `json:"last_failure_at,omitempty"`
	LastErrorCode       string     `json:"last_error_code,omitempty"`
	LatencyMS           int64      `json:"latency_ms"`
	ProbeIntervalMS     int64      `json:"probe_interval_ms"`
	ConsecutiveFailures int        `json:"consecutive_failures"`
}

type Service struct {
	pool   *pgxpool.Pool
	idem   *idempotency.Executor
	audit  *audit.Repository
	outbox *outbox.Store
}

func NewService(pool *pgxpool.Pool, idem *idempotency.Executor, auditRepository *audit.Repository) (*Service, error) {
	if pool == nil || idem == nil || auditRepository == nil {
		return nil, errors.New("diagnostic service dependencies are required")
	}
	return &Service{pool: pool, idem: idem, audit: auditRepository, outbox: outbox.NewStore(pool)}, nil
}

type actor struct{ tenantID, userID, userPublicID string }

func (s *Service) resolveActor(ctx context.Context, principal identity.Principal, tenantPublicID string, roles ...string) (actor, error) {
	if _, err := uuid.Parse(tenantPublicID); err != nil {
		return actor{}, ErrInvalid
	}
	var result actor
	var role string
	err := s.pool.QueryRow(ctx, `SELECT t.id::text,ua.id::text,ua.public_id::text,ua.platform_role FROM tenant t JOIN user_account ua ON ua.tenant_id=t.id AND ua.status='active' AND ua.account_kind='operator' WHERE t.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, tenantPublicID, principal.Issuer, principal.Subject).Scan(&result.tenantID, &result.userID, &result.userPublicID, &role)
	if errors.Is(err, pgx.ErrNoRows) {
		return actor{}, ErrAccessDenied
	}
	if err != nil {
		return actor{}, err
	}
	for _, allowed := range roles {
		if role == allowed {
			return result, nil
		}
	}
	return actor{}, ErrAccessDenied
}

func (s *Service) Search(ctx context.Context, principal identity.Principal, tenantPublicID, field, value string) ([]Record, error) {
	field, value = strings.TrimSpace(field), strings.TrimSpace(value)
	if !searchFieldAllowed[field] || value == "" || len(value) > 200 {
		return nil, ErrInvalid
	}
	if field == "device_id" || field == "conversation_id" {
		if _, err := uuid.Parse(value); err != nil {
			return nil, ErrInvalid
		}
	} else if !requestIDPattern.MatchString(value) {
		return nil, ErrInvalid
	}
	currentActor, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin")
	if err != nil {
		return nil, err
	}
	condition := map[string]string{
		"device_id":         "d.public_id=$2::uuid",
		"conversation_id":   "c.public_id=$2::uuid",
		"room_id":           "rt.room_id=$2",
		"provider_task_ref": "COALESCE(rt.provider_task_ref,hs.provider_session_ref)=$2",
		"request_id":        "EXISTS(SELECT 1 FROM audit_event ae WHERE ae.tenant_id=c.tenant_id AND ae.request_id=$2 AND ae.resource_id IN(c.id,c.public_id,d.id,d.public_id))",
	}[field]
	query := diagnosticSelect + " WHERE d.tenant_id=$1::uuid AND " + condition + " ORDER BY c.created_at DESC NULLS LAST LIMIT 50"
	rows, err := s.pool.Query(ctx, query, currentActor.tenantID, value)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []Record{}
	for rows.Next() {
		item, scanErr := scanRecord(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Service) ListExternalDependencies(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]ExternalDependency, error) {
	if _, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin"); err != nil {
		return nil, err
	}
	rows, err := s.pool.Query(ctx, `SELECT service_name,dependency,status,checked_at,last_success_at,last_failure_at,COALESCE(last_error_code,''),latency_ms,probe_interval_ms,consecutive_failures FROM external_dependency_probe_state ORDER BY service_name,dependency`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []ExternalDependency{}
	for rows.Next() {
		var item ExternalDependency
		if err = rows.Scan(&item.ServiceName, &item.Dependency, &item.Status, &item.CheckedAt, &item.LastSuccessAt, &item.LastFailureAt, &item.LastErrorCode, &item.LatencyMS, &item.ProbeIntervalMS, &item.ConsecutiveFailures); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Service) StopConversation(ctx context.Context, principal identity.Principal, tenantPublicID, conversationPublicID, key, requestID, reason string) (ActionResult, bool, error) {
	reason = strings.TrimSpace(reason)
	if _, err := uuid.Parse(conversationPublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
		return ActionResult{}, false, ErrInvalid
	}
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return ActionResult{}, false, err
	}
	body, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": reason})
	result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/conversations/stop", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		var internalID, deviceID, status, runtimeStatus, runtimeType string
		err := tx.QueryRow(operationContext, `SELECT c.id::text,d.public_id::text,c.status,c.runtime_type,COALESCE(rt.status,hs.status,'') FROM conversation c JOIN device d ON d.id=c.device_id LEFT JOIN rtc_task rt ON rt.conversation_id=c.id LEFT JOIN hardware_agent_session hs ON hs.conversation_id=c.id WHERE c.tenant_id=$1::uuid AND c.public_id=$2::uuid FOR UPDATE OF c`, a.tenantID, conversationPublicID).Scan(&internalID, &deviceID, &status, &runtimeType, &runtimeStatus)
		if errors.Is(err, pgx.ErrNoRows) {
			return idempotency.Result{}, ErrNotFound
		}
		if err != nil {
			return idempotency.Result{}, err
		}
		queued := false
		fromStatus := status
		switch status {
		case "COMPLETED", "FAILED", "EXPIRED", "STOPPING":
		case "PREPARING", "ROOM_ISSUED":
			status = "EXPIRED"
			if _, err = tx.Exec(operationContext, `UPDATE conversation SET status='EXPIRED',stop_reason='ops_stop_before_start',ended_at=now(),lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, internalID); err == nil && runtimeType == "volc_hardware_agent" {
				_, err = tx.Exec(operationContext, `UPDATE hardware_agent_session SET status='EXPIRED',updated_at=now() WHERE conversation_id=$1::uuid`, internalID)
			} else if err == nil {
				_, err = tx.Exec(operationContext, `UPDATE rtc_task SET status='COMPLETED',lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`, internalID)
			}
		case "STARTING_AGENT", "ACTIVE":
			status, queued = "STOPPING", runtimeType != "volc_hardware_agent"
			if _, err = tx.Exec(operationContext, `UPDATE conversation SET status='STOPPING',stop_reason='ops_stop',lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, internalID); err == nil && runtimeType == "volc_hardware_agent" {
				_, err = tx.Exec(operationContext, `UPDATE hardware_agent_session SET status='STOPPING',updated_at=now() WHERE conversation_id=$1::uuid`, internalID)
			} else if err == nil {
				_, err = tx.Exec(operationContext, `UPDATE rtc_task SET status=CASE WHEN status='ACTIVE' THEN 'STOPPING' ELSE 'UNKNOWN' END,lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`, internalID)
			}
			if err == nil && runtimeType != "volc_hardware_agent" {
				dedupe := "ops-stop:" + internalID
				payload, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": "ops_stop"})
				_, err = s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{TenantID: &a.tenantID, AggregateType: "conversation", AggregateID: internalID, EventType: "conversation.agent_stop_requested", DeduplicationKey: &dedupe, Payload: payload})
			}
		default:
			return idempotency.Result{}, ErrStateConflict
		}
		if err != nil {
			return idempotency.Result{}, err
		}
		if status != fromStatus {
			if _, err = tx.Exec(operationContext, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source) VALUES($1::uuid,$2,$3,$4,'operator')`, internalID, fromStatus, status, "ops diagnostics: "+reason); err != nil {
				return idempotency.Result{}, err
			}
		}
		if err = s.audit.Record(operationContext, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.conversation.stop", ResourceType: "conversation", ResourceID: &conversationPublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"runtime_type": runtimeType, "from_runtime_status": runtimeStatus, "to_status": status}}); err != nil {
			return idempotency.Result{}, err
		}
		output := ActionResult{Action: "stop_conversation", DeviceID: deviceID, ConversationID: conversationPublicID, ConversationStatus: status, Queued: queued}
		encoded, _ := json.Marshal(output)
		return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &internalID}, nil
	})
	if err != nil {
		return ActionResult{}, false, err
	}
	var output ActionResult
	if err = json.Unmarshal(result.Body, &output); err != nil {
		return ActionResult{}, false, err
	}
	return output, !result.Replayed, nil
}

func (s *Service) ListDeadJobs(ctx context.Context, principal identity.Principal, tenantPublicID string) ([]Job, error) {
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin")
	if err != nil {
		return nil, err
	}
	rows, err := s.pool.Query(ctx, `SELECT public_id::text,event_type,aggregate_type,status,attempt_count,manual_retry_count,COALESCE(last_error_code,''),last_manual_retry_at,created_at FROM outbox_event WHERE tenant_id=$1::uuid AND status='dead' ORDER BY created_at DESC LIMIT 100`, a.tenantID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []Job{}
	for rows.Next() {
		var item Job
		if err := rows.Scan(&item.ID, &item.EventType, &item.AggregateType, &item.Status, &item.AttemptCount, &item.ManualRetryCount, &item.LastErrorCode, &item.LastManualRetryAt, &item.CreatedAt); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Service) RetryDeadJob(ctx context.Context, principal identity.Principal, tenantPublicID, jobPublicID, key, requestID, reason string) (ActionResult, bool, error) {
	reason = strings.TrimSpace(reason)
	if _, err := uuid.Parse(jobPublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
		return ActionResult{}, false, ErrInvalid
	}
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return ActionResult{}, false, err
	}
	body, _ := json.Marshal(map[string]string{"job_id": jobPublicID, "reason": reason})
	result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/jobs/retry", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) {
		var internalID, eventType, status string
		var manualRetries int
		if queryErr := tx.QueryRow(op, `SELECT id::text,event_type,status,manual_retry_count FROM outbox_event WHERE tenant_id=$1::uuid AND public_id=$2::uuid FOR UPDATE`, a.tenantID, jobPublicID).Scan(&internalID, &eventType, &status, &manualRetries); errors.Is(queryErr, pgx.ErrNoRows) {
			return idempotency.Result{}, ErrNotFound
		} else if queryErr != nil {
			return idempotency.Result{}, queryErr
		}
		if status != "dead" {
			return idempotency.Result{}, ErrStateConflict
		}
		if !retryableJobTypeAllowed[eventType] {
			return idempotency.Result{}, ErrStateConflict
		}
		manualRetries++
		if _, updateErr := tx.Exec(op, `UPDATE outbox_event SET status='pending',attempt_count=0,available_at=now(),locked_at=NULL,locked_by=NULL,dead_at=NULL,last_error_code=NULL,manual_retry_count=$2,last_manual_retry_at=now(),last_manual_retry_by=$3::uuid,updated_at=now() WHERE id=$1::uuid`, internalID, manualRetries, a.userID); updateErr != nil {
			return idempotency.Result{}, updateErr
		}
		if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.job.retry", ResourceType: "outbox_job", ResourceID: &jobPublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"event_type": eventType, "from_status": status, "to_status": "pending", "manual_retry_count": manualRetries}}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		output := ActionResult{Action: "retry_job", JobID: jobPublicID, JobType: eventType, ManualRetryCount: manualRetries, Queued: true}
		encoded, _ := json.Marshal(output)
		return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &internalID}, nil
	})
	if err != nil {
		return ActionResult{}, false, err
	}
	var output ActionResult
	if err = json.Unmarshal(result.Body, &output); err != nil {
		return ActionResult{}, false, err
	}
	return output, !result.Replayed, nil
}

func (s *Service) RevokeDevice(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, key, requestID, reason string) (ActionResult, bool, error) {
	reason = strings.TrimSpace(reason)
	if _, err := uuid.Parse(devicePublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
		return ActionResult{}, false, ErrInvalid
	}
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_admin")
	if err != nil {
		return ActionResult{}, false, err
	}
	body, _ := json.Marshal(map[string]string{"device_id": devicePublicID, "reason": reason})
	result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/devices/revoke", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) {
		var deviceID, oldStatus string
		if queryErr := tx.QueryRow(op, `SELECT id::text,status FROM device WHERE tenant_id=$1::uuid AND public_id=$2::uuid FOR UPDATE`, a.tenantID, devicePublicID).Scan(&deviceID, &oldStatus); errors.Is(queryErr, pgx.ErrNoRows) {
			return idempotency.Result{}, ErrNotFound
		} else if queryErr != nil {
			return idempotency.Result{}, queryErr
		}
		if oldStatus != "revoked" {
			if _, updateErr := tx.Exec(op, `UPDATE device SET status='revoked',revoked_at=COALESCE(revoked_at,now()),version=version+1,updated_at=now() WHERE id=$1::uuid`, deviceID); updateErr != nil {
				return idempotency.Result{}, updateErr
			}
		}

		var conversationInternalID, conversationPublicID, conversationStatus, runtimeStatus, runtimeType string
		queued := false
		conversationQueryErr := tx.QueryRow(op, `SELECT c.id::text,c.public_id::text,c.status,c.runtime_type,COALESCE(rt.status,hs.status,'') FROM conversation c LEFT JOIN rtc_task rt ON rt.conversation_id=c.id LEFT JOIN hardware_agent_session hs ON hs.conversation_id=c.id WHERE c.tenant_id=$1::uuid AND c.device_id=$2::uuid AND c.status NOT IN('COMPLETED','FAILED','EXPIRED') ORDER BY c.created_at DESC LIMIT 1 FOR UPDATE OF c`, a.tenantID, deviceID).Scan(&conversationInternalID, &conversationPublicID, &conversationStatus, &runtimeType, &runtimeStatus)
		if conversationQueryErr != nil && !errors.Is(conversationQueryErr, pgx.ErrNoRows) {
			return idempotency.Result{}, conversationQueryErr
		}
		if conversationQueryErr == nil {
			fromStatus := conversationStatus
			switch conversationStatus {
			case "PREPARING", "ROOM_ISSUED":
				conversationStatus = "EXPIRED"
				if _, updateErr := tx.Exec(op, `UPDATE conversation SET status='EXPIRED',stop_reason='device_revoked_before_start',ended_at=now(),lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, conversationInternalID); updateErr != nil {
					return idempotency.Result{}, updateErr
				}
				query := `UPDATE rtc_task SET status='COMPLETED',lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`
				if runtimeType == "volc_hardware_agent" {
					query = `UPDATE hardware_agent_session SET status='EXPIRED',updated_at=now() WHERE conversation_id=$1::uuid`
				}
				if _, updateErr := tx.Exec(op, query, conversationInternalID); updateErr != nil {
					return idempotency.Result{}, updateErr
				}
			case "STARTING_AGENT", "ACTIVE":
				conversationStatus, queued = "STOPPING", runtimeType != "volc_hardware_agent"
				if _, updateErr := tx.Exec(op, `UPDATE conversation SET status='STOPPING',stop_reason='device_revoked',lease_expires_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, conversationInternalID); updateErr != nil {
					return idempotency.Result{}, updateErr
				}
				query := `UPDATE rtc_task SET status=CASE WHEN status='ACTIVE' THEN 'STOPPING' ELSE 'UNKNOWN' END,lease_expires_at=NULL,updated_at=now() WHERE conversation_id=$1::uuid`
				if runtimeType == "volc_hardware_agent" {
					query = `UPDATE hardware_agent_session SET status='STOPPING',updated_at=now() WHERE conversation_id=$1::uuid`
				}
				if _, updateErr := tx.Exec(op, query, conversationInternalID); updateErr != nil {
					return idempotency.Result{}, updateErr
				}
				if runtimeType != "volc_hardware_agent" {
					dedupe := "ops-stop:" + conversationInternalID
					payload, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": "device_revoked"})
					if _, enqueueErr := s.outbox.Enqueue(op, tx, outbox.NewEvent{TenantID: &a.tenantID, AggregateType: "conversation", AggregateID: conversationInternalID, EventType: "conversation.agent_stop_requested", DeduplicationKey: &dedupe, Payload: payload}); enqueueErr != nil {
						return idempotency.Result{}, enqueueErr
					}
				}
			case "STOPPING":
			default:
				return idempotency.Result{}, ErrStateConflict
			}
			if conversationStatus != fromStatus {
				if _, historyErr := tx.Exec(op, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source) VALUES($1::uuid,$2,$3,$4,'operator')`, conversationInternalID, fromStatus, conversationStatus, "device revoked: "+reason); historyErr != nil {
					return idempotency.Result{}, historyErr
				}
			}
		}

		detail := map[string]any{"from_status": oldStatus, "to_status": "revoked", "conversation_queued": queued}
		if conversationPublicID != "" {
			detail["conversation_id"] = conversationPublicID
			detail["conversation_status"] = conversationStatus
			detail["runtime_type"] = runtimeType
			detail["from_runtime_status"] = runtimeStatus
		}
		if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.device.revoke", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: detail}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		output := ActionResult{Action: "revoke_device", DeviceID: devicePublicID, ConversationID: conversationPublicID, ConversationStatus: conversationStatus, Queued: queued}
		encoded, _ := json.Marshal(output)
		return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &deviceID}, nil
	})
	if err != nil {
		return ActionResult{}, false, err
	}
	var output ActionResult
	if err = json.Unmarshal(result.Body, &output); err != nil {
		return ActionResult{}, false, err
	}
	return output, !result.Replayed, nil
}

func (s *Service) ExportBundle(ctx context.Context, principal identity.Principal, tenantPublicID, field, value, key, requestID, reason string) (Bundle, bool, error) {
	reason = strings.TrimSpace(reason)
	if reason == "" || len([]rune(reason)) > 500 {
		return Bundle{}, false, ErrInvalid
	}
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return Bundle{}, false, err
	}
	items, err := s.Search(ctx, principal, tenantPublicID, field, value)
	if err != nil {
		return Bundle{}, false, err
	}
	body, _ := json.Marshal(map[string]string{"field": field, "value": value, "reason": reason})
	result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/bundles", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(op context.Context, tx pgx.Tx) (idempotency.Result, error) {
		bundle := Bundle{ID: uuid.NewString(), GeneratedAt: time.Now().UTC(), SearchField: field, SearchValue: value, RedactionPolicy: "ops-diagnostic-minimal-v1", Items: items}
		if auditErr := s.audit.Record(op, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.bundle.export", ResourceType: "diagnostic_bundle", ResourceID: &bundle.ID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"search_field": field, "item_count": len(items), "redaction_policy": bundle.RedactionPolicy}}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		encoded, _ := json.Marshal(bundle)
		return idempotency.Result{StatusCode: 201, Body: encoded, ResourceID: &bundle.ID}, nil
	})
	if err != nil {
		return Bundle{}, false, err
	}
	var bundle Bundle
	if err = json.Unmarshal(result.Body, &bundle); err != nil {
		return Bundle{}, false, err
	}
	return bundle, !result.Replayed, nil
}

func (s *Service) ReconcileConversation(ctx context.Context, principal identity.Principal, tenantPublicID, conversationPublicID, key, requestID, reason string) (ActionResult, bool, error) {
	reason = strings.TrimSpace(reason)
	if _, err := uuid.Parse(conversationPublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
		return ActionResult{}, false, ErrInvalid
	}
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return ActionResult{}, false, err
	}
	body, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "reason": reason})
	result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/conversations/reconcile", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		var internalID, deviceID, status, taskRef, runtimeType string
		if queryErr := tx.QueryRow(operationContext, `SELECT c.id::text,d.public_id::text,c.status,COALESCE(rt.provider_task_ref,hs.provider_session_ref),c.runtime_type FROM conversation c JOIN device d ON d.id=c.device_id LEFT JOIN rtc_task rt ON rt.conversation_id=c.id LEFT JOIN hardware_agent_session hs ON hs.conversation_id=c.id WHERE c.tenant_id=$1::uuid AND c.public_id=$2::uuid FOR UPDATE OF c`, a.tenantID, conversationPublicID).Scan(&internalID, &deviceID, &status, &taskRef, &runtimeType); errors.Is(queryErr, pgx.ErrNoRows) {
			return idempotency.Result{}, ErrNotFound
		} else if queryErr != nil {
			return idempotency.Result{}, queryErr
		}
		eventType := ""
		if status == "STARTING_AGENT" {
			eventType = "conversation.agent_start_reconcile_requested"
		} else if status == "STOPPING" {
			eventType = "conversation.agent_stop_reconcile_requested"
		}
		queued := eventType != "" && runtimeType != "volc_hardware_agent"
		if queued {
			dedupe := "ops-reconcile:" + taskRef + ":" + key
			payload, _ := json.Marshal(map[string]string{"conversation_id": conversationPublicID, "provider_task_ref": taskRef, "reason": "ops_reconcile"})
			if _, enqueueErr := s.outbox.Enqueue(operationContext, tx, outbox.NewEvent{TenantID: &a.tenantID, AggregateType: "conversation", AggregateID: internalID, EventType: eventType, DeduplicationKey: &dedupe, Payload: payload}); enqueueErr != nil {
				return idempotency.Result{}, enqueueErr
			}
		}
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.conversation.reconcile", ResourceType: "conversation", ResourceID: &conversationPublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"status": status, "runtime_type": runtimeType, "queued": queued}}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		output := ActionResult{Action: "reconcile_conversation", DeviceID: deviceID, ConversationID: conversationPublicID, ConversationStatus: status, Queued: queued}
		encoded, _ := json.Marshal(output)
		return idempotency.Result{StatusCode: 200, Body: encoded, ResourceID: &internalID}, nil
	})
	if err != nil {
		return ActionResult{}, false, err
	}
	var output ActionResult
	if err = json.Unmarshal(result.Body, &output); err != nil {
		return ActionResult{}, false, err
	}
	return output, !result.Replayed, nil
}

func (s *Service) RefreshDeviceConfig(ctx context.Context, principal identity.Principal, tenantPublicID, devicePublicID, key, requestID, reason string) (ActionResult, bool, error) {
	reason = strings.TrimSpace(reason)
	if _, err := uuid.Parse(devicePublicID); err != nil || reason == "" || len([]rune(reason)) > 500 {
		return ActionResult{}, false, ErrInvalid
	}
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return ActionResult{}, false, err
	}
	body, _ := json.Marshal(map[string]string{"device_id": devicePublicID, "reason": reason})
	result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID, Scope: "POST /api/v1/ops/diagnostics/devices/config-refresh", Key: key, RequestHash: idempotency.Hash(body), ExpiresAfter: 24 * time.Hour}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		var deviceID string
		var revision int64
		if queryErr := tx.QueryRow(operationContext, `SELECT d.id::text,dt.desired_revision FROM device d JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id WHERE d.tenant_id=$1::uuid AND d.public_id=$2::uuid AND d.status='registered' FOR SHARE OF d`, a.tenantID, devicePublicID).Scan(&deviceID, &revision); errors.Is(queryErr, pgx.ErrNoRows) {
			return idempotency.Result{}, ErrNotFound
		} else if queryErr != nil {
			return idempotency.Result{}, queryErr
		}
		commandPublicID := uuid.NewString()
		commandInternalID := uuid.NewString()
		payload, _ := json.Marshal(map[string]any{"desired_revision": revision, "reason_code": "ops_config_refresh"})
		digest := idempotency.Hash(payload)
		_, insertErr := tx.Exec(operationContext, `INSERT INTO device_command(id,public_id,tenant_id,device_id,command_type,payload,status,requested_by,idempotency_key,request_digest,expires_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,'apply_config',$5::jsonb,'QUEUED',$6::uuid,$7,$8,now()+interval '1 hour')`, commandInternalID, commandPublicID, a.tenantID, deviceID, payload, a.userID, "ops-refresh:"+key, digest)
		if insertErr != nil {
			return idempotency.Result{}, insertErr
		}
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID, Action: "ops.diagnostics.device.config_refresh", ResourceType: "device", ResourceID: &devicePublicID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"desired_revision": revision, "command_id": commandPublicID}}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		output := ActionResult{Action: "refresh_config", DeviceID: devicePublicID, CommandID: commandPublicID, CommandStatus: "QUEUED", Queued: true}
		encoded, _ := json.Marshal(output)
		return idempotency.Result{StatusCode: 201, Body: encoded, ResourceID: &commandInternalID}, nil
	})
	if err != nil {
		return ActionResult{}, false, err
	}
	var output ActionResult
	if err = json.Unmarshal(result.Body, &output); err != nil {
		return ActionResult{}, false, err
	}
	return output, !result.Replayed, nil
}

const diagnosticSelect = `SELECT
	d.public_id::text,d.status,COALESCE(dp.online,false),COALESCE(dt.firmware_version,''),
	COALESCE(dt.desired_revision,1),dt.reported_revision,COALESCE(dt.last_error_code,''),dt.last_error_at,
	COALESCE(c.public_id::text,''),COALESCE(c.status,''),COALESCE(c.failure_code,''),c.allocation_expires_at,
	rt.credential_issued_at,rt.credential_expires_at,
	COALESCE(rt.room_id,''),COALESCE(rt.provider,hs.provider,''),COALESCE(rt.provider_task_ref,hs.provider_session_ref,''),COALESCE(rt.status,hs.status,''),
	COALESCE(rt.start_attempts,0),COALESCE(rt.stop_attempts,0),COALESCE(pe.event_count,0),
	COALESCE(pe.event_type,''),COALESCE(pe.event_status,''),pe.received_at,COALESCE(ae.request_id,''),
	c.created_at,ch.room_issued_at,ch.device_activated_at,ch.agent_active_at,ch.stop_requested_at,c.ended_at
FROM device d
LEFT JOIN device_twin dt ON dt.device_id=d.id AND dt.tenant_id=d.tenant_id
LEFT JOIN device_presence_state dp ON dp.device_id=d.id AND dp.tenant_id=d.tenant_id
LEFT JOIN conversation c ON c.device_id=d.id AND c.tenant_id=d.tenant_id
LEFT JOIN rtc_task rt ON rt.conversation_id=c.id
LEFT JOIN hardware_agent_session hs ON hs.conversation_id=c.id
LEFT JOIN LATERAL (
	SELECT count(*) OVER() event_count,event_type,event_status,received_at FROM (
		SELECT event_type,COALESCE(sanitized_payload->>'status','') event_status,received_at
		FROM provider_event WHERE conversation_id=c.id
		UNION ALL
		SELECT event_type,event_type,occurred_at FROM hardware_agent_event WHERE conversation_id=c.id
	) runtime_event ORDER BY received_at DESC LIMIT 1
) pe ON true
LEFT JOIN LATERAL (
	SELECT
		MIN(occurred_at) FILTER (WHERE to_status='ROOM_ISSUED') room_issued_at,
		MIN(occurred_at) FILTER (WHERE to_status='STARTING_AGENT') device_activated_at,
		MIN(occurred_at) FILTER (WHERE to_status='ACTIVE') agent_active_at,
		MIN(occurred_at) FILTER (WHERE to_status='STOPPING') stop_requested_at
	FROM conversation_state_history WHERE conversation_id=c.id
) ch ON c.id IS NOT NULL
LEFT JOIN LATERAL (
	SELECT request_id FROM audit_event
	WHERE tenant_id=d.tenant_id AND resource_id IN(d.id,d.public_id,c.id,c.public_id) AND request_id IS NOT NULL
	ORDER BY occurred_at DESC LIMIT 1
) ae ON true`

type rowScanner interface{ Scan(...any) error }

func scanRecord(row rowScanner) (Record, error) {
	var item Record
	var createdAt, roomIssuedAt, deviceActivatedAt, agentActiveAt, stopRequestedAt, endedAt *time.Time
	err := row.Scan(&item.DeviceID, &item.DeviceStatus, &item.DeviceOnline, &item.FirmwareVersion, &item.DesiredConfigRevision, &item.ReportedConfigRevision, &item.LastDeviceErrorCode, &item.LastDeviceErrorAt, &item.ConversationID, &item.ConversationStatus, &item.ConversationFailureCode, &item.AllocationExpiresAt, &item.CredentialIssuedAt, &item.CredentialExpiresAt, &item.RoomID, &item.Provider, &item.ProviderTaskRef, &item.ProviderTaskStatus, &item.ProviderStartAttempts, &item.ProviderStopAttempts, &item.ProviderNotificationCount, &item.LastProviderEvent, &item.LastProviderEventStatus, &item.LastProviderEventAt, &item.LastRequestID, &createdAt, &roomIssuedAt, &deviceActivatedAt, &agentActiveAt, &stopRequestedAt, &endedAt)
	if err == nil && createdAt != nil {
		item.Timings = &Timings{
			ConversationCreatedAt: *createdAt,
			RoomIssuedAt:          roomIssuedAt,
			DeviceActivatedAt:     deviceActivatedAt,
			AgentActiveAt:         agentActiveAt,
			StopRequestedAt:       stopRequestedAt,
			ConversationEndedAt:   endedAt,
			RoomIssueMS:           elapsedMilliseconds(createdAt, roomIssuedAt),
			DeviceJoinMS:          elapsedMilliseconds(roomIssuedAt, deviceActivatedAt),
			AgentStartMS:          elapsedMilliseconds(deviceActivatedAt, agentActiveAt),
			AgentStopMS:           elapsedMilliseconds(stopRequestedAt, endedAt),
			TotalMS:               elapsedMilliseconds(createdAt, endedAt),
		}
	}
	return item, err
}

func elapsedMilliseconds(start, end *time.Time) *int64 {
	if start == nil || end == nil || end.Before(*start) {
		return nil
	}
	value := end.Sub(*start).Milliseconds()
	return &value
}

func (r Record) String() string {
	return fmt.Sprintf("device=%s conversation=%s", r.DeviceID, r.ConversationID)
}