geo/server/internal/tenant/app/monitoring_callback_service.go

package app

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/jackc/pgx/v5/pgxpool"
	"go.uber.org/zap"
	"golang.org/x/net/publicsuffix"

	sharedauth "github.com/geo-platform/tenant-api/internal/shared/auth"
	sharedllm "github.com/geo-platform/tenant-api/internal/shared/llm"
	"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
	"github.com/geo-platform/tenant-api/internal/shared/response"
	"github.com/geo-platform/tenant-api/internal/tenant/repository"
)

const (
	defaultMonitoringLeaseLimit        = 3
	maxMonitoringLeaseLimit            = 50
	monitoringLeaseTTL                 = 15 * time.Minute
	monitoringTaskResultQueueIOTimeout = 5 * time.Second
	// The current primary renews the DB lease only after this interval. With the
	// desktop client's 25s heartbeat and a 90s lease TTL, this keeps normal
	// heartbeats read-only while still tolerating one missed heartbeat.
	monitoringHeartbeatPrimaryRenewAfter = desktopClientPresenceTTL / 3
	// Access snapshots are semantic state, not a liveness clock. Stable heartbeat
	// reports skip writes and only refresh the row at this interval so dashboards
	// do not depend on per-heartbeat database churn.
	monitoringHeartbeatAccessSnapshotRefreshAfter = 10 * time.Minute
)

var monitoringCitationMarkerPattern = regexp.MustCompile(`(?i)\[\s*citation\s*:\s*\d+\s*\]`)

type MonitoringCallbackService struct {
	businessPool   *pgxpool.Pool
	monitoringPool *pgxpool.Pool
	rabbitMQ       *rabbitmq.Client
	llm            sharedllm.Client
	logger         *zap.Logger
}

func NewMonitoringCallbackService(
	businessPool, monitoringPool *pgxpool.Pool,
	rabbitMQClient *rabbitmq.Client,
	llmClient sharedllm.Client,
	logger *zap.Logger,
) *MonitoringCallbackService {
	return &MonitoringCallbackService{
		businessPool:   businessPool,
		monitoringPool: monitoringPool,
		rabbitMQ:       rabbitMQClient,
		llm:            llmClient,
		logger:         logger,
	}
}

type MonitoringTaskResultRequest struct {
	LeaseToken           string                 `json:"lease_token" binding:"required"`
	Status               string                 `json:"status"`
	ProviderModel        *string                `json:"provider_model"`
	ProviderRequestID    *string                `json:"provider_request_id"`
	RequestID            *string                `json:"request_id"`
	Answer               *string                `json:"answer"`
	RawResponseJSON      map[string]interface{} `json:"raw_response_json"`
	Citations            []MonitoringSourceItem `json:"citations"`
	SearchResults        []MonitoringSourceItem `json:"search_results"`
	BrandMentioned       *bool                  `json:"brand_mentioned"`
	BrandMentionPosition *string                `json:"brand_mention_position"`
	FirstRecommended     *bool                  `json:"first_recommended"`
	SentimentLabel       *string                `json:"sentiment_label"`
	MatchedBrandTerms    []string               `json:"matched_brand_terms"`
	ErrorMessage         *string                `json:"error_message"`
	CompletedAt          *time.Time             `json:"completed_at"`
}

type MonitoringSourceItem struct {
	URL                  string  `json:"url"`
	Title                *string `json:"title"`
	SiteName             *string `json:"site_name"`
	SiteKey              *string `json:"site_key"`
	NormalizedURL        *string `json:"normalized_url"`
	Host                 *string `json:"host"`
	RegistrableDomain    *string `json:"registrable_domain"`
	Subdomain            *string `json:"subdomain"`
	Suffix               *string `json:"suffix"`
	ArticleID            *int64  `json:"article_id"`
	PublishRecordID      *int64  `json:"publish_record_id"`
	ResolutionStatus     *string `json:"resolution_status"`
	ResolutionConfidence *string `json:"resolution_confidence"`
}

type MonitoringHeartbeatRequest struct {
	Platforms []MonitoringHeartbeatPlatform `json:"platforms"`
}

type MonitoringHeartbeatPlatform struct {
	AIPlatformID  string     `json:"ai_platform_id"`
	AccessStatus  *string    `json:"access_status"`
	Connected     *bool      `json:"connected"`
	Accessible    *bool      `json:"accessible"`
	DetectedAt    *time.Time `json:"detected_at"`
	ReasonCode    *string    `json:"reason_code"`
	ReasonMessage *string    `json:"reason_message"`
}

type MonitoringHeartbeatResponse struct {
	InstallationID             string `json:"installation_id"`
	BusinessDate               string `json:"business_date"`
	UpdatedPlatformCount       int    `json:"updated_platform_count"`
	ReconciledSkippedTaskCount int64  `json:"reconciled_skipped_task_count"`
}

type MonitoringLeaseTasksRequest struct {
	Limit         int      `json:"limit"`
	AIPlatformIDs []string `json:"ai_platform_ids"`
}

type MonitoringResumeTasksRequest struct {
	AIPlatformIDs []string `json:"ai_platform_ids"`
}

type MonitoringLeaseTasksResponse struct {
	BusinessDate string                `json:"business_date"`
	LeaseCount   int                   `json:"lease_count"`
	Tasks        []MonitoringLeaseTask `json:"tasks"`
}

type MonitoringResumeTasksResponse struct {
	BusinessDate string                `json:"business_date"`
	ResumeCount  int                   `json:"resume_count"`
	Tasks        []MonitoringLeaseTask `json:"tasks"`
}

type MonitoringLeaseTask struct {
	TaskID         int64  `json:"task_id"`
	BrandID        int64  `json:"brand_id"`
	QuestionID     int64  `json:"question_id"`
	QuestionHash   string `json:"question_hash"`
	QuestionText   string `json:"question_text"`
	AIPlatformID   string `json:"ai_platform_id"`
	PlatformName   string `json:"platform_name"`
	CollectorType  string `json:"collector_type"`
	RunMode        string `json:"run_mode"`
	TriggerSource  string `json:"trigger_source"`
	BusinessDate   string `json:"business_date"`
	Priority       int    `json:"priority,omitempty"`
	Lane           string `json:"lane,omitempty"`
	LeaseToken     string `json:"lease_token"`
	LeaseExpiresAt string `json:"lease_expires_at"`
}

type MonitoringTaskResultResponse struct {
	TaskID               int64  `json:"task_id"`
	RunID                *int64 `json:"run_id"`
	TaskStatus           string `json:"task_status"`
	CitationSourceCount  int    `json:"citation_source_count"`
	ContentCitationCount int    `json:"content_citation_count"`
}

type MonitoringSkipTaskRequest struct {
	LeaseToken   string     `json:"lease_token" binding:"required"`
	SkipReason   *string    `json:"skip_reason"`
	ErrorMessage *string    `json:"error_message"`
	CompletedAt  *time.Time `json:"completed_at"`
}

type MonitoringSkipTaskResponse struct {
	TaskID      int64   `json:"task_id"`
	TaskStatus  string  `json:"task_status"`
	SkipReason  string  `json:"skip_reason"`
	CompletedAt *string `json:"completed_at"`
}

type MonitoringCancelTaskRequest struct {
	LeaseToken string  `json:"lease_token" binding:"required"`
	Reason     *string `json:"reason"`
}

type MonitoringCancelTaskResponse struct {
	TaskID     int64  `json:"task_id"`
	TaskStatus string `json:"task_status"`
}

type monitoringInstallationIdentity struct {
	ID          string
	TenantID    int64
	WorkspaceID int64
}

func monitoringInstallationFromDesktopClient(client *repository.DesktopClient) *monitoringInstallationIdentity {
	if client == nil {
		return nil
	}
	return &monitoringInstallationIdentity{
		ID:          client.ID.String(),
		TenantID:    client.TenantID,
		WorkspaceID: client.WorkspaceID,
	}
}

type monitoringPlatformAccessSnapshotState struct {
	AccessStatus  string
	Connected     bool
	Accessible    bool
	ReasonCode    *string
	ReasonMessage *string
	UpdatedAt     time.Time
}

type monitoringHeartbeatSnapshotDecision struct {
	WriteSnapshot             bool
	StateChanged              bool
	RefreshDue                bool
	PendingReconcileCandidate bool
	ReconcilePending          bool
	SnapshotOperation         string
}

type monitoringCollectTask struct {
	ID                 int64
	TenantID           int64
	BrandID            int64
	QuestionID         int64
	QuestionHash       []byte
	AIPlatformID       string
	CollectorType      string
	RunMode            string
	BusinessDate       time.Time
	Status             string
	ExecutionOwner     string
	LeaseTokenHash     sql.NullString
	LeasedToExecutor   sql.NullString
	LeaseExpiresAt     sql.NullTime
	CallbackReceivedAt sql.NullTime
	CompletedAt        sql.NullTime
	SkipReason         sql.NullString
	TriggerSource      string
	QuestionText       string
}

type monitoringDesktopExecutionLease struct {
	DesktopTaskID       uuid.UUID
	Status              string
	TargetClientID      uuid.UUID
	LeaseTokenHash      []byte
	InterruptGeneration int
}

type monitoringLeasedTaskRow struct {
	ID            int64
	BrandID       int64
	QuestionID    int64
	QuestionHash  []byte
	AIPlatformID  string
	CollectorType string
	RunMode       string
	TriggerSource string
	BusinessDate  time.Time
	Priority      int
	Lane          string
	QuestionText  string
}

type monitoringAliasResolution struct {
	ArticleID        *int64
	PublishRecordID  *int64
	NormalizedURLKey string
}

type monitoringCitationFact struct {
	CitedURL             string
	CitedTitle           *string
	NormalizedURL        string
	Host                 string
	RegistrableDomain    string
	Subdomain            *string
	Suffix               *string
	SiteKey              string
	ArticleID            *int64
	PublishRecordID      *int64
	ResolutionStatus     string
	ResolutionConfidence string
}

type monitoringTaskResultEnvelope struct {
	Version               string                      `json:"version"`
	TaskID                int64                       `json:"task_id"`
	TenantID              int64                       `json:"tenant_id"`
	InstallationID        string                      `json:"installation_id"`
	ReceivedAt            time.Time                   `json:"received_at"`
	QueueStatus           string                      `json:"queue_status"`
	QueueClaimedAt        *time.Time                  `json:"queue_claimed_at"`
	QueueEnqueuedAt       *time.Time                  `json:"queue_enqueued_at"`
	LastQueuePublishError *string                     `json:"last_queue_publish_error"`
	ReceivedAlertedAt     *time.Time                  `json:"received_alerted_at"`
	ReceivedAlertCount    int                         `json:"received_alert_count"`
	CallbackResult        MonitoringTaskResultRequest `json:"callback_result"`
}

type monitoringTaskResultQueueMetaPatch struct {
	QueueStatus           string     `json:"queue_status"`
	QueueClaimedAt        *time.Time `json:"queue_claimed_at"`
	QueueEnqueuedAt       *time.Time `json:"queue_enqueued_at"`
	LastQueuePublishError *string    `json:"last_queue_publish_error"`
}

type MonitoringTaskResultQueueMetaPatch = monitoringTaskResultQueueMetaPatch

func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	return s.handleTaskResultForInstallation(ctx, installation, taskID, req)
}

func (s *MonitoringCallbackService) HandleTaskResultForDesktopClient(
	ctx context.Context,
	client *repository.DesktopClient,
	taskID int64,
	req MonitoringTaskResultRequest,
) (*MonitoringTaskResultResponse, error) {
	if client == nil {
		return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
	}
	return s.handleTaskResultForInstallation(ctx, monitoringInstallationFromDesktopClient(client), taskID, req)
}

func (s *MonitoringCallbackService) handleTaskResultForInstallation(
	ctx context.Context,
	installation *monitoringInstallationIdentity,
	taskID int64,
	req MonitoringTaskResultRequest,
) (*MonitoringTaskResultResponse, error) {
	task, err := s.loadCollectTask(ctx, installation.TenantID, taskID)
	if err != nil {
		return nil, err
	}

	if task.CollectorType != monitoringCollectorType {
		return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not desktop")
	}

	if task.Status == "completed" {
		runID, citationSourceCount, contentCitationCount, loadErr := s.loadExistingRunSummary(ctx, task)
		if loadErr != nil {
			return nil, loadErr
		}
		_ = s.touchInstallation(ctx, installation.ID)
		return &MonitoringTaskResultResponse{
			TaskID:               task.ID,
			RunID:                runID,
			TaskStatus:           task.Status,
			CitationSourceCount:  citationSourceCount,
			ContentCitationCount: contentCitationCount,
		}, nil
	}

	if task.ExecutionOwner == "desktop_tasks" {
		if err := s.validateDesktopMonitorExecutionLease(ctx, task, installation.ID, req.LeaseToken); err != nil {
			return nil, err
		}
	} else {
		if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installation.ID {
			return nil, response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
		}

		if strings.TrimSpace(task.LeaseTokenHash.String) == "" {
			return nil, response.ErrConflict(40941, "lease_token_missing", "task does not have an active lease token")
		}

		if sharedauth.HashToken(strings.TrimSpace(req.LeaseToken)) != task.LeaseTokenHash.String {
			return nil, response.ErrUnauthorized(40141, "invalid_lease_token", "lease token is invalid")
		}
	}

	switch task.Status {
	case "leased", "received", "failed", "expired", "pending":
	default:
		return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow result submission")
	}

	runStatus := normalizeRunStatus(req.Status)
	if runStatus == "" {
		return nil, response.ErrBadRequest(40041, "invalid_status", "status must be succeeded or failed")
	}
	if err := validateMonitoringTaskResultSubmission(task.AIPlatformID, runStatus, req); err != nil {
		return nil, err
	}

	receivedAt := time.Now().UTC().Round(0)
	envelope := newMonitoringTaskResultEnvelope(task, installation.ID, receivedAt, req)
	requestPayloadJSON, err := json.Marshal(envelope)
	if err != nil {
		return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode monitoring callback payload")
	}

	tx, err := s.monitoringPool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring callback receive")
	}
	defer tx.Rollback(ctx)

	if err := s.markTaskResultReceived(ctx, tx, task, receivedAt, requestPayloadJSON); err != nil {
		return nil, err
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring callback receive")
	}

	_ = s.touchInstallation(ctx, installation.ID)

	publishCtx, cancel := context.WithTimeout(context.Background(), monitoringTaskResultQueueIOTimeout)
	defer cancel()

	if err := s.publishTaskResultEnvelope(publishCtx, requestPayloadJSON); err != nil {
		s.tryPatchTaskResultQueueMeta(task.ID, receivedAt, monitoringTaskResultQueueMetaPatch{
			QueueStatus:           "publish_failed",
			LastQueuePublishError: optionalString(err.Error()),
		})
		return nil, response.ErrServiceUnavailable(50341, "monitoring_result_queue_unavailable", "failed to queue monitoring result")
	}

	enqueuedAt := time.Now().UTC().Round(0)
	s.tryPatchTaskResultQueueMeta(task.ID, receivedAt, monitoringTaskResultQueueMetaPatch{
		QueueStatus:     "queued",
		QueueEnqueuedAt: &enqueuedAt,
	})

	if task.ExecutionOwner == "desktop_tasks" {
		if finalizeErr := s.completeLinkedDesktopMonitorTask(ctx, task, req); finalizeErr != nil && s.logger != nil {
			s.logger.Warn("phase2 monitor desktop task finalize failed",
				zap.Int64("monitor_task_id", task.ID),
				zap.String("ai_platform_id", task.AIPlatformID),
				zap.Error(finalizeErr),
			)
		}
	}

	return &MonitoringTaskResultResponse{
		TaskID:               task.ID,
		TaskStatus:           "received",
		CitationSourceCount:  0,
		ContentCitationCount: 0,
	}, nil
}

func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID string, installationToken string, req MonitoringHeartbeatRequest) (_ *MonitoringHeartbeatResponse, err error) {
	startedAt := time.Now()
	snapshotMetrics := monitoringHeartbeatSnapshotMetricsEvent{}
	defer func() {
		if !snapshotMetrics.Active {
			return
		}
		snapshotMetrics.Duration = time.Since(startedAt)
		snapshotMetrics.Err = err
		recordMonitoringHeartbeatSnapshotMetrics(snapshotMetrics)
	}()

	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	isPrimaryInstallation, err := s.resolveHeartbeatPrimaryInstallation(ctx, installation.TenantID, installation)
	if err != nil {
		return nil, err
	}

	now := time.Now().UTC()
	businessDay, businessDate := monitoringBusinessDayAndDateAt(now)
	platforms, err := normalizeHeartbeatPlatforms(req.Platforms, now)
	if err != nil {
		return nil, err
	}
	snapshotMetrics.Active = true
	snapshotMetrics.PlatformCount = len(platforms)

	if len(platforms) == 0 {
		_ = s.touchInstallation(ctx, installation.ID)
		return &MonitoringHeartbeatResponse{
			InstallationID:             installation.ID,
			BusinessDate:               businessDate,
			UpdatedPlatformCount:       0,
			ReconciledSkippedTaskCount: 0,
		}, nil
	}

	platformIDs := monitoringHeartbeatPlatformIDs(platforms)
	previousStates, err := s.loadPlatformAccessSnapshotStates(ctx, installation.TenantID, installation.ID, businessDate, platformIDs)
	if err != nil {
		return nil, err
	}

	type heartbeatPlatformAction struct {
		platform MonitoringHeartbeatPlatform
		decision monitoringHeartbeatSnapshotDecision
	}

	actions := make([]heartbeatPlatformAction, 0, len(platforms))
	inaccessiblePlatformIDs := make([]string, 0, len(platforms))
	needsTransaction := false
	for _, platform := range platforms {
		nextState := monitoringPlatformAccessSnapshotStateFromHeartbeat(platform)
		decision := decideMonitoringHeartbeatSnapshot(previousStates[platform.AIPlatformID], nextState, now, isPrimaryInstallation)
		if decision.WriteSnapshot {
			needsTransaction = true
		} else {
			snapshotMetrics.SnapshotSkippedCount++
		}
		if decision.PendingReconcileCandidate {
			inaccessiblePlatformIDs = append(inaccessiblePlatformIDs, platform.AIPlatformID)
		}
		actions = append(actions, heartbeatPlatformAction{
			platform: platform,
			decision: decision,
		})
	}

	if isPrimaryInstallation && len(inaccessiblePlatformIDs) > 0 {
		pendingPlatforms, pendingErr := s.loadPendingMonitoringTaskPlatformIDs(ctx, installation.TenantID, businessDate, inaccessiblePlatformIDs)
		if pendingErr != nil {
			return nil, pendingErr
		}
		for idx := range actions {
			if !actions[idx].decision.PendingReconcileCandidate {
				continue
			}
			if _, ok := pendingPlatforms[actions[idx].platform.AIPlatformID]; ok {
				actions[idx].decision.ReconcilePending = true
				needsTransaction = true
			}
		}
	}

	if !needsTransaction {
		_ = s.touchInstallation(ctx, installation.ID)
		return &MonitoringHeartbeatResponse{
			InstallationID:             installation.ID,
			BusinessDate:               businessDate,
			UpdatedPlatformCount:       len(platforms),
			ReconciledSkippedTaskCount: 0,
		}, nil
	}

	tx, err := s.monitoringPool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring heartbeat")
	}
	snapshotMetrics.TransactionOpened = true
	defer tx.Rollback(ctx)

	var reconciledSkippedTaskCount int64
	var snapshotInsertedCount int64
	var snapshotUpdatedCount int64
	var snapshotRefreshedCount int64
	var projectionRebuildCount int64
	affectedBrandIDs := make(map[int64]struct{})
	for _, action := range actions {
		if action.decision.WriteSnapshot {
			if err := s.upsertPlatformAccessSnapshot(ctx, tx, installation, businessDate, action.platform); err != nil {
				return nil, err
			}
			switch action.decision.SnapshotOperation {
			case "insert":
				snapshotInsertedCount++
			case "update":
				snapshotUpdatedCount++
			case "refresh":
				snapshotRefreshedCount++
			}
		}

		if !isPrimaryInstallation {
			continue
		}

		needsProjectionRefresh := action.decision.StateChanged
		if action.decision.ReconcilePending {
			count, reconcileErr := s.reconcilePendingTasksForPlatform(ctx, tx, installation.TenantID, businessDate, action.platform.AIPlatformID)
			if reconcileErr != nil {
				return nil, reconcileErr
			}
			reconciledSkippedTaskCount += count
			if count > 0 {
				needsProjectionRefresh = true
			}
		}

		if !needsProjectionRefresh {
			continue
		}

		brandIDs, loadErr := s.loadMonitoringAffectedBrandIDs(ctx, tx, installation.TenantID, businessDay, action.platform.AIPlatformID)
		if loadErr != nil {
			return nil, loadErr
		}
		for _, brandID := range brandIDs {
			affectedBrandIDs[brandID] = struct{}{}
		}
	}

	if isPrimaryInstallation {
		for brandID := range affectedBrandIDs {
			if err := s.rebuildBrandDailyProjection(ctx, tx, installation.TenantID, brandID, businessDay); err != nil {
				return nil, err
			}
			projectionRebuildCount++
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring heartbeat")
	}

	snapshotMetrics.SnapshotInsertedCount = snapshotInsertedCount
	snapshotMetrics.SnapshotUpdatedCount = snapshotUpdatedCount
	snapshotMetrics.SnapshotRefreshedCount = snapshotRefreshedCount
	snapshotMetrics.ReconciledSkippedTaskCount = reconciledSkippedTaskCount
	snapshotMetrics.ProjectionRebuildCount = projectionRebuildCount

	_ = s.touchInstallation(ctx, installation.ID)
	if isPrimaryInstallation && len(affectedBrandIDs) > 0 {
		s.dispatchMonitoringProjectionRebuilds(installation.TenantID, businessDay, affectedBrandIDs, "heartbeat")
	}

	return &MonitoringHeartbeatResponse{
		InstallationID:             installation.ID,
		BusinessDate:               businessDate,
		UpdatedPlatformCount:       len(platforms),
		ReconciledSkippedTaskCount: reconciledSkippedTaskCount,
	}, nil
}

func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installationID string, installationToken string, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	return s.leaseTasksForInstallation(ctx, installation, req)
}

func (s *MonitoringCallbackService) LeaseTasksForDesktopClient(ctx context.Context, client *repository.DesktopClient, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) {
	if client == nil {
		return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
	}
	return s.leaseTasksForInstallation(ctx, monitoringInstallationFromDesktopClient(client), req)
}

func (s *MonitoringCallbackService) leaseTasksForInstallation(
	ctx context.Context,
	installation *monitoringInstallationIdentity,
	req MonitoringLeaseTasksRequest,
) (*MonitoringLeaseTasksResponse, error) {
	limit := normalizeMonitoringLeaseLimit(req.Limit)
	eligiblePlatformIDs := normalizeMonitoringLeasePlatformIDs(req.AIPlatformIDs)
	now := time.Now().UTC()
	businessDay, businessDate := monitoringBusinessDayAndDateAt(now)
	leaseExpiresAt := now.Add(monitoringLeaseTTL)

	tx, err := s.monitoringPool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring task lease")
	}
	defer tx.Rollback(ctx)

	if _, err := expireMonitoringLeasedTasks(ctx, tx, &installation.TenantID, now); err != nil {
		return nil, err
	}
	if _, err := requeueMonitoringExpiredTasks(ctx, tx, installation.TenantID, businessDay); err != nil {
		return nil, err
	}
	if _, err := skipMonitoringTasksBeforeBusinessDate(ctx, tx, installation.TenantID, businessDay, "legacy"); err != nil {
		return nil, err
	}

	rows, err := s.selectPendingLeaseTasks(ctx, tx, installation, businessDate, limit, eligiblePlatformIDs)
	if err != nil {
		return nil, err
	}
	if len(rows) == 0 {
		if err := tx.Commit(ctx); err != nil {
			return nil, response.ErrInternal(50041, "commit_failed", "failed to finish empty monitoring lease")
		}
		_ = s.touchInstallation(ctx, installation.ID)
		return &MonitoringLeaseTasksResponse{
			BusinessDate: businessDate,
			LeaseCount:   0,
			Tasks:        []MonitoringLeaseTask{},
		}, nil
	}

	tasks := make([]MonitoringLeaseTask, 0, len(rows))
	for _, row := range rows {
		leaseToken, tokenErr := newSessionToken()
		if tokenErr != nil {
			return nil, response.ErrInternal(50041, "lease_token_failed", "failed to generate monitoring lease token")
		}
		if err := s.markTaskLeased(ctx, tx, installation.ID, row.ID, leaseToken, now, leaseExpiresAt); err != nil {
			return nil, err
		}
		tasks = append(tasks, MonitoringLeaseTask{
			TaskID:         row.ID,
			BrandID:        row.BrandID,
			QuestionID:     row.QuestionID,
			QuestionHash:   encodeQuestionHash(row.QuestionHash),
			QuestionText:   row.QuestionText,
			AIPlatformID:   row.AIPlatformID,
			PlatformName:   platformDisplayName(row.AIPlatformID),
			CollectorType:  row.CollectorType,
			RunMode:        row.RunMode,
			TriggerSource:  row.TriggerSource,
			BusinessDate:   row.BusinessDate.Format("2006-01-02"),
			Priority:       row.Priority,
			Lane:           row.Lane,
			LeaseToken:     leaseToken,
			LeaseExpiresAt: leaseExpiresAt.Format(time.RFC3339),
		})
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring task lease")
	}

	_ = s.touchInstallation(ctx, installation.ID)

	return &MonitoringLeaseTasksResponse{
		BusinessDate: businessDate,
		LeaseCount:   len(tasks),
		Tasks:        tasks,
	}, nil
}

func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installationID string, installationToken string, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	return s.resumeTasksForInstallation(ctx, installation, req)
}

func (s *MonitoringCallbackService) ResumeTasksForDesktopClient(ctx context.Context, client *repository.DesktopClient, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) {
	if client == nil {
		return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
	}
	return s.resumeTasksForInstallation(ctx, monitoringInstallationFromDesktopClient(client), req)
}

func (s *MonitoringCallbackService) resumeTasksForInstallation(
	ctx context.Context,
	installation *monitoringInstallationIdentity,
	req MonitoringResumeTasksRequest,
) (*MonitoringResumeTasksResponse, error) {
	eligiblePlatformIDs := normalizeMonitoringLeasePlatformIDs(req.AIPlatformIDs)
	now := time.Now().UTC()
	businessDay, businessDate := monitoringBusinessDayAndDateAt(now)
	leaseExpiresAt := now.Add(monitoringLeaseTTL)

	tx, err := s.monitoringPool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring task resume")
	}
	defer tx.Rollback(ctx)

	if _, err := expireMonitoringLeasedTasks(ctx, tx, &installation.TenantID, now); err != nil {
		return nil, err
	}
	if _, err := requeueMonitoringExpiredTasks(ctx, tx, installation.TenantID, businessDay); err != nil {
		return nil, err
	}
	if _, err := skipMonitoringTasksBeforeBusinessDate(ctx, tx, installation.TenantID, businessDay, "legacy"); err != nil {
		return nil, err
	}

	rows, err := s.selectLeasedTasksForInstallation(ctx, tx, installation.TenantID, installation.ID, businessDate, eligiblePlatformIDs)
	if err != nil {
		return nil, err
	}
	if len(rows) == 0 {
		if err := tx.Commit(ctx); err != nil {
			return nil, response.ErrInternal(50041, "commit_failed", "failed to finish empty monitoring resume")
		}
		_ = s.touchInstallation(ctx, installation.ID)
		return &MonitoringResumeTasksResponse{
			BusinessDate: businessDate,
			ResumeCount:  0,
			Tasks:        []MonitoringLeaseTask{},
		}, nil
	}

	tasks := make([]MonitoringLeaseTask, 0, len(rows))
	for _, row := range rows {
		leaseToken, tokenErr := newSessionToken()
		if tokenErr != nil {
			return nil, response.ErrInternal(50041, "lease_token_failed", "failed to generate monitoring resume token")
		}
		if err := s.markTaskLeased(ctx, tx, installation.ID, row.ID, leaseToken, now, leaseExpiresAt); err != nil {
			return nil, err
		}
		tasks = append(tasks, MonitoringLeaseTask{
			TaskID:         row.ID,
			BrandID:        row.BrandID,
			QuestionID:     row.QuestionID,
			QuestionHash:   encodeQuestionHash(row.QuestionHash),
			QuestionText:   row.QuestionText,
			AIPlatformID:   row.AIPlatformID,
			PlatformName:   platformDisplayName(row.AIPlatformID),
			CollectorType:  row.CollectorType,
			RunMode:        row.RunMode,
			TriggerSource:  row.TriggerSource,
			BusinessDate:   row.BusinessDate.Format("2006-01-02"),
			Priority:       row.Priority,
			Lane:           row.Lane,
			LeaseToken:     leaseToken,
			LeaseExpiresAt: leaseExpiresAt.Format(time.RFC3339),
		})
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring task resume")
	}

	_ = s.touchInstallation(ctx, installation.ID)

	return &MonitoringResumeTasksResponse{
		BusinessDate: businessDate,
		ResumeCount:  len(tasks),
		Tasks:        tasks,
	}, nil
}

func (s *MonitoringCallbackService) SkipTask(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringSkipTaskRequest) (*MonitoringSkipTaskResponse, error) {
	installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
	if err != nil {
		return nil, err
	}

	return s.skipTaskForInstallation(ctx, installation, taskID, req)
}

func (s *MonitoringCallbackService) SkipTaskForDesktopClient(
	ctx context.Context,
	client *repository.DesktopClient,
	taskID int64,
	req MonitoringSkipTaskRequest,
) (*MonitoringSkipTaskResponse, error) {
	if client == nil {
		return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
	}
	return s.skipTaskForInstallation(ctx, monitoringInstallationFromDesktopClient(client), taskID, req)
}

func (s *MonitoringCallbackService) skipTaskForInstallation(
	ctx context.Context,
	installation *monitoringInstallationIdentity,
	taskID int64,
	req MonitoringSkipTaskRequest,
) (*MonitoringSkipTaskResponse, error) {
	task, err := s.loadCollectTask(ctx, installation.TenantID, taskID)
	if err != nil {
		return nil, err
	}

	if task.CollectorType != monitoringCollectorType {
		return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not plugin")
	}
	if err := s.validateCollectTaskExecution(ctx, task, installation.ID, req.LeaseToken); err != nil {
		return nil, err
	}

	if task.Status == "skipped" {
		_ = s.touchInstallation(ctx, installation.ID)
		return &MonitoringSkipTaskResponse{
			TaskID:      task.ID,
			TaskStatus:  task.Status,
			SkipReason:  strings.TrimSpace(task.SkipReason.String),
			CompletedAt: formatNullTime(task.CompletedAt),
		}, nil
	}

	switch task.Status {
	case "leased", "received", "failed", "expired", "pending":
	default:
		return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow skip submission")
	}

	completedAt := time.Now().UTC()
	if req.CompletedAt != nil && !req.CompletedAt.IsZero() {
		completedAt = req.CompletedAt.UTC()
	}

	skipReason := normalizeSkipReason(req.SkipReason)

	tx, err := s.monitoringPool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring task skip")
	}
	defer tx.Rollback(ctx)

	if _, err := tx.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = 'skipped',
		    callback_received_at = NOW(),
		    completed_at = $2,
		    skip_reason = $3,
		    error_message = $4,
		    updated_at = NOW()
		WHERE id = $1
	`, task.ID, completedAt, skipReason, nullableString(req.ErrorMessage)); err != nil {
		return nil, response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task skip status")
	}

	if err := s.rebuildBrandDailyProjection(ctx, tx, task.TenantID, task.BrandID, task.BusinessDate); err != nil {
		return nil, err
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring task skip")
	}

	_ = s.touchInstallation(ctx, installation.ID)

	completedAtText := completedAt.Format(time.RFC3339)
	return &MonitoringSkipTaskResponse{
		TaskID:      task.ID,
		TaskStatus:  "skipped",
		SkipReason:  skipReason,
		CompletedAt: &completedAtText,
	}, nil
}

func (s *MonitoringCallbackService) CancelTaskForDesktopClient(
	ctx context.Context,
	client *repository.DesktopClient,
	taskID int64,
	req MonitoringCancelTaskRequest,
) (*MonitoringCancelTaskResponse, error) {
	if client == nil {
		return nil, response.ErrUnauthorized(40141, "desktop_client_missing", "desktop client context is required")
	}

	installation := monitoringInstallationFromDesktopClient(client)
	task, err := s.loadCollectTask(ctx, installation.TenantID, taskID)
	if err != nil {
		return nil, err
	}

	if task.CollectorType != monitoringCollectorType {
		return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not desktop")
	}
	if err := s.validateCollectTaskExecution(ctx, task, installation.ID, req.LeaseToken); err != nil {
		return nil, err
	}
	var desktopTaskCanceled *repository.DesktopTask
	if task.ExecutionOwner == "desktop_tasks" {
		desktopExecution, leaseErr := s.loadActiveDesktopMonitorExecutionLease(ctx, task.ID)
		if leaseErr != nil {
			return nil, leaseErr
		}
		if desktopExecution == nil || desktopExecution.Status != "in_progress" {
			return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow lease cancel")
		}
		errorJSON, marshalErr := json.Marshal(map[string]any{
			"reason": optionalStringValue(req.Reason),
			"source": "legacy_monitor_cancel",
		})
		if marshalErr != nil {
			return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode phase2 desktop cancel payload")
		}
		taskRepo := repository.NewDesktopTaskRepository(s.businessPool)
		canceledTask, cancelErr := taskRepo.CancelByLease(ctx, desktopExecution.DesktopTaskID, desktopExecution.LeaseTokenHash, errorJSON)
		if cancelErr != nil {
			if errors.Is(cancelErr, pgx.ErrNoRows) {
				return nil, response.ErrConflict(40941, "task_status_invalid", "phase2 monitor desktop task is no longer active")
			}
			return nil, response.ErrInternal(50041, "desktop_task_cancel_failed", "failed to cancel phase2 monitor desktop task lease")
		}
		desktopTaskCanceled = canceledTask
	} else if task.Status != "leased" {
		return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow lease cancel")
	}

	if _, err := s.monitoringPool.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = 'pending',
		    lease_token_hash = NULL,
		    leased_to_executor = NULL,
		    leased_at = NULL,
		    lease_expires_at = NULL,
		    callback_received_at = NULL,
		    dispatch_after = NOW(),
		    error_message = $2,
		    updated_at = NOW()
		WHERE id = $1
	`, task.ID, nullableString(req.Reason)); err != nil {
		return nil, response.ErrInternal(50041, "task_update_failed", "failed to cancel monitoring task lease")
	}

	if desktopTaskCanceled != nil {
		title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(
			desktopTaskCanceled.Kind,
			desktopTaskCanceled.Payload,
		)
		publishDesktopTaskEvent(ctx, s.rabbitMQ, DesktopTaskEvent{
			Type:                "task_canceled",
			TaskID:              desktopTaskCanceled.DesktopID.String(),
			JobID:               desktopTaskCanceled.JobID.String(),
			WorkspaceID:         desktopTaskCanceled.WorkspaceID,
			TargetAccountID:     desktopTaskCanceled.TargetAccountID.String(),
			TargetClientID:      desktopTaskCanceled.TargetClientID.String(),
			Platform:            desktopTaskCanceled.Platform,
			Title:               title,
			BusinessDate:        businessDate,
			SchedulerGroupKey:   schedulerGroupKey,
			QuestionText:        questionText,
			Status:              desktopTaskCanceled.Status,
			Kind:                desktopTaskCanceled.Kind,
			Priority:            desktopTaskCanceled.Priority,
			Lane:                desktopTaskCanceled.Lane,
			InterruptGeneration: desktopTaskCanceled.InterruptGeneration,
			UpdatedAt:           desktopTaskCanceled.UpdatedAt,
		})
	}

	return &MonitoringCancelTaskResponse{
		TaskID:     task.ID,
		TaskStatus: "pending",
	}, nil
}

func (s *MonitoringCallbackService) ProcessQueuedTaskResult(ctx context.Context, envelope monitoringTaskResultEnvelope) (*MonitoringTaskResultResponse, error) {
	if envelope.TaskID <= 0 {
		return nil, response.ErrBadRequest(40041, "invalid_task_id", "task id is required")
	}

	task, err := s.loadCollectTaskByID(ctx, envelope.TaskID)
	if err != nil {
		if appErr, ok := err.(*response.AppError); ok && appErr.HTTPStatus == 404 {
			return nil, nil
		}
		return nil, err
	}

	if envelope.TenantID > 0 && envelope.TenantID != task.TenantID {
		return nil, response.ErrConflict(40941, "task_tenant_mismatch", "task tenant does not match queued monitoring result")
	}

	if !monitoringTaskReadyForQueuedResult(task) {
		return nil, nil
	}

	return s.processTaskResult(ctx, task, envelope.CallbackResult)
}

func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID string, installationToken string) (*monitoringInstallationIdentity, error) {
	tokenHash := HashDesktopClientToken(strings.TrimSpace(installationToken))

	row := &monitoringInstallationIdentity{}
	if err := s.businessPool.QueryRow(ctx, `
		SELECT id::text, tenant_id, workspace_id
		FROM desktop_clients
		WHERE id = $1::uuid
		  AND token_hash = $2
		  AND revoked_at IS NULL
	`, strings.TrimSpace(installationID), tokenHash).Scan(&row.ID, &row.TenantID, &row.WorkspaceID); err != nil {
		if err == pgx.ErrNoRows {
			return nil, response.ErrUnauthorized(40141, "desktop_client_unauthorized", "desktop client is invalid or expired")
		}
		return nil, response.ErrInternal(50041, "desktop_client_lookup_failed", "failed to validate desktop client")
	}

	return row, nil
}

func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) (bool, error) {
	if installation == nil || strings.TrimSpace(installation.ID) == "" {
		return false, nil
	}
	isPrimary, err := s.claimHeartbeatPrimaryLease(ctx, tenantID, installation.WorkspaceID, installation.ID)
	if err != nil {
		return false, err
	}
	return isPrimary, nil
}

func (s *MonitoringCallbackService) claimHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
	startedAt := time.Now()
	installationID = strings.TrimSpace(installationID)
	leaseTTL := formatPgInterval(desktopClientPresenceTTL)

	state, err := s.loadHeartbeatPrimaryLeaseState(ctx, tenantID, workspaceID, installationID)
	if err != nil {
		recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
		return false, err
	}
	if !state.Found {
		inserted, insertErr := s.tryInsertHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, leaseTTL)
		if insertErr != nil {
			recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
			return false, insertErr
		}
		if inserted {
			recordHeartbeatPrimaryLeaseMetric("inserted", time.Since(startedAt))
			recordHeartbeatPrimaryLeaseWriteMetric("insert")
			return true, nil
		}
		recordHeartbeatPrimaryLeaseMetric("contention", time.Since(startedAt))
		return s.resolveHeartbeatPrimaryAfterContention(ctx, tenantID, workspaceID, installationID)
	}

	decision := decideHeartbeatPrimaryLease(state)
	switch decision.Action {
	case heartbeatPrimaryLeaseActionPrimaryHit:
		recordHeartbeatPrimaryLeaseMetric("primary_hit", time.Since(startedAt))
		return true, nil
	case heartbeatPrimaryLeaseActionNonPrimaryLive:
		recordHeartbeatPrimaryLeaseMetric("non_primary_live", time.Since(startedAt))
		return false, nil
	case heartbeatPrimaryLeaseActionRenew:
		updated, updateErr := s.renewHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, leaseTTL)
		if updateErr != nil {
			recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
			return false, updateErr
		}
		if updated {
			recordHeartbeatPrimaryLeaseMetric("renewed", time.Since(startedAt))
			recordHeartbeatPrimaryLeaseWriteMetric("renew")
			return true, nil
		}
	case heartbeatPrimaryLeaseActionClaimExpired, heartbeatPrimaryLeaseActionClaimUnavailable:
		updated, updateErr := s.claimExpiredHeartbeatPrimaryLease(ctx, tenantID, workspaceID, installationID, state.PrimaryInstallationID, leaseTTL, decision.Action)
		if updateErr != nil {
			recordHeartbeatPrimaryLeaseMetric("error", time.Since(startedAt))
			return false, updateErr
		}
		if updated {
			if decision.Action == heartbeatPrimaryLeaseActionClaimExpired {
				recordHeartbeatPrimaryLeaseMetric("claimed_expired", time.Since(startedAt))
			} else {
				recordHeartbeatPrimaryLeaseMetric("claimed_unavailable", time.Since(startedAt))
			}
			recordHeartbeatPrimaryLeaseWriteMetric("claim")
			return true, nil
		}
	}

	recordHeartbeatPrimaryLeaseMetric("contention", time.Since(startedAt))
	return s.resolveHeartbeatPrimaryAfterContention(ctx, tenantID, workspaceID, installationID)
}

func (s *MonitoringCallbackService) tryInsertHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID, leaseTTL string) (bool, error) {
	tag, err := s.businessPool.Exec(ctx, `
		INSERT INTO desktop_client_primary_leases (
			tenant_id,
			workspace_id,
			primary_client_id,
			lease_expires_at
		)
		VALUES ($1, $2, $3::uuid, NOW() + $4::interval)
		ON CONFLICT (tenant_id, workspace_id) DO NOTHING
	`, tenantID, workspaceID, installationID, leaseTTL)
	if err != nil {
		return false, response.ErrInternal(50041, "primary_lease_insert_failed", "failed to initialize monitoring desktop primary lease")
	}
	return tag.RowsAffected() > 0, nil
}

type heartbeatPrimaryLeaseState struct {
	Found                 bool
	PrimaryInstallationID string
	CurrentIsPrimary      bool
	LeaseLive             bool
	RenewDue              bool
	PrimaryAvailable      bool
}

func (s *MonitoringCallbackService) loadHeartbeatPrimaryLeaseState(ctx context.Context, tenantID, workspaceID int64, installationID string) (heartbeatPrimaryLeaseState, error) {
	var state heartbeatPrimaryLeaseState
	renewAfter := formatPgInterval(monitoringHeartbeatPrimaryRenewAfter)
	if err := s.businessPool.QueryRow(ctx, `
		SELECT
			l.primary_client_id::text,
			l.primary_client_id = $3::uuid AS current_is_primary,
			l.lease_expires_at > NOW() AS lease_live,
			l.updated_at <= NOW() - $4::interval AS renew_due,
			dc.id IS NOT NULL AND dc.revoked_at IS NULL AS primary_available
		FROM desktop_client_primary_leases l
		LEFT JOIN desktop_clients dc ON dc.id = l.primary_client_id
		WHERE l.tenant_id = $1
		  AND l.workspace_id = $2
	`, tenantID, workspaceID, installationID, renewAfter).Scan(
		&state.PrimaryInstallationID,
		&state.CurrentIsPrimary,
		&state.LeaseLive,
		&state.RenewDue,
		&state.PrimaryAvailable,
	); err != nil {
		if err == pgx.ErrNoRows {
			return state, nil
		}
		return state, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop primary lease")
	}
	state.Found = true
	return state, nil
}

type heartbeatPrimaryLeaseAction string

const (
	heartbeatPrimaryLeaseActionPrimaryHit       heartbeatPrimaryLeaseAction = "primary_hit"
	heartbeatPrimaryLeaseActionNonPrimaryLive   heartbeatPrimaryLeaseAction = "non_primary_live"
	heartbeatPrimaryLeaseActionRenew            heartbeatPrimaryLeaseAction = "renew"
	heartbeatPrimaryLeaseActionClaimExpired     heartbeatPrimaryLeaseAction = "claim_expired"
	heartbeatPrimaryLeaseActionClaimUnavailable heartbeatPrimaryLeaseAction = "claim_unavailable"
)

type heartbeatPrimaryLeaseDecision struct {
	Action heartbeatPrimaryLeaseAction
}

func decideHeartbeatPrimaryLease(state heartbeatPrimaryLeaseState) heartbeatPrimaryLeaseDecision {
	if !state.Found {
		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionNonPrimaryLive}
	}
	if state.CurrentIsPrimary {
		if !state.LeaseLive || !state.PrimaryAvailable || state.RenewDue {
			return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionRenew}
		}
		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionPrimaryHit}
	}
	if !state.LeaseLive {
		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionClaimExpired}
	}
	if !state.PrimaryAvailable {
		return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionClaimUnavailable}
	}
	return heartbeatPrimaryLeaseDecision{Action: heartbeatPrimaryLeaseActionNonPrimaryLive}
}

func (s *MonitoringCallbackService) renewHeartbeatPrimaryLease(ctx context.Context, tenantID, workspaceID int64, installationID, leaseTTL string) (bool, error) {
	tag, err := s.businessPool.Exec(ctx, `
		UPDATE desktop_client_primary_leases
		SET lease_expires_at = NOW() + $4::interval,
		    updated_at = NOW()
		WHERE tenant_id = $1
		  AND workspace_id = $2
		  AND primary_client_id = $3::uuid
	`, tenantID, workspaceID, installationID, leaseTTL)
	if err != nil {
		return false, response.ErrInternal(50041, "primary_lease_update_failed", "failed to renew monitoring desktop primary lease")
	}
	return tag.RowsAffected() > 0, nil
}

func (s *MonitoringCallbackService) claimExpiredHeartbeatPrimaryLease(
	ctx context.Context,
	tenantID, workspaceID int64,
	installationID string,
	previousPrimaryInstallationID string,
	leaseTTL string,
	action heartbeatPrimaryLeaseAction,
) (bool, error) {
	claimPredicate := "lease_expires_at <= NOW()"
	if action == heartbeatPrimaryLeaseActionClaimUnavailable {
		claimPredicate = `NOT EXISTS (
			SELECT 1
			FROM desktop_clients dc
			WHERE dc.id = desktop_client_primary_leases.primary_client_id
			  AND dc.revoked_at IS NULL
		)`
	}
	query := `
		UPDATE desktop_client_primary_leases
		SET primary_client_id = $3::uuid,
		    lease_expires_at = NOW() + $5::interval,
		    updated_at = NOW()
		WHERE tenant_id = $1
		  AND workspace_id = $2
		  AND primary_client_id = $4::uuid
		  AND ` + claimPredicate
	tag, err := s.businessPool.Exec(ctx, query, tenantID, workspaceID, installationID, previousPrimaryInstallationID, leaseTTL)
	if err != nil {
		return false, response.ErrInternal(50041, "primary_lease_update_failed", "failed to claim monitoring desktop primary lease")
	}
	return tag.RowsAffected() > 0, nil
}

func (s *MonitoringCallbackService) resolveHeartbeatPrimaryAfterContention(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
	state, err := s.loadHeartbeatPrimaryLeaseState(ctx, tenantID, workspaceID, installationID)
	if err != nil {
		return false, err
	}
	return state.CurrentIsPrimary && state.LeaseLive && state.PrimaryAvailable, nil
}

func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantID, taskID int64) (*monitoringCollectTask, error) {
	item := &monitoringCollectTask{}
	if err := s.monitoringPool.QueryRow(ctx, `
		SELECT
			id,
			tenant_id,
			brand_id,
			question_id,
			question_hash,
			ai_platform_id,
			collector_type,
			run_mode,
			business_date,
			status,
			COALESCE(execution_owner, 'legacy') AS execution_owner,
			lease_token_hash,
			leased_to_executor,
			lease_expires_at,
			callback_received_at,
			completed_at,
			skip_reason,
			trigger_source,
			COALESCE((
				SELECT question_text_snapshot
				FROM monitoring_question_config_snapshots
				WHERE tenant_id = t.tenant_id
				  AND brand_id = t.brand_id
				  AND question_id = t.question_id
				  AND question_hash = t.question_hash
				  AND deleted_at IS NULL
				ORDER BY projected_at DESC, id DESC
				LIMIT 1
			), '') AS question_text_snapshot
		FROM monitoring_collect_tasks t
		WHERE id = $1
		  AND tenant_id = $2
	`, taskID, tenantID).Scan(
		&item.ID,
		&item.TenantID,
		&item.BrandID,
		&item.QuestionID,
		&item.QuestionHash,
		&item.AIPlatformID,
		&item.CollectorType,
		&item.RunMode,
		&item.BusinessDate,
		&item.Status,
		&item.ExecutionOwner,
		&item.LeaseTokenHash,
		&item.LeasedToExecutor,
		&item.LeaseExpiresAt,
		&item.CallbackReceivedAt,
		&item.CompletedAt,
		&item.SkipReason,
		&item.TriggerSource,
		&item.QuestionText,
	); err != nil {
		if err == pgx.ErrNoRows {
			return nil, response.ErrNotFound(40441, "monitoring_task_not_found", "monitoring task not found")
		}
		return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring task")
	}

	return item, nil
}

func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, taskID int64) (*monitoringCollectTask, error) {
	item := &monitoringCollectTask{}
	if err := s.monitoringPool.QueryRow(ctx, `
		SELECT
			id,
			tenant_id,
			brand_id,
			question_id,
			question_hash,
			ai_platform_id,
			collector_type,
			run_mode,
			business_date,
			status,
			COALESCE(execution_owner, 'legacy') AS execution_owner,
			lease_token_hash,
			leased_to_executor,
			lease_expires_at,
			callback_received_at,
			completed_at,
			skip_reason,
			trigger_source,
			COALESCE((
				SELECT question_text_snapshot
				FROM monitoring_question_config_snapshots
				WHERE tenant_id = t.tenant_id
				  AND brand_id = t.brand_id
				  AND question_id = t.question_id
				  AND question_hash = t.question_hash
				  AND deleted_at IS NULL
				ORDER BY projected_at DESC, id DESC
				LIMIT 1
			), '') AS question_text_snapshot
		FROM monitoring_collect_tasks t
		WHERE id = $1
	`, taskID).Scan(
		&item.ID,
		&item.TenantID,
		&item.BrandID,
		&item.QuestionID,
		&item.QuestionHash,
		&item.AIPlatformID,
		&item.CollectorType,
		&item.RunMode,
		&item.BusinessDate,
		&item.Status,
		&item.ExecutionOwner,
		&item.LeaseTokenHash,
		&item.LeasedToExecutor,
		&item.LeaseExpiresAt,
		&item.CallbackReceivedAt,
		&item.CompletedAt,
		&item.SkipReason,
		&item.TriggerSource,
		&item.QuestionText,
	); err != nil {
		if err == pgx.ErrNoRows {
			return nil, response.ErrNotFound(40441, "monitoring_task_not_found", "monitoring task not found")
		}
		return nil, response.ErrInternal(50041, "query_failed", "failed to load monitoring task")
	}

	return item, nil
}

func (s *MonitoringCallbackService) loadActiveDesktopMonitorExecutionLease(
	ctx context.Context,
	monitorTaskID int64,
) (*monitoringDesktopExecutionLease, error) {
	item := &monitoringDesktopExecutionLease{}
	if err := s.businessPool.QueryRow(ctx, `
		SELECT
			desktop_id,
			status,
			target_client_id,
			lease_token_hash,
			interrupt_generation
		FROM desktop_tasks
		WHERE kind = 'monitor'
		  AND monitor_task_id = $1
		  AND status IN ('queued', 'in_progress', 'unknown')
		ORDER BY created_at DESC, id DESC
		LIMIT 1
	`, monitorTaskID).Scan(
		&item.DesktopTaskID,
		&item.Status,
		&item.TargetClientID,
		&item.LeaseTokenHash,
		&item.InterruptGeneration,
	); err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, response.ErrInternal(50041, "query_failed", "failed to inspect phase2 monitor desktop execution lease")
	}
	return item, nil
}

func (s *MonitoringCallbackService) selectPendingLeaseTasks(
	ctx context.Context,
	tx pgx.Tx,
	installation *monitoringInstallationIdentity,
	businessDate string,
	limit int,
	eligiblePlatformIDs []string,
) ([]monitoringLeasedTaskRow, error) {
	queryArgs := []interface{}{installation.TenantID, monitoringCollectorType, businessDate, limit, installation.ID}
	query := `
			SELECT
				t.id,
			t.brand_id,
			t.question_id,
			t.question_hash,
			t.ai_platform_id,
			t.collector_type,
			t.run_mode,
			t.trigger_source,
			t.business_date,
			COALESCE(t.dispatch_priority, 100) AS dispatch_priority,
			COALESCE(t.dispatch_lane, 'normal') AS dispatch_lane,
			COALESCE((
				SELECT question_text_snapshot
				FROM monitoring_question_config_snapshots
				WHERE tenant_id = t.tenant_id
				  AND brand_id = t.brand_id
				  AND question_id = t.question_id
				  AND question_hash = t.question_hash
				  AND deleted_at IS NULL
				ORDER BY projected_at DESC, id DESC
				LIMIT 1
			), '') AS question_text_snapshot
			FROM monitoring_collect_tasks t
			WHERE t.tenant_id = $1
			  AND t.collector_type = $2
			  AND t.status = 'pending'
			  AND t.business_date = $3::date
			  AND COALESCE(t.execution_owner, 'legacy') = 'legacy'
			  AND (t.dispatch_after IS NULL OR t.dispatch_after <= NOW())
			  AND (
				t.target_client_id IS NULL
				OR t.target_client_id = $5::uuid
			  )
			  AND EXISTS (
				SELECT 1
				FROM monitoring_question_config_snapshots s
				WHERE s.tenant_id = t.tenant_id
				  AND s.brand_id = t.brand_id
				  AND s.question_id = t.question_id
				  AND s.question_hash = t.question_hash
				  AND s.monitor_enabled = TRUE
				  AND s.deleted_at IS NULL
				  AND s.superseded_at IS NULL
			  )
		`
	if len(eligiblePlatformIDs) > 0 {
		query += `
		  AND t.ai_platform_id = ANY($6::text[])
		`
		queryArgs = append(queryArgs, eligiblePlatformIDs)
	}
	query += `
		ORDER BY CASE t.dispatch_lane
			WHEN 'high' THEN 70
			WHEN 'normal_boosted' THEN 50
			WHEN 'normal' THEN 40
			WHEN 'retry' THEN 20
			ELSE 0
		END DESC,
		t.dispatch_priority DESC,
		t.dispatch_after ASC NULLS FIRST,
		t.planned_at ASC,
		t.id ASC
		FOR UPDATE SKIP LOCKED
		LIMIT $4
	`

	rows, err := tx.Query(ctx, query, queryArgs...)
	if err != nil {
		return nil, response.ErrInternal(50041, "lease_query_failed", "failed to select monitoring lease tasks")
	}
	defer rows.Close()

	result := make([]monitoringLeasedTaskRow, 0, limit)
	for rows.Next() {
		var item monitoringLeasedTaskRow
		if scanErr := rows.Scan(
			&item.ID,
			&item.BrandID,
			&item.QuestionID,
			&item.QuestionHash,
			&item.AIPlatformID,
			&item.CollectorType,
			&item.RunMode,
			&item.TriggerSource,
			&item.BusinessDate,
			&item.Priority,
			&item.Lane,
			&item.QuestionText,
		); scanErr != nil {
			return nil, response.ErrInternal(50041, "lease_scan_failed", "failed to parse monitoring lease task")
		}
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50041, "lease_scan_failed", "failed to iterate monitoring lease tasks")
	}

	return result, nil
}

func (s *MonitoringCallbackService) selectLeasedTasksForInstallation(
	ctx context.Context,
	tx pgx.Tx,
	tenantID int64,
	installationID string,
	businessDate string,
	eligiblePlatformIDs []string,
) ([]monitoringLeasedTaskRow, error) {
	queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, installationID}
	query := `
			SELECT
				t.id,
			t.brand_id,
			t.question_id,
			t.question_hash,
			t.ai_platform_id,
			t.collector_type,
			t.run_mode,
			t.trigger_source,
			t.business_date,
			COALESCE(t.dispatch_priority, 100) AS dispatch_priority,
			COALESCE(t.dispatch_lane, 'normal') AS dispatch_lane,
			COALESCE((
				SELECT question_text_snapshot
				FROM monitoring_question_config_snapshots
				WHERE tenant_id = t.tenant_id
				  AND brand_id = t.brand_id
				  AND question_id = t.question_id
				  AND question_hash = t.question_hash
				  AND deleted_at IS NULL
				ORDER BY projected_at DESC, id DESC
				LIMIT 1
			), '') AS question_text_snapshot
			FROM monitoring_collect_tasks t
			WHERE t.tenant_id = $1
			  AND t.collector_type = $2
			  AND t.status = 'leased'
			  AND t.business_date = $3::date
			  AND COALESCE(t.execution_owner, 'legacy') = 'legacy'
			  AND t.leased_to_executor = $4
			  AND EXISTS (
				SELECT 1
				FROM monitoring_question_config_snapshots s
				WHERE s.tenant_id = t.tenant_id
				  AND s.brand_id = t.brand_id
				  AND s.question_id = t.question_id
				  AND s.question_hash = t.question_hash
				  AND s.monitor_enabled = TRUE
				  AND s.deleted_at IS NULL
				  AND s.superseded_at IS NULL
			  )
		`
	if len(eligiblePlatformIDs) > 0 {
		query += `
		  AND t.ai_platform_id = ANY($5::text[])
		`
		queryArgs = append(queryArgs, eligiblePlatformIDs)
	}
	query += `
		ORDER BY t.leased_at ASC NULLS LAST, t.planned_at ASC, t.id ASC
		FOR UPDATE SKIP LOCKED
	`

	rows, err := tx.Query(ctx, query, queryArgs...)
	if err != nil {
		return nil, response.ErrInternal(50041, "resume_query_failed", "failed to select monitoring resumed tasks")
	}
	defer rows.Close()

	result := make([]monitoringLeasedTaskRow, 0)
	for rows.Next() {
		var item monitoringLeasedTaskRow
		if scanErr := rows.Scan(
			&item.ID,
			&item.BrandID,
			&item.QuestionID,
			&item.QuestionHash,
			&item.AIPlatformID,
			&item.CollectorType,
			&item.RunMode,
			&item.TriggerSource,
			&item.BusinessDate,
			&item.Priority,
			&item.Lane,
			&item.QuestionText,
		); scanErr != nil {
			return nil, response.ErrInternal(50041, "resume_scan_failed", "failed to parse monitoring resumed task")
		}
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50041, "resume_scan_failed", "failed to iterate monitoring resumed tasks")
	}

	return result, nil
}

func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.Tx, installationID string, taskID int64, leaseToken string, leasedAt, leaseExpiresAt time.Time) error {
	if _, err := tx.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = 'leased',
		    lease_token_hash = $2,
		    leased_to_executor = $3,
		    leased_at = $4,
		    lease_expires_at = $5,
		    last_dispatched_at = NOW(),
		    dispatch_attempts = COALESCE(dispatch_attempts, 0) + 1,
		    skip_reason = NULL,
		    error_message = NULL,
		    updated_at = NOW()
		WHERE id = $1
	`, taskID, sharedauth.HashToken(strings.TrimSpace(leaseToken)), installationID, leasedAt, leaseExpiresAt); err != nil {
		return response.ErrInternal(50041, "lease_update_failed", "failed to update monitoring lease task")
	}
	return nil
}

func (s *MonitoringCallbackService) markTaskResultReceived(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, receivedAt time.Time, requestPayloadJSON []byte) error {
	if task == nil {
		return response.ErrBadRequest(40041, "invalid_task_id", "monitoring task is required")
	}

	nextStatus := "received"
	if task.ExecutionOwner == "desktop_tasks" {
		nextStatus = "pending"
	}

	if _, err := tx.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = $2,
		    callback_received_at = $3,
		    completed_at = NULL,
		    skip_reason = NULL,
		    error_message = NULL,
		    request_payload_json = (
				CASE
					WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
						THEN '{}'::jsonb
					ELSE request_payload_json
				END
			) || $4::jsonb,
		    updated_at = NOW()
		WHERE id = $1
	`, task.ID, nextStatus, receivedAt, requestPayloadJSON); err != nil {
		return response.ErrInternal(50041, "task_receive_failed", "failed to persist monitoring callback payload")
	}
	return nil
}

func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Context, tx pgx.Tx, installation *monitoringInstallationIdentity, businessDate string, platform MonitoringHeartbeatPlatform) error {
	if _, err := tx.Exec(ctx, `
		INSERT INTO monitoring_platform_access_snapshots (
			tenant_id, client_id, ai_platform_id, business_date, access_status,
			connected, accessible, detected_at, reason_code, reason_message
		)
		VALUES ($1, $2::uuid, $3, $4::date, $5, $6, $7, $8, $9, $10)
		ON CONFLICT (tenant_id, client_id, ai_platform_id, business_date)
		DO UPDATE SET
			access_status = EXCLUDED.access_status,
			connected = EXCLUDED.connected,
			accessible = EXCLUDED.accessible,
			detected_at = EXCLUDED.detected_at,
			reason_code = EXCLUDED.reason_code,
			reason_message = EXCLUDED.reason_message,
			updated_at = NOW()
	`,
		installation.TenantID,
		installation.ID,
		platform.AIPlatformID,
		businessDate,
		monitoringStringValue(platform.AccessStatus),
		derefBool(platform.Connected),
		derefBool(platform.Accessible),
		resolveHeartbeatDetectedAt(platform.DetectedAt),
		nullableString(platform.ReasonCode),
		nullableString(platform.ReasonMessage),
	); err != nil {
		return response.ErrInternal(50041, "access_snapshot_upsert_failed", "failed to persist monitoring platform access snapshot")
	}
	return nil
}

func (s *MonitoringCallbackService) loadPlatformAccessSnapshotStates(ctx context.Context, tenantID int64, installationID string, businessDate string, aiPlatformIDs []string) (map[string]*monitoringPlatformAccessSnapshotState, error) {
	states := make(map[string]*monitoringPlatformAccessSnapshotState, len(aiPlatformIDs))
	if len(aiPlatformIDs) == 0 {
		return states, nil
	}

	rows, err := s.monitoringPool.Query(ctx, `
		SELECT ai_platform_id, access_status, connected, accessible, reason_code, reason_message, updated_at
		FROM monitoring_platform_access_snapshots
		WHERE tenant_id = $1
		  AND client_id = $2::uuid
		  AND business_date = $3::date
		  AND ai_platform_id = ANY($4::text[])
	`, tenantID, installationID, businessDate, aiPlatformIDs)
	if err != nil {
		return nil, response.ErrInternal(50041, "access_snapshot_lookup_failed", "failed to inspect monitoring platform access snapshots")
	}
	defer rows.Close()

	for rows.Next() {
		item := &monitoringPlatformAccessSnapshotState{}
		var platformID string
		var reasonCode sql.NullString
		var reasonMessage sql.NullString
		if scanErr := rows.Scan(
			&platformID,
			&item.AccessStatus,
			&item.Connected,
			&item.Accessible,
			&reasonCode,
			&reasonMessage,
			&item.UpdatedAt,
		); scanErr != nil {
			return nil, response.ErrInternal(50041, "access_snapshot_scan_failed", "failed to parse monitoring platform access snapshots")
		}
		if reasonCode.Valid {
			item.ReasonCode = optionalString(reasonCode.String)
		}
		if reasonMessage.Valid {
			item.ReasonMessage = optionalString(reasonMessage.String)
		}
		states[platformID] = item
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50041, "access_snapshot_scan_failed", "failed to iterate monitoring platform access snapshots")
	}

	return states, nil
}

func monitoringPlatformAccessSnapshotStateFromHeartbeat(platform MonitoringHeartbeatPlatform) monitoringPlatformAccessSnapshotState {
	return monitoringPlatformAccessSnapshotState{
		AccessStatus:  monitoringStringValue(platform.AccessStatus),
		Connected:     derefBool(platform.Connected),
		Accessible:    derefBool(platform.Accessible),
		ReasonCode:    optionalString(monitoringStringValue(platform.ReasonCode)),
		ReasonMessage: optionalString(monitoringStringValue(platform.ReasonMessage)),
	}
}

func decideMonitoringHeartbeatSnapshot(previous *monitoringPlatformAccessSnapshotState, next monitoringPlatformAccessSnapshotState, now time.Time, isPrimary bool) monitoringHeartbeatSnapshotDecision {
	stateChanged := hasPlatformAccessSnapshotChanged(previous, next)
	refreshDue := false
	if previous != nil && !stateChanged {
		refreshDue = previous.UpdatedAt.IsZero() || !previous.UpdatedAt.After(now.Add(-monitoringHeartbeatAccessSnapshotRefreshAfter))
	}

	operation := "skip"
	if previous == nil {
		operation = "insert"
	} else if stateChanged {
		operation = "update"
	} else if refreshDue {
		operation = "refresh"
	}

	return monitoringHeartbeatSnapshotDecision{
		WriteSnapshot:             stateChanged || refreshDue,
		StateChanged:              stateChanged,
		RefreshDue:                refreshDue,
		PendingReconcileCandidate: isPrimary && next.AccessStatus != "accessible",
		SnapshotOperation:         operation,
	}
}

func hasPlatformAccessSnapshotChanged(previous *monitoringPlatformAccessSnapshotState, next monitoringPlatformAccessSnapshotState) bool {
	if previous == nil {
		return true
	}

	if previous.AccessStatus != next.AccessStatus ||
		previous.Connected != next.Connected ||
		previous.Accessible != next.Accessible {
		return true
	}
	if monitoringStringValue(previous.ReasonCode) != monitoringStringValue(next.ReasonCode) {
		return true
	}
	if monitoringStringValue(previous.ReasonMessage) != monitoringStringValue(next.ReasonMessage) {
		return true
	}

	return false
}

func (s *MonitoringCallbackService) loadPendingMonitoringTaskPlatformIDs(ctx context.Context, tenantID int64, businessDate string, aiPlatformIDs []string) (map[string]struct{}, error) {
	result := make(map[string]struct{}, len(aiPlatformIDs))
	if len(aiPlatformIDs) == 0 {
		return result, nil
	}

	rows, err := s.monitoringPool.Query(ctx, `
		SELECT ai_platform_id
		FROM monitoring_collect_tasks
		WHERE tenant_id = $1
		  AND ai_platform_id = ANY($2::text[])
		  AND collector_type = $3
		  AND business_date = $4::date
		  AND status = 'pending'
		GROUP BY ai_platform_id
	`, tenantID, aiPlatformIDs, monitoringCollectorType, businessDate)
	if err != nil {
		return nil, response.ErrInternal(50041, "pending_task_platform_lookup_failed", "failed to inspect pending monitoring task platforms")
	}
	defer rows.Close()

	for rows.Next() {
		var platformID string
		if scanErr := rows.Scan(&platformID); scanErr != nil {
			return nil, response.ErrInternal(50041, "pending_task_platform_scan_failed", "failed to parse pending monitoring task platforms")
		}
		result[platformID] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return nil, response.ErrInternal(50041, "pending_task_platform_scan_failed", "failed to iterate pending monitoring task platforms")
	}

	return result, nil
}

func (s *MonitoringCallbackService) reconcilePendingTasksForPlatform(ctx context.Context, tx pgx.Tx, tenantID int64, businessDate, aiPlatformID string) (int64, error) {
	tag, err := tx.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = 'skipped',
		    skip_reason = 'platform_not_accessible',
		    error_message = 'platform heartbeat reported not accessible',
		    completed_at = NOW(),
		    updated_at = NOW()
		WHERE tenant_id = $1
		  AND ai_platform_id = $2
		  AND collector_type = $3
		  AND business_date = $4::date
		  AND status = 'pending'
	`, tenantID, aiPlatformID, monitoringCollectorType, businessDate)
	if err != nil {
		return 0, response.ErrInternal(50041, "task_reconcile_failed", "failed to reconcile pending monitoring tasks")
	}
	return tag.RowsAffected(), nil
}

func (s *MonitoringCallbackService) upsertRun(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, req MonitoringTaskResultRequest, runStatus string, completedAt time.Time, rawPayload []byte) (int64, error) {
	var runID int64
	if err := tx.QueryRow(ctx, `
		INSERT INTO question_monitor_runs (
			tenant_id, brand_id, question_id, question_text_snapshot, question_hash,
			ai_platform_id, collector_type, trigger_source, run_mode, business_date,
			provider_model, provider_request_id, request_id, raw_answer_text, raw_response_json,
			status, completed_at
		)
		VALUES (
			$1, $2, $3, $4, $5,
			$6, $7, $8, $9, $10::date,
			$11, $12, $13, $14, $15,
			$16, $17
		)
		ON CONFLICT (
			tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date
		)
		DO UPDATE SET
			question_text_snapshot = EXCLUDED.question_text_snapshot,
			question_hash = EXCLUDED.question_hash,
			provider_model = EXCLUDED.provider_model,
			provider_request_id = EXCLUDED.provider_request_id,
			request_id = EXCLUDED.request_id,
			raw_answer_text = EXCLUDED.raw_answer_text,
			raw_response_json = EXCLUDED.raw_response_json,
			status = EXCLUDED.status,
			completed_at = EXCLUDED.completed_at,
			updated_at = NOW()
		RETURNING id
	`,
		task.TenantID,
		task.BrandID,
		task.QuestionID,
		task.QuestionText,
		task.QuestionHash,
		task.AIPlatformID,
		task.CollectorType,
		task.TriggerSource,
		task.RunMode,
		task.BusinessDate.Format("2006-01-02"),
		nullableString(req.ProviderModel),
		nullableString(req.ProviderRequestID),
		nullableString(req.RequestID),
		nullableString(req.Answer),
		rawPayload,
		runStatus,
		completedAt,
	).Scan(&runID); err != nil {
		return 0, response.ErrInternal(50041, "run_upsert_failed", "failed to persist monitoring run")
	}
	return runID, nil
}

func (s *MonitoringCallbackService) upsertParseResult(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, runID int64, req MonitoringTaskResultRequest, parseVersion string) error {
	matchedTermsJSON, err := json.Marshal(req.MatchedBrandTerms)
	if err != nil {
		return response.ErrInternal(50041, "parse_payload_failed", "failed to encode matched brand terms")
	}
	if strings.TrimSpace(parseVersion) == "" {
		parseVersion = "server_heuristic_v1"
	}

	if _, err := tx.Exec(ctx, `
		INSERT INTO question_monitor_parse_results (
			tenant_id, run_id, brand_id, question_id, question_hash, ai_platform_id, business_date,
			brand_mentioned, brand_mention_position, first_recommended, sentiment_label, matched_brand_terms,
			parse_version, parse_status
		)
		VALUES (
			$1, $2, $3, $4, $5, $6, $7::date,
			$8, $9, $10, $11, $12,
			$13, 'succeeded'
		)
		ON CONFLICT (tenant_id, run_id)
		DO UPDATE SET
			brand_mentioned = EXCLUDED.brand_mentioned,
			brand_mention_position = EXCLUDED.brand_mention_position,
			first_recommended = EXCLUDED.first_recommended,
			sentiment_label = EXCLUDED.sentiment_label,
			matched_brand_terms = EXCLUDED.matched_brand_terms,
			parse_version = EXCLUDED.parse_version,
			parse_status = EXCLUDED.parse_status,
			updated_at = NOW()
	`,
		task.TenantID,
		runID,
		task.BrandID,
		task.QuestionID,
		task.QuestionHash,
		task.AIPlatformID,
		task.BusinessDate.Format("2006-01-02"),
		nullableBool(req.BrandMentioned),
		nullableString(req.BrandMentionPosition),
		nullableBool(req.FirstRecommended),
		nullableString(req.SentimentLabel),
		matchedTermsJSON,
		parseVersion,
	); err != nil {
		return response.ErrInternal(50041, "parse_result_upsert_failed", "failed to persist parse result")
	}
	return nil
}

func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task *monitoringCollectTask, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
	req = normalizeMonitoringTaskResultRequestValue(req)

	runStatus := normalizeRunStatus(req.Status)
	if runStatus == "" {
		return nil, response.ErrBadRequest(40041, "invalid_status", "status must be succeeded or failed")
	}

	parseReq := req
	parseVersion := ""
	if runStatus == "succeeded" {
		var enrichErr error
		parseReq, parseVersion, enrichErr = s.enrichMonitoringParsePayload(ctx, task, req)
		if enrichErr != nil {
			return nil, enrichErr
		}
	}

	completedAt := time.Now().UTC().Round(0)
	if req.CompletedAt != nil && !req.CompletedAt.IsZero() {
		completedAt = req.CompletedAt.UTC().Round(0)
	}

	rawPayload, sourceInputs := buildMonitoringRawPayload(task.AIPlatformID, req)

	tx, err := s.monitoringPool.Begin(ctx)
	if err != nil {
		return nil, response.ErrInternal(50041, "begin_failed", "failed to start monitoring result ingest")
	}
	defer tx.Rollback(ctx)

	runID, err := s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload)
	if err != nil {
		return nil, err
	}

	if runStatus == "succeeded" {
		if err := s.upsertParseResult(ctx, tx, task, runID, parseReq, parseVersion); err != nil {
			return nil, err
		}
	}

	citationSourceCount := 0
	contentCitationCount := 0
	if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil {
		return nil, err
	}
	if runStatus == "succeeded" {
		facts, buildErr := s.buildCitationFacts(ctx, tx, task.TenantID, task.AIPlatformID, sourceInputs)
		if buildErr != nil {
			return nil, buildErr
		}
		for _, fact := range facts {
			if err := s.insertCitationFact(ctx, tx, task, runID, fact); err != nil {
				return nil, err
			}
			citationSourceCount++
			if fact.ArticleID != nil {
				contentCitationCount++
			}
		}
	}

	taskStatus := "completed"
	if runStatus == "failed" {
		taskStatus = "failed"
	}
	if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
		return nil, err
	}

	if err := s.rebuildBrandDailyProjection(ctx, tx, task.TenantID, task.BrandID, task.BusinessDate); err != nil {
		return nil, err
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring result ingest")
	}

	return &MonitoringTaskResultResponse{
		TaskID:               task.ID,
		RunID:                &runID,
		TaskStatus:           taskStatus,
		CitationSourceCount:  citationSourceCount,
		ContentCitationCount: contentCitationCount,
	}, nil
}

func buildDesktopMonitorTaskResultValue(task *monitoringCollectTask, req MonitoringTaskResultRequest, runStatus string) map[string]any {
	result := map[string]any{
		"monitoring_task_id": task.ID,
		"status":             runStatus,
	}

	if task != nil {
		if strings.TrimSpace(task.AIPlatformID) != "" {
			result["platform"] = strings.TrimSpace(task.AIPlatformID)
		}
		if strings.TrimSpace(task.QuestionText) != "" {
			result["question_text"] = strings.TrimSpace(task.QuestionText)
		}
	}
	if value := strings.TrimSpace(optionalStringValue(req.ProviderModel)); value != "" {
		result["provider_model"] = value
	}
	if value := strings.TrimSpace(optionalStringValue(req.ProviderRequestID)); value != "" {
		result["provider_request_id"] = value
	}
	if value := strings.TrimSpace(optionalStringValue(req.RequestID)); value != "" {
		result["request_id"] = value
	}
	if value := strings.TrimSpace(optionalStringValue(req.Answer)); value != "" {
		result["answer"] = value
	}
	if len(req.Citations) > 0 {
		result["citations"] = req.Citations
	}
	if len(req.SearchResults) > 0 {
		result["search_results"] = req.SearchResults
	}
	if len(req.RawResponseJSON) > 0 {
		result["raw_response_json"] = req.RawResponseJSON
	}
	return result
}

func buildDesktopMonitorTaskErrorValue(task *monitoringCollectTask, req MonitoringTaskResultRequest) map[string]any {
	message := strings.TrimSpace(optionalStringValue(req.ErrorMessage))
	if message == "" {
		message = "monitoring task failed"
	}

	result := map[string]any{
		"code":               "monitoring_task_failed",
		"message":            message,
		"monitoring_task_id": task.ID,
	}
	if task != nil && strings.TrimSpace(task.AIPlatformID) != "" {
		result["platform"] = strings.TrimSpace(task.AIPlatformID)
	}
	if len(req.RawResponseJSON) > 0 {
		result["raw_response_json"] = req.RawResponseJSON
	}
	return result
}

func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask(
	ctx context.Context,
	task *monitoringCollectTask,
	req MonitoringTaskResultRequest,
) error {
	if s == nil || s.businessPool == nil || task == nil || task.ExecutionOwner != "desktop_tasks" {
		return nil
	}

	runStatus := normalizeRunStatus(req.Status)
	if runStatus == "" {
		return nil
	}

	tx, err := s.businessPool.Begin(ctx)
	if err != nil {
		return response.ErrInternal(50041, "desktop_task_complete_begin_failed", "failed to start phase2 desktop task completion")
	}
	defer tx.Rollback(ctx)

	var (
		desktopID       uuid.UUID
		workspaceID     int64
		activeAttemptID pgtype.UUID
	)
	if err := tx.QueryRow(ctx, `
		SELECT desktop_id, workspace_id, active_attempt_id
		FROM desktop_tasks
		WHERE kind = 'monitor'
		  AND monitor_task_id = $1
		  AND status IN ('in_progress', 'unknown')
		ORDER BY created_at DESC, id DESC
		LIMIT 1
		FOR UPDATE
	`, task.ID).Scan(&desktopID, &workspaceID, &activeAttemptID); err != nil {
		if err == pgx.ErrNoRows {
			return nil
		}
		return response.ErrInternal(50041, "desktop_task_lookup_failed", "failed to inspect phase2 desktop task state")
	}

	resultJSON, err := marshalOptionalJSON(buildDesktopMonitorTaskResultValue(task, req, runStatus))
	if err != nil {
		return response.ErrInternal(50041, "desktop_task_payload_encode_failed", "failed to encode phase2 desktop task result payload")
	}

	var errorJSON []byte
	if runStatus == "failed" {
		errorJSON, err = marshalOptionalJSON(buildDesktopMonitorTaskErrorValue(task, req))
		if err != nil {
			return response.ErrInternal(50041, "desktop_task_payload_encode_failed", "failed to encode phase2 desktop task error payload")
		}
	}

	if _, err := tx.Exec(ctx, `
		UPDATE desktop_tasks
		SET status = $2,
		    result = $3,
		    error = $4,
		    active_attempt_id = NULL,
		    lease_token_hash = NULL,
		    lease_expires_at = NULL,
		    updated_at = NOW()
		WHERE desktop_id = $1
	`, desktopID, runStatus, resultJSON, errorJSON); err != nil {
		return response.ErrInternal(50041, "desktop_task_complete_failed", "failed to finalize phase2 desktop task")
	}

	taskRepo := repository.NewDesktopTaskRepository(tx)
	if activeAttemptID.Valid {
		resolvedAttemptID, convErr := uuid.FromBytes(activeAttemptID.Bytes[:])
		if convErr != nil {
			return response.ErrInternal(50041, "desktop_task_attempt_decode_failed", "failed to decode phase2 desktop task attempt id")
		}
		if finishErr := taskRepo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
			TaskID:      desktopID,
			AttemptID:   resolvedAttemptID,
			FinalStatus: &runStatus,
			Error:       errorJSON,
		}); finishErr != nil {
			return response.ErrInternal(50041, "desktop_task_attempt_finish_failed", "failed to finalize phase2 desktop task attempt")
		}
	}

	finalizedTask, err := taskRepo.GetByDesktopID(ctx, desktopID, workspaceID)
	if err != nil {
		return response.ErrInternal(50041, "desktop_task_reload_failed", "failed to reload finalized phase2 desktop task")
	}

	if err := tx.Commit(ctx); err != nil {
		return response.ErrInternal(50041, "desktop_task_complete_commit_failed", "failed to commit phase2 desktop task completion")
	}

	title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(
		finalizedTask.Kind,
		finalizedTask.Payload,
	)
	if publishErr := publishDesktopTaskEvent(ctx, s.rabbitMQ, DesktopTaskEvent{
		Type:                "task_completed",
		TaskID:              finalizedTask.DesktopID.String(),
		JobID:               finalizedTask.JobID.String(),
		WorkspaceID:         finalizedTask.WorkspaceID,
		TargetAccountID:     finalizedTask.TargetAccountID.String(),
		TargetClientID:      finalizedTask.TargetClientID.String(),
		Platform:            finalizedTask.Platform,
		Title:               title,
		BusinessDate:        businessDate,
		SchedulerGroupKey:   schedulerGroupKey,
		QuestionText:        questionText,
		Status:              finalizedTask.Status,
		Kind:                finalizedTask.Kind,
		Priority:            finalizedTask.Priority,
		Lane:                finalizedTask.Lane,
		InterruptGeneration: finalizedTask.InterruptGeneration,
		UpdatedAt:           finalizedTask.UpdatedAt,
	}); publishErr != nil && s.logger != nil {
		s.logger.Warn("phase2 desktop task completion event publish failed",
			zap.Int64("monitor_task_id", task.ID),
			zap.String("desktop_task_id", finalizedTask.DesktopID.String()),
			zap.Error(publishErr),
		)
	}

	return nil
}

func (s *MonitoringCallbackService) replaceCitationFacts(ctx context.Context, tx pgx.Tx, tenantID, runID int64) error {
	if _, err := tx.Exec(ctx, `
		DELETE FROM monitoring_citation_facts
		WHERE tenant_id = $1 AND run_id = $2
	`, tenantID, runID); err != nil {
		return response.ErrInternal(50041, "citation_cleanup_failed", "failed to reset existing citation facts")
	}
	return nil
}

func (s *MonitoringCallbackService) loadMonitoringBrandName(ctx context.Context, tenantID, brandID int64) (string, error) {
	var brandName string
	if err := s.businessPool.QueryRow(ctx, `
		SELECT name
		FROM brands
		WHERE id = $1
		  AND tenant_id = $2
		  AND deleted_at IS NULL
	`, brandID, tenantID).Scan(&brandName); err != nil {
		if err == pgx.ErrNoRows {
			return "", response.ErrInternal(50041, "brand_lookup_failed", "failed to resolve monitoring brand")
		}
		return "", response.ErrInternal(50041, "brand_lookup_failed", "failed to resolve monitoring brand")
	}
	return strings.TrimSpace(brandName), nil
}

func (s *MonitoringCallbackService) enrichMonitoringParsePayload(ctx context.Context, task *monitoringCollectTask, req MonitoringTaskResultRequest) (MonitoringTaskResultRequest, string, error) {
	brandName, err := s.loadMonitoringBrandName(ctx, task.TenantID, task.BrandID)
	if err != nil {
		return req, "", err
	}

	answer := monitoringStringValue(req.Answer)
	stored := buildMonitoringStoredParseFields(req)
	summary := resolveMonitoringAnswerParseSummary(answer, brandName, stored)
	parseVersion := resolveMonitoringParseVersion(req)

	if shouldUseMonitoringLLMParse(req) && strings.TrimSpace(answer) != "" && s.llm != nil && s.llm.Validate() == nil {
		llmSummary, llmErr := parseMonitoringAnswerWithLLM(ctx, s.llm, answer, brandName)
		if llmErr != nil {
			if s.logger != nil {
				s.logger.Warn("monitoring answer llm parse failed",
					zap.Int64("task_id", task.ID),
					zap.Int64("tenant_id", task.TenantID),
					zap.Int64("brand_id", task.BrandID),
					zap.Int64("question_id", task.QuestionID),
					zap.String("ai_platform_id", task.AIPlatformID),
					zap.Error(llmErr),
				)
			}
		} else {
			summary = overlayMonitoringStoredParseSummary(llmSummary, stored)
			if !hasMonitoringParsePayload(req) {
				parseVersion = "server_llm_v1"
			}
		}
	}

	enriched := req
	enriched.BrandMentioned = monitoringBoolPointer(summary.BrandMentioned)
	enriched.BrandMentionPosition = optionalString(summary.BrandMentionPosition)
	enriched.FirstRecommended = monitoringBoolPointer(summary.FirstRecommended)
	enriched.SentimentLabel = optionalString(summary.SentimentLabel)
	enriched.MatchedBrandTerms = append([]string(nil), summary.MatchedBrandTerms...)
	return enriched, parseVersion, nil
}

func buildMonitoringStoredParseFields(req MonitoringTaskResultRequest) monitoringStoredParseFields {
	return monitoringStoredParseFields{
		BrandMentioned:       cloneBool(req.BrandMentioned),
		BrandMentionPosition: cloneString(req.BrandMentionPosition),
		FirstRecommended:     cloneBool(req.FirstRecommended),
		SentimentLabel:       cloneString(req.SentimentLabel),
		MatchedBrandTerms:    append([]string(nil), req.MatchedBrandTerms...),
	}
}

func resolveMonitoringParseVersion(req MonitoringTaskResultRequest) string {
	if hasMonitoringParsePayload(req) {
		return "plugin_v1"
	}
	return "server_heuristic_v1"
}

func shouldUseMonitoringLLMParse(req MonitoringTaskResultRequest) bool {
	return !hasCompleteMonitoringParsePayload(req)
}

func (s *MonitoringCallbackService) buildCitationFacts(ctx context.Context, tx pgx.Tx, tenantID int64, platformID string, inputs []MonitoringSourceItem) ([]monitoringCitationFact, error) {
	if len(inputs) == 0 {
		return nil, nil
	}

	resolveContentCitations := monitoringPlatformResolvesContentCitations(platformID)
	aliasMap := map[string]monitoringAliasResolution{}
	if resolveContentCitations {
		normalizedURLs := make([]string, 0, len(inputs))
		seenURLs := make(map[string]struct{}, len(inputs))
		for _, item := range inputs {
			key := strings.TrimSpace(monitoringStringValue(item.NormalizedURL))
			if key == "" {
				key = normalizeCitationURL(item.URL)
			}
			if key == "" {
				continue
			}
			if _, ok := seenURLs[key]; ok {
				continue
			}
			seenURLs[key] = struct{}{}
			normalizedURLs = append(normalizedURLs, key)
		}

		var err error
		aliasMap, err = s.loadAliasResolutions(ctx, tx, tenantID, normalizedURLs)
		if err != nil {
			return nil, err
		}
	}

	result := make([]monitoringCitationFact, 0, len(inputs))
	seenFacts := make(map[string]struct{}, len(inputs))
	for _, input := range inputs {
		fact, ok := buildMonitoringCitationFact(input, aliasMap, resolveContentCitations)
		if !ok {
			continue
		}

		dedupKey := fact.NormalizedURL + "|" + strconv.FormatInt(derefInt64(fact.ArticleID), 10) + "|" + strconv.FormatInt(derefInt64(fact.PublishRecordID), 10)
		if _, exists := seenFacts[dedupKey]; exists {
			continue
		}
		seenFacts[dedupKey] = struct{}{}
		result = append(result, fact)
	}

	return result, nil
}

func (s *MonitoringCallbackService) loadAliasResolutions(ctx context.Context, tx pgx.Tx, tenantID int64, normalizedURLs []string) (map[string]monitoringAliasResolution, error) {
	result := make(map[string]monitoringAliasResolution)
	if len(normalizedURLs) == 0 {
		return result, nil
	}

	rows, err := tx.Query(ctx, `
		SELECT normalized_url, article_id, publish_record_id
		FROM monitoring_article_url_aliases
		WHERE tenant_id = $1
		  AND normalized_url = ANY($2)
	`, tenantID, normalizedURLs)
	if err != nil {
		return nil, response.ErrInternal(50041, "alias_lookup_failed", "failed to resolve article aliases")
	}
	defer rows.Close()

	for rows.Next() {
		var normalizedURL string
		var articleID sql.NullInt64
		var publishRecordID sql.NullInt64
		if scanErr := rows.Scan(&normalizedURL, &articleID, &publishRecordID); scanErr != nil {
			return nil, response.ErrInternal(50041, "alias_scan_failed", "failed to parse article alias resolution")
		}
		result[normalizedURL] = monitoringAliasResolution{
			ArticleID:        nullableInt64Value(articleID),
			PublishRecordID:  nullableInt64Value(publishRecordID),
			NormalizedURLKey: normalizedURL,
		}
	}

	return result, nil
}

func (s *MonitoringCallbackService) insertCitationFact(ctx context.Context, tx pgx.Tx, task *monitoringCollectTask, runID int64, fact monitoringCitationFact) error {
	if _, err := tx.Exec(ctx, `
		INSERT INTO monitoring_citation_facts (
			tenant_id, run_id, brand_id, ai_platform_id, business_date,
			cited_url, cited_title, normalized_url, host, registrable_domain,
			subdomain, suffix, site_key, article_id, publish_record_id,
			resolution_status, resolution_confidence
		)
		VALUES (
			$1, $2, $3, $4, $5::date,
			$6, $7, $8, $9, $10,
			$11, $12, $13, $14, $15,
			$16, $17
		)
	`,
		task.TenantID,
		runID,
		task.BrandID,
		task.AIPlatformID,
		task.BusinessDate.Format("2006-01-02"),
		fact.CitedURL,
		nullableString(fact.CitedTitle),
		fact.NormalizedURL,
		fact.Host,
		fact.RegistrableDomain,
		nullableString(fact.Subdomain),
		nullableString(fact.Suffix),
		fact.SiteKey,
		nullableInt64(fact.ArticleID),
		nullableInt64(fact.PublishRecordID),
		fact.ResolutionStatus,
		fact.ResolutionConfidence,
	); err != nil {
		return response.ErrInternal(50041, "citation_insert_failed", "failed to persist citation facts")
	}
	return nil
}

func (s *MonitoringCallbackService) updateCollectTaskStatus(ctx context.Context, tx pgx.Tx, taskID int64, status string, completedAt time.Time, errorMessage *string) error {
	if _, err := tx.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET status = $2,
		    callback_received_at = NOW(),
		    completed_at = $3,
		    error_message = $4,
		    updated_at = NOW()
		WHERE id = $1
	`, taskID, status, completedAt, nullableString(errorMessage)); err != nil {
		return response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task status")
	}
	return nil
}

func (s *MonitoringCallbackService) loadExistingRunSummary(ctx context.Context, task *monitoringCollectTask) (*int64, int, int, error) {
	var runID sql.NullInt64
	if err := s.monitoringPool.QueryRow(ctx, `
		SELECT id
		FROM question_monitor_runs
		WHERE tenant_id = $1
		  AND brand_id = $2
		  AND question_id = $3
		  AND ai_platform_id = $4
		  AND collector_type = $5
		  AND run_mode = $6
		  AND business_date = $7::date
		LIMIT 1
	`,
		task.TenantID,
		task.BrandID,
		task.QuestionID,
		task.AIPlatformID,
		task.CollectorType,
		task.RunMode,
		task.BusinessDate.Format("2006-01-02"),
	).Scan(&runID); err != nil && err != pgx.ErrNoRows {
		return nil, 0, 0, response.ErrInternal(50041, "run_lookup_failed", "failed to inspect existing monitoring run")
	}

	if !runID.Valid {
		return nil, 0, 0, nil
	}

	var citationSourceCount int
	var contentCitationCount int
	if err := s.monitoringPool.QueryRow(ctx, `
		SELECT
			COUNT(*)::int AS citation_source_count,
			COUNT(*) FILTER (WHERE article_id IS NOT NULL AND $3::text <> 'qwen')::int AS content_citation_count
		FROM monitoring_citation_facts
		WHERE tenant_id = $1
		  AND run_id = $2
	`, task.TenantID, runID.Int64, normalizeMonitoringPlatformID(task.AIPlatformID)).Scan(&citationSourceCount, &contentCitationCount); err != nil {
		return nil, 0, 0, response.ErrInternal(50041, "citation_lookup_failed", "failed to inspect existing citation facts")
	}

	value := runID.Int64
	return &value, citationSourceCount, contentCitationCount, nil
}

func (s *MonitoringCallbackService) touchInstallation(ctx context.Context, installationID string) error {
	_, err := s.businessPool.Exec(ctx, `
		UPDATE desktop_clients
		SET last_seen_at = NOW()
		WHERE id = $1::uuid
	`, installationID)
	return err
}

func (s *MonitoringCallbackService) publishTaskResultEnvelope(ctx context.Context, payload []byte) error {
	if s == nil || s.rabbitMQ == nil {
		return response.ErrServiceUnavailable(50341, "monitoring_result_queue_unavailable", "rabbitmq client is not configured")
	}
	if err := s.rabbitMQ.PublishMonitorResultIngest(ctx, payload); err != nil {
		return response.ErrServiceUnavailable(50341, "monitoring_result_queue_unavailable", err.Error())
	}
	return nil
}

func (s *MonitoringCallbackService) PublishTaskResultEnvelope(ctx context.Context, payload []byte) error {
	return s.publishTaskResultEnvelope(ctx, payload)
}

func (s *MonitoringCallbackService) tryPatchTaskResultQueueMeta(taskID int64, receivedAt time.Time, patch monitoringTaskResultQueueMetaPatch) {
	if s == nil || s.monitoringPool == nil {
		return
	}

	patchJSON, err := json.Marshal(patch)
	if err != nil {
		if s.logger != nil {
			s.logger.Warn("monitoring queue metadata patch encode failed", zap.Error(err), zap.Int64("task_id", taskID))
		}
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), monitoringTaskResultQueueIOTimeout)
	defer cancel()

	if _, err := s.monitoringPool.Exec(ctx, `
		UPDATE monitoring_collect_tasks
		SET request_payload_json = (
				CASE
					WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
						THEN '{}'::jsonb
					ELSE request_payload_json
				END
			) || $3::jsonb,
			updated_at = NOW()
		WHERE id = $1
		  AND callback_received_at = $2
	`, taskID, receivedAt, patchJSON); err != nil && s.logger != nil {
		s.logger.Warn("monitoring queue metadata patch failed", zap.Error(err), zap.Int64("task_id", taskID))
	}
}

func (s *MonitoringCallbackService) PatchTaskResultQueueMeta(taskID int64, receivedAt time.Time, patch MonitoringTaskResultQueueMetaPatch) {
	s.tryPatchTaskResultQueueMeta(taskID, receivedAt, patch)
}

func buildMonitoringRawPayload(platformID string, req MonitoringTaskResultRequest) ([]byte, []MonitoringSourceItem) {
	payload := make(map[string]interface{}, len(req.RawResponseJSON)+4)
	for key, value := range req.RawResponseJSON {
		payload[key] = value
	}

	if _, ok := payload["answer"]; !ok && req.Answer != nil {
		payload["answer"] = strings.TrimSpace(*req.Answer)
	}
	if _, ok := payload["citations"]; !ok && len(req.Citations) > 0 {
		payload["citations"] = req.Citations
	}
	if _, ok := payload["search_results"]; !ok && len(req.SearchResults) > 0 {
		payload["search_results"] = req.SearchResults
	}
	if _, ok := payload["status"]; !ok && strings.TrimSpace(req.Status) != "" {
		payload["status"] = strings.TrimSpace(req.Status)
	}

	sourceInputs := buildMonitoringCitationSourceInputs(platformID, req, payload)
	if len(payload) == 0 {
		return nil, sourceInputs
	}

	data, err := json.Marshal(payload)
	if err != nil {
		return nil, sourceInputs
	}
	return data, sourceInputs
}

func buildMonitoringCitationSourceInputs(platformID string, req MonitoringTaskResultRequest, payload map[string]interface{}) []MonitoringSourceItem {
	includeSearchResults := !monitoringPlatformUsesDedicatedCitationPanel(platformID)

	sourceInputs := append([]MonitoringSourceItem{}, req.Citations...)
	if includeSearchResults {
		sourceInputs = append(sourceInputs, req.SearchResults...)
	}
	sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["citations"])...)
	sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["references"])...)
	sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["sources"])...)
	sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["thought_references"])...)
	sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["thinking_references"])...)
	if includeSearchResults {
		sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["search_results"])...)
		sourceInputs = append(sourceInputs, extractMonitoringSourceItems(payload["searchResults"])...)
	}
	return sourceInputs
}

func monitoringPlatformUsesDedicatedCitationPanel(platformID string) bool {
	return false
}

func monitoringPlatformResolvesContentCitations(platformID string) bool {
	switch normalizeMonitoringPlatformID(platformID) {
	case "qwen":
		return false
	default:
		return true
	}
}

func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID string, receivedAt time.Time, req MonitoringTaskResultRequest) monitoringTaskResultEnvelope {
	return monitoringTaskResultEnvelope{
		Version:               "v1",
		TaskID:                task.ID,
		TenantID:              task.TenantID,
		InstallationID:        installationID,
		ReceivedAt:            receivedAt,
		QueueStatus:           "pending",
		QueueClaimedAt:        nil,
		QueueEnqueuedAt:       nil,
		LastQueuePublishError: nil,
		ReceivedAlertedAt:     nil,
		ReceivedAlertCount:    0,
		CallbackResult:        req,
	}
}

func extractMonitoringSourceItems(value interface{}) []MonitoringSourceItem {
	items, ok := value.([]interface{})
	if !ok {
		return nil
	}

	result := make([]MonitoringSourceItem, 0, len(items))
	for _, item := range items {
		switch typed := item.(type) {
		case string:
			if strings.TrimSpace(typed) == "" {
				continue
			}
			result = append(result, MonitoringSourceItem{URL: typed})
		case map[string]interface{}:
			entry := MonitoringSourceItem{
				URL:                  firstString(typed, "url", "cited_url", "link"),
				Title:                optionalString(firstString(typed, "title", "cited_title", "name", "text")),
				SiteName:             optionalString(firstString(typed, "site_name", "siteName")),
				SiteKey:              optionalString(firstString(typed, "site_key")),
				NormalizedURL:        optionalString(firstString(typed, "normalized_url")),
				Host:                 optionalString(firstString(typed, "host")),
				RegistrableDomain:    optionalString(firstString(typed, "registrable_domain")),
				Subdomain:            optionalString(firstString(typed, "subdomain")),
				Suffix:               optionalString(firstString(typed, "suffix")),
				ResolutionStatus:     optionalString(firstString(typed, "resolution_status")),
				ResolutionConfidence: optionalString(firstString(typed, "resolution_confidence")),
			}
			if articleID := firstInt64(typed, "article_id"); articleID != nil {
				entry.ArticleID = articleID
			}
			if publishRecordID := firstInt64(typed, "publish_record_id"); publishRecordID != nil {
				entry.PublishRecordID = publishRecordID
			}
			result = append(result, entry)
		}
	}

	return result
}

func buildMonitoringCitationFact(input MonitoringSourceItem, aliasMap map[string]monitoringAliasResolution, resolveContentCitations bool) (monitoringCitationFact, bool) {
	rawURL := strings.TrimSpace(input.URL)
	normalizedURL := strings.TrimSpace(monitoringStringValue(input.NormalizedURL))
	if normalizedURL == "" {
		normalizedURL = normalizeCitationURL(rawURL)
	}

	host := strings.TrimSpace(monitoringStringValue(input.Host))
	registrableDomain := strings.TrimSpace(monitoringStringValue(input.RegistrableDomain))
	subdomain := cloneString(input.Subdomain)
	suffix := cloneString(input.Suffix)

	if parsedHost, parsedRegistrableDomain, parsedSubdomain, parsedSuffix := deriveCitationURLParts(rawURL, normalizedURL); host == "" {
		host = parsedHost
		if registrableDomain == "" {
			registrableDomain = parsedRegistrableDomain
		}
		if subdomain == nil && parsedSubdomain != nil {
			subdomain = parsedSubdomain
		}
		if suffix == nil && parsedSuffix != nil {
			suffix = parsedSuffix
		}
	}

	if host == "" {
		host = strings.TrimSpace(monitoringStringValue(input.SiteKey))
	}
	if host == "" {
		host = registrableDomain
	}
	if registrableDomain == "" {
		registrableDomain = host
	}
	if normalizedURL == "" {
		normalizedURL = strings.TrimSpace(rawURL)
	}
	if rawURL == "" {
		rawURL = normalizedURL
	}

	if rawURL == "" || normalizedURL == "" || host == "" || registrableDomain == "" {
		return monitoringCitationFact{}, false
	}

	siteKey := strings.TrimSpace(monitoringStringValue(input.SiteKey))
	if siteKey == "" {
		siteKey = registrableDomain
	}

	var articleID *int64
	var publishRecordID *int64
	if resolveContentCitations {
		articleID = cloneInt64(input.ArticleID)
		publishRecordID = cloneInt64(input.PublishRecordID)
		if alias, ok := aliasMap[normalizedURL]; ok {
			if articleID == nil {
				articleID = cloneInt64(alias.ArticleID)
			}
			if publishRecordID == nil {
				publishRecordID = cloneInt64(alias.PublishRecordID)
			}
		}
	}

	resolutionStatus := strings.TrimSpace(monitoringStringValue(input.ResolutionStatus))
	if resolutionStatus == "" {
		resolutionStatus = "resolved"
	}
	resolutionConfidence := strings.TrimSpace(monitoringStringValue(input.ResolutionConfidence))
	if resolutionConfidence == "" {
		resolutionConfidence = "high"
	}

	return monitoringCitationFact{
		CitedURL:             rawURL,
		CitedTitle:           cloneString(input.Title),
		NormalizedURL:        normalizedURL,
		Host:                 host,
		RegistrableDomain:    registrableDomain,
		Subdomain:            subdomain,
		Suffix:               suffix,
		SiteKey:              siteKey,
		ArticleID:            articleID,
		PublishRecordID:      publishRecordID,
		ResolutionStatus:     resolutionStatus,
		ResolutionConfidence: resolutionConfidence,
	}, true
}

func normalizeRunStatus(value string) string {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "", "succeeded", "success", "completed":
		return "succeeded"
	case "failed", "error":
		return "failed"
	default:
		return ""
	}
}

func validateMonitoringTaskResultSubmission(platformID string, runStatus string, req MonitoringTaskResultRequest) error {
	if runStatus != "succeeded" {
		return nil
	}
	answer := strings.TrimSpace(monitoringStringValue(req.Answer))
	if answer == "" {
		return response.ErrBadRequest(40041, "invalid_monitoring_result_answer", "succeeded monitoring result requires a non-empty answer")
	}
	if monitoringCitationMarkerPattern.MatchString(answer) {
		_, sourceInputs := buildMonitoringRawPayload(platformID, req)
		if len(sourceInputs) == 0 {
			return response.ErrBadRequest(40041, "invalid_monitoring_result_citations", "succeeded monitoring result with citation markers requires citation sources")
		}
	}
	return nil
}

func normalizeMonitoringLeaseLimit(value int) int {
	if value <= 0 {
		return defaultMonitoringLeaseLimit
	}
	if value > maxMonitoringLeaseLimit {
		return maxMonitoringLeaseLimit
	}
	return value
}

func normalizeMonitoringLeasePlatformIDs(values []string) []string {
	if len(values) == 0 {
		return nil
	}

	result := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		platformID := strings.TrimSpace(value)
		if platformID == "" {
			continue
		}
		if _, ok := seen[platformID]; ok {
			continue
		}
		seen[platformID] = struct{}{}
		result = append(result, platformID)
	}
	if len(result) == 0 {
		return nil
	}
	return result
}

func monitoringHeartbeatPlatformIDs(platforms []MonitoringHeartbeatPlatform) []string {
	if len(platforms) == 0 {
		return nil
	}
	result := make([]string, 0, len(platforms))
	for _, platform := range platforms {
		platformID := strings.TrimSpace(platform.AIPlatformID)
		if platformID != "" {
			result = append(result, platformID)
		}
	}
	return result
}

func normalizeHeartbeatPlatforms(items []MonitoringHeartbeatPlatform, fallbackDetectedAt time.Time) ([]MonitoringHeartbeatPlatform, error) {
	if len(items) == 0 {
		return nil, nil
	}

	result := make([]MonitoringHeartbeatPlatform, 0, len(items))
	seen := make(map[string]struct{}, len(items))
	for _, item := range items {
		platformID := strings.TrimSpace(item.AIPlatformID)
		if platformID == "" {
			return nil, response.ErrBadRequest(40041, "invalid_platform_status", "ai_platform_id is required")
		}

		status, connected, accessible, err := normalizeMonitoringAccessStatus(item.AccessStatus, item.Connected, item.Accessible)
		if err != nil {
			return nil, err
		}
		item.AIPlatformID = platformID
		item.AccessStatus = &status
		item.Connected = &connected
		item.Accessible = &accessible
		if item.DetectedAt == nil || item.DetectedAt.IsZero() {
			value := fallbackDetectedAt
			item.DetectedAt = &value
		} else {
			value := item.DetectedAt.UTC()
			item.DetectedAt = &value
		}
		if item.ReasonCode != nil {
			item.ReasonCode = optionalString(*item.ReasonCode)
		}
		if item.ReasonMessage != nil {
			item.ReasonMessage = optionalString(*item.ReasonMessage)
		}

		if _, ok := seen[platformID]; ok {
			continue
		}
		seen[platformID] = struct{}{}
		result = append(result, item)
	}

	return result, nil
}

func normalizeMonitoringAccessStatus(accessStatus *string, connected *bool, accessible *bool) (string, bool, bool, error) {
	if accessStatus != nil {
		switch strings.ToLower(strings.TrimSpace(*accessStatus)) {
		case "accessible":
			return "accessible", true, true, nil
		case "not_logged_in":
			return "not_logged_in", false, false, nil
		case "unavailable":
			return "unavailable", derefBool(connected), false, nil
		case "":
		default:
			return "", false, false, response.ErrBadRequest(40041, "invalid_access_status", "access_status must be accessible, not_logged_in, or unavailable")
		}
	}

	if accessible != nil {
		if *accessible {
			return "accessible", true, true, nil
		}
		if connected != nil && !*connected {
			return "not_logged_in", false, false, nil
		}
		return "unavailable", derefBool(connected), false, nil
	}

	if connected != nil {
		if *connected {
			return "accessible", true, true, nil
		}
		return "not_logged_in", false, false, nil
	}

	return "", false, false, response.ErrBadRequest(40041, "invalid_access_status", "access status requires access_status, connected, or accessible")
}

func (s *MonitoringCallbackService) validateCollectTaskExecution(
	ctx context.Context,
	task *monitoringCollectTask,
	installationID string,
	leaseToken string,
) error {
	if task == nil {
		return response.ErrBadRequest(40041, "invalid_task_id", "monitoring task is required")
	}
	if task.ExecutionOwner == "desktop_tasks" {
		return s.validateDesktopMonitorExecutionLease(ctx, task, installationID, leaseToken)
	}
	return validateTaskLease(task, installationID, leaseToken)
}

func (s *MonitoringCallbackService) validateDesktopMonitorExecutionLease(
	ctx context.Context,
	task *monitoringCollectTask,
	installationID string,
	leaseToken string,
) error {
	desktopExecution, err := s.loadActiveDesktopMonitorExecutionLease(ctx, task.ID)
	if err != nil {
		return err
	}
	if desktopExecution == nil {
		return response.ErrConflict(40941, "desktop_task_missing", "phase2 monitor desktop task is not active")
	}
	if desktopExecution.TargetClientID.String() != installationID {
		return response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
	}
	if desktopExecution.Status != "in_progress" {
		return response.ErrConflict(40941, "task_status_invalid", "phase2 monitor desktop task is not currently in progress")
	}
	if len(desktopExecution.LeaseTokenHash) == 0 {
		return response.ErrConflict(40941, "lease_token_missing", "phase2 monitor desktop task does not have an active lease token")
	}
	if !bytes.Equal(HashDesktopClientToken(strings.TrimSpace(leaseToken)), desktopExecution.LeaseTokenHash) {
		return response.ErrUnauthorized(40141, "invalid_lease_token", "lease token is invalid")
	}
	return nil
}

func validateTaskLease(task *monitoringCollectTask, installationID string, leaseToken string) error {
	if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installationID {
		return response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
	}

	if strings.TrimSpace(task.LeaseTokenHash.String) == "" {
		return response.ErrConflict(40941, "lease_token_missing", "task does not have an active lease token")
	}

	if sharedauth.HashToken(strings.TrimSpace(leaseToken)) != task.LeaseTokenHash.String {
		return response.ErrUnauthorized(40141, "invalid_lease_token", "lease token is invalid")
	}

	return nil
}

func monitoringTaskReadyForQueuedResult(task *monitoringCollectTask) bool {
	if task == nil {
		return false
	}
	if task.ExecutionOwner == "desktop_tasks" {
		return task.Status == "pending" && task.CallbackReceivedAt.Valid
	}
	return task.Status == "received"
}

func normalizeSkipReason(value *string) string {
	if value == nil {
		return "plugin_skipped"
	}
	trimmed := strings.TrimSpace(*value)
	if trimmed == "" {
		return "plugin_skipped"
	}
	return trimmed
}

func hasMonitoringParsePayload(req MonitoringTaskResultRequest) bool {
	return req.BrandMentioned != nil ||
		strings.TrimSpace(monitoringStringValue(req.BrandMentionPosition)) != "" ||
		req.FirstRecommended != nil ||
		strings.TrimSpace(monitoringStringValue(req.SentimentLabel)) != "" ||
		len(req.MatchedBrandTerms) > 0
}

func hasCompleteMonitoringParsePayload(req MonitoringTaskResultRequest) bool {
	return req.BrandMentioned != nil &&
		strings.TrimSpace(monitoringStringValue(req.BrandMentionPosition)) != "" &&
		req.FirstRecommended != nil &&
		strings.TrimSpace(monitoringStringValue(req.SentimentLabel)) != "" &&
		req.MatchedBrandTerms != nil
}

func normalizeCitationURL(raw string) string {
	trimmed := strings.TrimSpace(raw)
	if trimmed == "" {
		return ""
	}

	parsed, err := url.Parse(trimmed)
	if err != nil {
		return trimmed
	}

	parsed.Host = strings.ToLower(parsed.Host)
	parsed.Fragment = ""
	parsed.RawFragment = ""
	parsed.RawQuery = ""

	if parsed.Path != "/" {
		parsed.Path = strings.TrimRight(parsed.Path, "/")
	}

	return parsed.String()
}

func deriveCitationURLParts(rawURL, normalizedURL string) (string, string, *string, *string) {
	candidate := strings.TrimSpace(normalizedURL)
	if candidate == "" {
		candidate = strings.TrimSpace(rawURL)
	}
	if candidate == "" {
		return "", "", nil, nil
	}

	parsed, err := url.Parse(candidate)
	if err != nil {
		return "", "", nil, nil
	}

	host := strings.ToLower(strings.TrimSpace(parsed.Hostname()))
	if host == "" {
		return "", "", nil, nil
	}

	registrableDomain, err := publicsuffix.EffectiveTLDPlusOne(host)
	if err != nil {
		registrableDomain = host
	}

	suffixValue, _ := publicsuffix.PublicSuffix(host)
	var suffix *string
	if suffixValue != "" {
		suffix = &suffixValue
	}

	var subdomain *string
	if registrableDomain != "" && host != registrableDomain {
		value := strings.TrimSuffix(host, "."+registrableDomain)
		value = strings.TrimSuffix(value, ".")
		if value != "" {
			subdomain = &value
		}
	}

	return host, registrableDomain, subdomain, suffix
}

func firstString(values map[string]interface{}, keys ...string) string {
	for _, key := range keys {
		raw, ok := values[key]
		if !ok {
			continue
		}
		switch typed := raw.(type) {
		case string:
			if trimmed := strings.TrimSpace(typed); trimmed != "" {
				return trimmed
			}
		}
	}
	return ""
}

func firstInt64(values map[string]interface{}, keys ...string) *int64 {
	for _, key := range keys {
		raw, ok := values[key]
		if !ok {
			continue
		}
		switch typed := raw.(type) {
		case float64:
			value := int64(typed)
			return &value
		case int64:
			value := typed
			return &value
		case int:
			value := int64(typed)
			return &value
		case string:
			trimmed := strings.TrimSpace(typed)
			if trimmed == "" {
				continue
			}
			value, err := strconv.ParseInt(trimmed, 10, 64)
			if err == nil {
				return &value
			}
		}
	}
	return nil
}

func optionalString(value string) *string {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	text := strings.TrimSpace(value)
	return &text
}

func ValidateMonitoringTaskResultEnvelope(payload []byte) error {
	_, err := decodeMonitoringTaskResultEnvelope(payload)
	return err
}

func monitoringStringValue(value *string) string {
	if value == nil {
		return ""
	}
	return strings.TrimSpace(*value)
}

func resolveHeartbeatDetectedAt(value *time.Time) time.Time {
	if value == nil || value.IsZero() {
		return time.Now().UTC()
	}
	return value.UTC()
}

func cloneString(value *string) *string {
	if value == nil {
		return nil
	}
	trimmed := strings.TrimSpace(*value)
	if trimmed == "" {
		return nil
	}
	copyValue := trimmed
	return &copyValue
}

func cloneBool(value *bool) *bool {
	if value == nil {
		return nil
	}
	copyValue := *value
	return &copyValue
}

func cloneInt64(value *int64) *int64 {
	if value == nil {
		return nil
	}
	copyValue := *value
	return &copyValue
}

func nullableBool(value *bool) interface{} {
	if value == nil {
		return nil
	}
	return *value
}

func monitoringBoolPointer(value bool) *bool {
	copyValue := value
	return &copyValue
}

func derefBool(value *bool) bool {
	if value == nil {
		return false
	}
	return *value
}

func derefInt64(value *int64) int64 {
	if value == nil {
		return 0
	}
	return *value
}