diff --git a/apps/desktop-client/src/main/adapters/deepseek.test.ts b/apps/desktop-client/src/main/adapters/deepseek.test.ts index ba9b71e..f46ecfe 100644 --- a/apps/desktop-client/src/main/adapters/deepseek.test.ts +++ b/apps/desktop-client/src/main/adapters/deepseek.test.ts @@ -69,6 +69,12 @@ describe('deepseek adapter helpers', () => { ) }) + it('does not use task title as monitoring question text', () => { + expect(() => extractQuestionText({ title: '监控任务 · DeepSeek' })).toThrow( + 'deepseek_question_text_missing', + ) + }) + it('dedupes sources by normalized url without hash fragments', () => { const items = dedupeSourceItems([ { diff --git a/apps/desktop-client/src/main/adapters/deepseek.ts b/apps/desktop-client/src/main/adapters/deepseek.ts index e2e6f2b..2163001 100644 --- a/apps/desktop-client/src/main/adapters/deepseek.ts +++ b/apps/desktop-client/src/main/adapters/deepseek.ts @@ -499,7 +499,6 @@ function extractQuestionText(payload: Record): string { payload.question, payload.prompt, payload.content, - payload.title, ] for (const candidate of candidates) { diff --git a/apps/desktop-client/src/main/adapters/doubao.test.ts b/apps/desktop-client/src/main/adapters/doubao.test.ts index 16a0cc0..eb33906 100644 --- a/apps/desktop-client/src/main/adapters/doubao.test.ts +++ b/apps/desktop-client/src/main/adapters/doubao.test.ts @@ -5,6 +5,7 @@ import { __doubaoTestUtils } from './doubao' const { buildSourceItem, dedupeSourceItems, + extractQuestionText, isNonCitationAssetDomain, isNonCitationAssetUrl, parseDoubaoStreamBody, @@ -12,6 +13,13 @@ const { } = __doubaoTestUtils describe('doubao adapter helpers', () => { + it('does not use task title as monitoring question text', () => { + expect(extractQuestionText({ question_text: '幽灵门轨道长度定制' })).toBe('幽灵门轨道长度定制') + expect(() => extractQuestionText({ title: '监控任务 · 豆包' })).toThrow( + 'doubao_question_text_missing', + ) + }) + it('filters byteimg and bytednsdoc asset CDN links from source items', () => { expect(isNonCitationAssetDomain('p3-spider-image-sign.byteimg.com')).toBe(true) expect(isNonCitationAssetDomain('lf3-static.bytednsdoc.com')).toBe(true) diff --git a/apps/desktop-client/src/main/adapters/doubao.ts b/apps/desktop-client/src/main/adapters/doubao.ts index 49cd381..915a42f 100644 --- a/apps/desktop-client/src/main/adapters/doubao.ts +++ b/apps/desktop-client/src/main/adapters/doubao.ts @@ -1221,7 +1221,6 @@ function extractQuestionText(payload: Record): string { payload.question, payload.prompt, payload.content, - payload.title, ] for (const candidate of candidates) { @@ -3256,6 +3255,7 @@ export const doubaoAdapter: MonitorAdapter = { export const __doubaoTestUtils = { buildSourceItem, dedupeSourceItems, + extractQuestionText, isNonCitationAssetDomain, isNonCitationAssetUrl, parseDoubaoStreamBody, diff --git a/apps/desktop-client/src/main/adapters/kimi.test.ts b/apps/desktop-client/src/main/adapters/kimi.test.ts index 43019b5..7181975 100644 --- a/apps/desktop-client/src/main/adapters/kimi.test.ts +++ b/apps/desktop-client/src/main/adapters/kimi.test.ts @@ -5,6 +5,7 @@ import { __kimiTestUtils } from './kimi' const { buildSourceItem, classifyKimiSources, + extractQuestionText, isKimiAnswerComplete, isKimiPromotionalAnswerNoise, mergeKimiSourceLinksIntoContentSnapshot, @@ -41,6 +42,13 @@ function buildSnapshot( } describe('kimi adapter helpers', () => { + it('does not use task title as monitoring question text', () => { + expect(extractQuestionText({ question_text: '幽灵门轨道长度定制' })).toBe('幽灵门轨道长度定制') + expect(() => extractQuestionText({ title: '监控任务 · Kimi' })).toThrow( + 'kimi_question_text_missing', + ) + }) + it('unwraps Kimi redirect URLs before storing source items', () => { expect( normalizeUrl('https://www.kimi.com/redirect?url=https%3A%2F%2Fexample.com%2Fsource%23ref'), diff --git a/apps/desktop-client/src/main/adapters/kimi.ts b/apps/desktop-client/src/main/adapters/kimi.ts index 2e05673..5ba0111 100644 --- a/apps/desktop-client/src/main/adapters/kimi.ts +++ b/apps/desktop-client/src/main/adapters/kimi.ts @@ -427,7 +427,6 @@ function extractQuestionText(payload: Record): string { payload.question, payload.prompt, payload.content, - payload.title, ] for (const candidate of candidates) { @@ -3288,6 +3287,7 @@ export const __kimiTestUtils = { buildSourceItem, classifyKimiSources, dedupeSourceItems, + extractQuestionText, isKimiAnswerComplete, isKimiPromotionalAnswerNoise, mergeKimiSourceLinksIntoContentSnapshot, diff --git a/apps/desktop-client/src/main/adapters/qwen.test.ts b/apps/desktop-client/src/main/adapters/qwen.test.ts new file mode 100644 index 0000000..98324dd --- /dev/null +++ b/apps/desktop-client/src/main/adapters/qwen.test.ts @@ -0,0 +1,17 @@ +import { describe, expect, it } from 'vitest' + +import { __qwenTestUtils } from './qwen' + +const { extractQuestionText } = __qwenTestUtils + +describe('qwen adapter helpers', () => { + it('extracts question text from monitoring payload', () => { + expect(extractQuestionText({ question_text: '幽灵门轨道长度定制' })).toBe('幽灵门轨道长度定制') + }) + + it('does not use task title as monitoring question text', () => { + expect(() => extractQuestionText({ title: '监控任务 · 通义千问' })).toThrow( + 'qwen_question_text_missing', + ) + }) +}) diff --git a/apps/desktop-client/src/main/adapters/qwen.ts b/apps/desktop-client/src/main/adapters/qwen.ts index 5df1a47..e01a5de 100644 --- a/apps/desktop-client/src/main/adapters/qwen.ts +++ b/apps/desktop-client/src/main/adapters/qwen.ts @@ -188,7 +188,6 @@ function extractQuestionText(payload: Record): string { payload.question, payload.prompt, payload.content, - payload.title, ] for (const candidate of candidates) { @@ -687,3 +686,7 @@ export const qwenAdapter: MonitorAdapter = { } }, } + +export const __qwenTestUtils = { + extractQuestionText, +} diff --git a/apps/desktop-client/src/main/adapters/wenxin.test.ts b/apps/desktop-client/src/main/adapters/wenxin.test.ts index 66f0b6d..45c33c0 100644 --- a/apps/desktop-client/src/main/adapters/wenxin.test.ts +++ b/apps/desktop-client/src/main/adapters/wenxin.test.ts @@ -4,6 +4,7 @@ import { __wenxinTestUtils } from './wenxin' const { buildSourceItem, + extractQuestionText, htmlTableToMarkdown, isObservationComplete, normalizeUrl, @@ -12,6 +13,13 @@ const { } = __wenxinTestUtils describe('wenxin adapter helpers', () => { + it('does not use task title as monitoring question text', () => { + expect(extractQuestionText({ question_text: '幽灵门轨道长度定制' })).toBe('幽灵门轨道长度定制') + expect(() => extractQuestionText({ title: '监控任务 · 文心一言' })).toThrow( + 'wenxin_question_text_missing', + ) + }) + it('parses conversation stream major and message events', () => { const sseBody = [ 'event:major', diff --git a/apps/desktop-client/src/main/adapters/wenxin.ts b/apps/desktop-client/src/main/adapters/wenxin.ts index f2edcff..f1ffacd 100644 --- a/apps/desktop-client/src/main/adapters/wenxin.ts +++ b/apps/desktop-client/src/main/adapters/wenxin.ts @@ -736,7 +736,6 @@ function extractQuestionText(payload: Record): string { payload.question, payload.prompt, payload.content, - payload.title, ] for (const candidate of candidates) { @@ -2555,6 +2554,7 @@ export const wenxinAdapter: MonitorAdapter = { export const __wenxinTestUtils = { buildSourceItem, + extractQuestionText, htmlTableToMarkdown, isObservationComplete, normalizeUrl, diff --git a/apps/desktop-client/src/main/adapters/yuanbao.test.ts b/apps/desktop-client/src/main/adapters/yuanbao.test.ts index 45fb756..ad79711 100644 --- a/apps/desktop-client/src/main/adapters/yuanbao.test.ts +++ b/apps/desktop-client/src/main/adapters/yuanbao.test.ts @@ -41,6 +41,9 @@ describe('yuanbao adapter helpers', () => { it('rejects browser and build diagnostics as monitoring questions', () => { expect(extractQuestionText({ question_text: '合肥全屋定制推荐' })).toBe('合肥全屋定制推荐') + expect(() => extractQuestionText({ title: '监控任务 · 混元 / 元宝' })).toThrow( + 'yuanbao_question_text_missing', + ) expect(() => extractQuestionText({ question_text: diff --git a/apps/desktop-client/src/main/adapters/yuanbao.ts b/apps/desktop-client/src/main/adapters/yuanbao.ts index 4f16312..e751d5b 100644 --- a/apps/desktop-client/src/main/adapters/yuanbao.ts +++ b/apps/desktop-client/src/main/adapters/yuanbao.ts @@ -769,7 +769,6 @@ function extractQuestionText(payload: Record): string { payload.question, payload.prompt, payload.content, - payload.title, ] let rejectedDiagnostic = false diff --git a/server/internal/scheduler/schedule_dispatch_worker.go b/server/internal/scheduler/schedule_dispatch_worker.go index 6bf9309..8ba1d97 100644 --- a/server/internal/scheduler/schedule_dispatch_worker.go +++ b/server/internal/scheduler/schedule_dispatch_worker.go @@ -398,7 +398,7 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC } now := time.Now() - tasks := make([]dueScheduleTask, 0, runtimeCfg.BatchSize) + candidates := make([]dueScheduleTask, 0, runtimeCfg.BatchSize) updates := make([]struct { cacheUpdate scheduleTaskCacheUpdate nextRun *time.Time @@ -469,17 +469,6 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC } task.PublishAccountIDs = decodeDueSchedulePublishAccountIDs(publishAccountIDsJSON) task.PublishEnterpriseSiteTargets = tenantapp.DecodeScheduleEnterpriseSiteTargets(publishEnterpriseSiteTargetsJSON) - if task.AutoPublish && len(task.PublishAccountIDs) > 0 { - accountIDs, _, err := tenantapp.LoadExistingSchedulePublishAccountIDsAndPlatformsByIDs(ctx, tx, task.TenantID, task.WorkspaceID, task.PublishAccountIDs) - if err != nil { - rows.Close() - return nil, err - } - task.PublishAccountIDs = accountIDs - } - if task.AutoPublish && len(task.PublishAccountIDs) == 0 && len(task.PublishEnterpriseSiteTargets) == 0 { - task.AutoPublish = false - } task.CoverRandomFolderID = int64PtrFromInt8(coverRandomFolderID) if task.CoverMode == "" { task.CoverMode = tenantapp.ScheduleCoverModeSpecific @@ -519,7 +508,7 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC }) if nextErr == nil && sharedschedule.IsRunAllowed(task.ScheduledFor, task.StartAt, task.EndAt) { - tasks = append(tasks, task) + candidates = append(candidates, task) } } if err := rows.Err(); err != nil { @@ -528,6 +517,21 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeC } rows.Close() + tasks := make([]dueScheduleTask, 0, len(candidates)) + for _, task := range candidates { + if task.AutoPublish && len(task.PublishAccountIDs) > 0 { + accountIDs, _, err := tenantapp.LoadExistingSchedulePublishAccountIDsAndPlatformsByIDs(ctx, tx, task.TenantID, task.WorkspaceID, task.PublishAccountIDs) + if err != nil { + return nil, err + } + task.PublishAccountIDs = accountIDs + } + if task.AutoPublish && len(task.PublishAccountIDs) == 0 && len(task.PublishEnterpriseSiteTargets) == 0 { + task.AutoPublish = false + } + tasks = append(tasks, task) + } + for _, update := range updates { if update.invalidErr != nil { message := truncateDispatchError(update.invalidErr) diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 28f2bd9..5c87ba2 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -514,7 +514,20 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask( } hasAccountIDs := len(accountIDs) > 0 - row := s.pool.QueryRow(ctx, ` + row := s.pool.QueryRow(ctx, leaseNextQueuedMonitorTaskSQL(), + client.TenantID, + client.WorkspaceID, + client.ID, + hasAccountIDs, + accountIDs, + params.AttemptID, + params.LeaseTokenHash, + ) + return scanRepositoryDesktopTask(row) +} + +func leaseNextQueuedMonitorTaskSQL() string { + return ` WITH candidate AS ( SELECT dt.desktop_id FROM desktop_tasks AS dt @@ -522,6 +535,29 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask( AND dt.workspace_id = $2 AND dt.kind = 'monitor' AND dt.status = 'queued' + AND NOT EXISTS ( + SELECT 1 + FROM desktop_tasks AS active + WHERE active.tenant_id = dt.tenant_id + AND active.workspace_id = dt.workspace_id + AND active.kind = 'monitor' + AND active.target_client_id = $3 + AND active.platform_id = dt.platform_id + AND active.status = 'in_progress' + AND active.desktop_id <> dt.desktop_id + ) + AND NOT EXISTS ( + SELECT 1 + FROM desktop_tasks AS recent + WHERE recent.tenant_id = dt.tenant_id + AND recent.workspace_id = dt.workspace_id + AND recent.kind = 'monitor' + AND recent.target_client_id = $3 + AND recent.platform_id = dt.platform_id + AND recent.status IN ('succeeded', 'failed', 'unknown', 'aborted') + AND recent.updated_at >= now() - interval '30 seconds' + AND recent.desktop_id <> dt.desktop_id + ) AND ( dt.target_client_id = $3 OR ($4::boolean AND dt.target_account_id = ANY($5::uuid[])) @@ -544,16 +580,7 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask( updated_at = now() FROM candidate WHERE t.desktop_id = candidate.desktop_id - RETURNING `+desktopTaskRepositoryReturningColumns, - client.TenantID, - client.WorkspaceID, - client.ID, - hasAccountIDs, - accountIDs, - params.AttemptID, - params.LeaseTokenHash, - ) - return scanRepositoryDesktopTask(row) + RETURNING ` + desktopTaskRepositoryReturningColumns } func (s *DesktopTaskService) leaseNextQueuedPublishTask( @@ -625,7 +652,22 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID( } hasAccountIDs := len(accountIDs) > 0 - row := s.pool.QueryRow(ctx, ` + row := s.pool.QueryRow(ctx, leaseQueuedDesktopTaskByIDSQL(), + desktopID, + client.WorkspaceID, + client.TenantID, + client.ID, + hasAccountIDs, + accountIDs, + params.AttemptID, + params.LeaseTokenHash, + desktopPublishMaxAttempts, + ) + return scanRepositoryDesktopTask(row) +} + +func leaseQueuedDesktopTaskByIDSQL() string { + return ` WITH candidate AS ( SELECT dt.desktop_id FROM desktop_tasks AS dt @@ -645,6 +687,35 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID( AND pa.delete_requested_at IS NULL ) ) + AND ( + dt.kind <> 'monitor' + OR NOT EXISTS ( + SELECT 1 + FROM desktop_tasks AS active + WHERE active.tenant_id = dt.tenant_id + AND active.workspace_id = dt.workspace_id + AND active.kind = 'monitor' + AND active.target_client_id = $4 + AND active.platform_id = dt.platform_id + AND active.status = 'in_progress' + AND active.desktop_id <> dt.desktop_id + ) + ) + AND ( + dt.kind <> 'monitor' + OR NOT EXISTS ( + SELECT 1 + FROM desktop_tasks AS recent + WHERE recent.tenant_id = dt.tenant_id + AND recent.workspace_id = dt.workspace_id + AND recent.kind = 'monitor' + AND recent.target_client_id = $4 + AND recent.platform_id = dt.platform_id + AND recent.status IN ('succeeded', 'failed', 'unknown', 'aborted') + AND recent.updated_at >= now() - interval '30 seconds' + AND recent.desktop_id <> dt.desktop_id + ) + ) AND ( ( dt.kind = 'monitor' @@ -673,18 +744,7 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID( updated_at = now() FROM candidate WHERE t.desktop_id = candidate.desktop_id - RETURNING `+desktopTaskRepositoryReturningColumns, - desktopID, - client.WorkspaceID, - client.TenantID, - client.ID, - hasAccountIDs, - accountIDs, - params.AttemptID, - params.LeaseTokenHash, - desktopPublishMaxAttempts, - ) - return scanRepositoryDesktopTask(row) + RETURNING ` + desktopTaskRepositoryReturningColumns } func (s *DesktopTaskService) loadMonitorLeaseAccountIDs(ctx context.Context, client *repository.DesktopClient) ([]uuid.UUID, error) { diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index ef19f67..43ee6f5 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -203,6 +203,44 @@ func TestBuildCountPublishTasksByStatusesQueryHidesDeletedPublishRecords(t *test } } +func TestLeaseNextQueuedMonitorTaskSQLSerializesPerClientPlatform(t *testing.T) { + t.Parallel() + + normalized := normalizeSQLForDesktopTaskTest(leaseNextQueuedMonitorTaskSQL()) + for _, fragment := range []string{ + "NOT EXISTS ( SELECT 1 FROM desktop_tasks AS active", + "active.target_client_id = $3", + "active.platform_id = dt.platform_id", + "active.status = 'in_progress'", + "active.desktop_id <> dt.desktop_id", + "recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')", + "recent.updated_at >= now() - interval '30 seconds'", + } { + if !strings.Contains(normalized, fragment) { + t.Fatalf("monitor lease query missing %q: %s", fragment, normalized) + } + } +} + +func TestLeaseQueuedDesktopTaskByIDSQLSerializesMonitorPerClientPlatform(t *testing.T) { + t.Parallel() + + normalized := normalizeSQLForDesktopTaskTest(leaseQueuedDesktopTaskByIDSQL()) + for _, fragment := range []string{ + "dt.kind <> 'monitor' OR NOT EXISTS", + "active.target_client_id = $4", + "active.platform_id = dt.platform_id", + "active.status = 'in_progress'", + "active.desktop_id <> dt.desktop_id", + "recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')", + "recent.updated_at >= now() - interval '30 seconds'", + } { + if !strings.Contains(normalized, fragment) { + t.Fatalf("task-specific lease query missing %q: %s", fragment, normalized) + } + } +} + func recoverDesktopTaskSelectColumns(query string) []string { re := regexp.MustCompile(`(?is)select\s+(.*?)\s+from\s+desktop_tasks`) match := re.FindStringSubmatch(query) diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index 20a5ca3..4b9edc3 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -1371,15 +1371,18 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI skip_reason, trigger_source, COALESCE(target_client_id::text, '') AS target_client_id, - COALESCE(( + COALESCE(NULLIF(t.question_text_snapshot, ''), ( 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 + AND btrim(question_text_snapshot) <> '' + ORDER BY + CASE WHEN deleted_at IS NULL THEN 0 ELSE 1 END, + projected_at DESC, + id DESC LIMIT 1 ), '') AS question_text_snapshot FROM monitoring_collect_tasks t @@ -1439,15 +1442,18 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas skip_reason, trigger_source, COALESCE(target_client_id::text, '') AS target_client_id, - COALESCE(( + COALESCE(NULLIF(t.question_text_snapshot, ''), ( 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 + AND btrim(question_text_snapshot) <> '' + ORDER BY + CASE WHEN deleted_at IS NULL THEN 0 ELSE 1 END, + projected_at DESC, + id DESC LIMIT 1 ), '') AS question_text_snapshot FROM monitoring_collect_tasks t @@ -1538,15 +1544,18 @@ func (s *MonitoringCallbackService) selectPendingLeaseTasks( t.business_date, COALESCE(t.dispatch_priority, 100) AS dispatch_priority, COALESCE(t.dispatch_lane, 'normal') AS dispatch_lane, - COALESCE(( + COALESCE(NULLIF(t.question_text_snapshot, ''), ( 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 + AND btrim(question_text_snapshot) <> '' + ORDER BY + CASE WHEN deleted_at IS NULL THEN 0 ELSE 1 END, + projected_at DESC, + id DESC LIMIT 1 ), '') AS question_text_snapshot FROM monitoring_collect_tasks t @@ -1650,15 +1659,18 @@ func (s *MonitoringCallbackService) selectLeasedTasksForInstallation( t.business_date, COALESCE(t.dispatch_priority, 100) AS dispatch_priority, COALESCE(t.dispatch_lane, 'normal') AS dispatch_lane, - COALESCE(( + COALESCE(NULLIF(t.question_text_snapshot, ''), ( 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 + AND btrim(question_text_snapshot) <> '' + ORDER BY + CASE WHEN deleted_at IS NULL THEN 0 ELSE 1 END, + projected_at DESC, + id DESC LIMIT 1 ), '') AS question_text_snapshot FROM monitoring_collect_tasks t @@ -2070,6 +2082,10 @@ func (s *MonitoringCallbackService) upsertParseResult(ctx context.Context, tx pg } func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task *monitoringCollectTask, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) { + if task == nil || strings.TrimSpace(task.QuestionText) == "" { + return nil, response.ErrInternal(50041, "missing_question_text_snapshot", "monitoring task question text is empty") + } + req = normalizeMonitoringTaskResultRequestValue(req) runStatus := normalizeRunStatus(req.Status) diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 67ab1df..d511d4a 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -933,23 +933,29 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( questionIDs := make([]int64, 0, len(candidates)) questionHashes := make([][]byte, 0, len(candidates)) + questionTexts := make([]string, 0, len(candidates)) platformIDs := make([]string, 0, len(candidates)) dispatchAfterValues := make([]time.Time, 0, len(candidates)) now := time.Now().UTC() for _, candidate := range candidates { + questionText, textErr := monitoringQuestionTextSnapshot(candidate.Question) + if textErr != nil { + return 0, textErr + } dispatchAfter := candidate.DispatchAfter if dispatchAfter.IsZero() { dispatchAfter = now } questionIDs = append(questionIDs, candidate.Question.ID) questionHashes = append(questionHashes, candidate.Question.QuestionHash) + questionTexts = append(questionTexts, questionText) platformIDs = append(platformIDs, candidate.Platform.ID) dispatchAfterValues = append(dispatchAfterValues, dispatchAfter) } tag, err := tx.Exec(ctx, ` INSERT INTO monitoring_collect_tasks ( - tenant_id, brand_id, question_id, question_hash, ai_platform_id, + tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id, collector_type, trigger_source, run_mode, business_date, planned_at, status, dispatch_priority, dispatch_lane, target_client_id, dispatch_after, ingest_shard_key, execution_owner @@ -959,6 +965,7 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( $2, input.question_id, input.question_hash, + input.question_text, input.ai_platform_id, $3, 'automatic', @@ -976,12 +983,17 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( $7::bigint[], $8::bytea[], $9::text[], - $10::timestamptz[] - ) AS input(question_id, question_hash, ai_platform_id, dispatch_after) + $10::text[], + $11::timestamptz[] + ) AS input(question_id, question_hash, question_text, ai_platform_id, dispatch_after) ON CONFLICT ( tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date ) - DO NOTHING + DO UPDATE SET + question_hash = EXCLUDED.question_hash, + question_text_snapshot = EXCLUDED.question_text_snapshot, + updated_at = NOW() + WHERE btrim(COALESCE(monitoring_collect_tasks.question_text_snapshot, '')) = '' `, plan.TenantID, brand.BrandID, @@ -991,6 +1003,7 @@ func (s *MonitoringService) ensureDailyMonitoringCollectTasks( executionOwner, questionIDs, questionHashes, + questionTexts, platformIDs, dispatchAfterValues, ) @@ -1060,15 +1073,18 @@ func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs( t.dispatch_priority, t.dispatch_lane, t.interrupt_generation, - COALESCE(( + COALESCE(NULLIF(t.question_text_snapshot, ''), ( SELECT question_text_snapshot 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.deleted_at IS NULL - ORDER BY s.projected_at DESC, s.id DESC + AND btrim(s.question_text_snapshot) <> '' + ORDER BY + CASE WHEN s.deleted_at IS NULL THEN 0 ELSE 1 END, + s.projected_at DESC, + s.id DESC LIMIT 1 ), '') AS question_text_snapshot FROM claimed_tasks t @@ -1105,6 +1121,10 @@ func (s *MonitoringService) loadDueDailyMonitorDesktopTaskSpecs( spec.WorkspaceID = workspaceID spec.BusinessDate = businessDay.Format("2006-01-02") spec.QuestionHash = encodeQuestionHash(questionHash) + spec.QuestionText = strings.TrimSpace(spec.QuestionText) + if spec.QuestionText == "" { + continue + } spec.SchedulerGroupKey = monitoringSchedulerGroupKey(spec.QuestionHash, spec.QuestionID, spec.QuestionText) result = append(result, spec) } diff --git a/server/internal/tenant/app/monitoring_daily_task_worker_test.go b/server/internal/tenant/app/monitoring_daily_task_worker_test.go index f016535..0384e19 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -129,6 +129,7 @@ func TestBuildMonitoringDailyTaskCandidatesIncludesQuestionsBeyondLibraryLimit(t for id := int64(1); id <= 30; id++ { questions = append(questions, monitoringConfiguredQuestion{ ID: id, + QuestionText: "搜索词 " + int64Text(id), QuestionHash: []byte("q" + int64Text(id)), }) } @@ -152,6 +153,80 @@ func TestBuildMonitoringDailyTaskCandidatesIncludesQuestionsBeyondLibraryLimit(t assert.Contains(t, seen, int64(30)) } +func TestEnsureDailyMonitoringCollectTasksPersistsQuestionTextSnapshot(t *testing.T) { + ctx := context.Background() + tx := &captureDailyMonitorTaskTx{commandTag: pgconn.NewCommandTag("INSERT 0 2")} + businessDay := time.Date(2026, 6, 22, 0, 0, 0, 0, time.UTC) + + created, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks( + ctx, + tx, + monitoringDailyPlan{TenantID: 4}, + monitoringDailyBrand{BrandID: 8}, + businessDay, + []monitoringDailyTaskCandidate{ + { + Question: monitoringConfiguredQuestion{ + ID: 145, + QuestionText: " 幽灵门轨道长度定制 ", + QuestionHash: []byte("q145"), + }, + Platform: monitoringPlatformMetadata{ID: "qwen"}, + }, + { + Question: monitoringConfiguredQuestion{ + ID: 145, + QuestionText: "幽灵门轨道长度定制", + QuestionHash: []byte("q145"), + }, + Platform: monitoringPlatformMetadata{ID: "doubao"}, + }, + }, + "desktop_tasks", + ) + + require.NoError(t, err) + assert.Equal(t, int64(2), created) + assert.True(t, tx.execCalled) + normalizedSQL := normalizeSQLWhitespace(tx.execSQL) + assert.Contains(t, normalizedSQL, "question_text_snapshot, ai_platform_id") + assert.Contains(t, normalizedSQL, "$10::text[]") + assert.Contains(t, normalizedSQL, "AS input(question_id, question_hash, question_text, ai_platform_id, dispatch_after)") + assert.Contains(t, normalizedSQL, "question_text_snapshot = EXCLUDED.question_text_snapshot") + require.Len(t, tx.execArgs, 11) + assert.Equal(t, []string{"幽灵门轨道长度定制", "幽灵门轨道长度定制"}, tx.execArgs[8]) + assert.Equal(t, []string{"qwen", "doubao"}, tx.execArgs[9]) +} + +func TestEnsureDailyMonitoringCollectTasksRejectsEmptyQuestionText(t *testing.T) { + ctx := context.Background() + tx := &captureDailyMonitorTaskTx{} + + created, err := (&MonitoringService{}).ensureDailyMonitoringCollectTasks( + ctx, + tx, + monitoringDailyPlan{TenantID: 4}, + monitoringDailyBrand{BrandID: 8}, + time.Date(2026, 6, 22, 0, 0, 0, 0, time.UTC), + []monitoringDailyTaskCandidate{ + { + Question: monitoringConfiguredQuestion{ + ID: 145, + QuestionText: " ", + QuestionHash: []byte("q145"), + }, + Platform: monitoringPlatformMetadata{ID: "qwen"}, + }, + }, + "desktop_tasks", + ) + + require.Error(t, err) + assert.Equal(t, int64(0), created) + assert.False(t, tx.execCalled) + assert.Contains(t, err.Error(), "missing_question_text_snapshot") +} + func TestDecodeDailyMonitoringEnabledPlatformsReportsMalformedJSON(t *testing.T) { platforms, err := decodeDailyMonitoringEnabledPlatforms([]byte(`["qwen"`)) require.Error(t, err) @@ -360,6 +435,7 @@ func TestLoadDueDailyMonitorDesktopTaskSpecsFiltersToDispatchablePlatforms(t *te require.NoError(t, err) assert.Empty(t, specs) assert.True(t, queryer.called) + assert.Contains(t, normalizeSQLWhitespace(queryer.sql), "COALESCE(NULLIF(t.question_text_snapshot, ''),") assert.Contains(t, normalizeSQLWhitespace(queryer.sql), "AND t.ai_platform_id = ANY($7::text[])") assert.Contains(t, normalizeSQLWhitespace(queryer.sql), "LIMIT $8") require.Len(t, queryer.args, 8) @@ -423,3 +499,50 @@ func (r emptyDailyMonitorRows) Scan(...any) error { r func (r emptyDailyMonitorRows) Values() ([]any, error) { return nil, nil } func (r emptyDailyMonitorRows) RawValues() [][]byte { return nil } func (r emptyDailyMonitorRows) Conn() *pgx.Conn { return nil } + +type captureDailyMonitorTaskTx struct { + execCalled bool + execSQL string + execArgs []any + commandTag pgconn.CommandTag +} + +func (tx *captureDailyMonitorTaskTx) Begin(context.Context) (pgx.Tx, error) { return tx, nil } +func (tx *captureDailyMonitorTaskTx) Commit(context.Context) error { return nil } +func (tx *captureDailyMonitorTaskTx) Rollback(context.Context) error { return nil } +func (tx *captureDailyMonitorTaskTx) CopyFrom(context.Context, pgx.Identifier, []string, pgx.CopyFromSource) (int64, error) { + return 0, nil +} +func (tx *captureDailyMonitorTaskTx) SendBatch(context.Context, *pgx.Batch) pgx.BatchResults { + return nil +} +func (tx *captureDailyMonitorTaskTx) LargeObjects() pgx.LargeObjects { return pgx.LargeObjects{} } +func (tx *captureDailyMonitorTaskTx) Prepare(context.Context, string, string) (*pgconn.StatementDescription, error) { + return nil, nil +} +func (tx *captureDailyMonitorTaskTx) Exec(_ context.Context, sql string, arguments ...any) (pgconn.CommandTag, error) { + tx.execCalled = true + tx.execSQL = sql + tx.execArgs = arguments + return tx.commandTag, nil +} +func (tx *captureDailyMonitorTaskTx) Query(context.Context, string, ...any) (pgx.Rows, error) { + return emptyDailyMonitorRows{}, nil +} +func (tx *captureDailyMonitorTaskTx) QueryRow(context.Context, string, ...any) pgx.Row { + return boolDailyMonitorRow(true) +} +func (tx *captureDailyMonitorTaskTx) Conn() *pgx.Conn { return nil } + +var _ pgx.Tx = (*captureDailyMonitorTaskTx)(nil) + +type boolDailyMonitorRow bool + +func (r boolDailyMonitorRow) Scan(dest ...any) error { + if len(dest) == 1 { + if value, ok := dest[0].(*bool); ok { + *value = bool(r) + } + } + return nil +} diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index d9cf26d..8231731 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -138,15 +138,18 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs( t.dispatch_priority, t.dispatch_lane, t.interrupt_generation, - COALESCE(( + COALESCE(NULLIF(t.question_text_snapshot, ''), ( SELECT question_text_snapshot 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.deleted_at IS NULL - ORDER BY s.projected_at DESC, s.id DESC + AND btrim(s.question_text_snapshot) <> '' + ORDER BY + CASE WHEN s.deleted_at IS NULL THEN 0 ELSE 1 END, + s.projected_at DESC, + s.id DESC LIMIT 1 ), '') AS question_text_snapshot FROM monitoring_collect_tasks t @@ -193,6 +196,10 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs( spec.RequestedByUserID = requestedByUserID spec.BusinessDate = businessDay.Format("2006-01-02") spec.QuestionHash = encodeQuestionHash(questionHash) + spec.QuestionText = strings.TrimSpace(spec.QuestionText) + if spec.QuestionText == "" { + continue + } spec.SchedulerGroupKey = monitoringSchedulerGroupKey(spec.QuestionHash, spec.QuestionID, spec.QuestionText) result = append(result, spec) } @@ -210,9 +217,13 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( return nil, nil, nil } - targets, err := s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs) - if err != nil { - return nil, nil, err + var targets map[string]monitorDesktopTaskTarget + if monitorDesktopTaskSpecsNeedTargetLookup(specs) { + var err error + targets, err = s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs) + if err != nil { + return nil, nil, err + } } tx, err := s.businessPool.Begin(ctx) @@ -225,13 +236,23 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( publishableTasks := make([]*repository.DesktopTask, 0, len(specs)) fallbackLegacyTaskIDs := make([]int64, 0) for _, spec := range specs { - target, ok := targets[normalizeMonitoringPlatformID(spec.PlatformID)] - if !ok || target.AccountID == uuid.Nil || target.ClientID == uuid.Nil { - fallbackLegacyTaskIDs = append(fallbackLegacyTaskIDs, spec.MonitorTaskID) + if spec.TargetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil { + target, ok := targets[normalizeMonitoringPlatformID(spec.PlatformID)] + if !ok || target.AccountID == uuid.Nil || target.ClientID == uuid.Nil { + fallbackLegacyTaskIDs = append(fallbackLegacyTaskIDs, spec.MonitorTaskID) + continue + } + spec.TargetAccountID = target.AccountID + spec.TargetClientID = target.ClientID + } + + available, throttleErr := acquireMonitorDesktopTaskPlatformSlot(ctx, tx, spec) + if throttleErr != nil { + return nil, nil, throttleErr + } + if !available { continue } - spec.TargetAccountID = target.AccountID - spec.TargetClientID = target.ClientID task, shouldPublish, fallbackLegacy, upsertErr := s.upsertMonitorDesktopTask(ctx, tx, repo, spec) if upsertErr != nil { @@ -252,6 +273,54 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( return publishableTasks, fallbackLegacyTaskIDs, nil } +func monitorDesktopTaskSpecsNeedTargetLookup(specs []monitorDesktopTaskSpec) bool { + for _, spec := range specs { + if spec.TargetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil { + return true + } + } + return false +} + +func acquireMonitorDesktopTaskPlatformSlot(ctx context.Context, tx pgx.Tx, spec monitorDesktopTaskSpec) (bool, error) { + if tx == nil || spec.TargetClientID == uuid.Nil || strings.TrimSpace(spec.PlatformID) == "" { + return false, nil + } + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskPlatformSlotLockKey(spec.TargetClientID, spec.PlatformID)); err != nil { + return false, response.ErrInternal(50123, "desktop_task_throttle_lock_failed", "failed to acquire phase2 monitor desktop platform slot") + } + + var busy bool + if err := tx.QueryRow(ctx, monitorDesktopTaskPlatformSlotBusySQL(), spec.TenantID, spec.WorkspaceID, spec.TargetClientID, normalizeMonitoringPlatformID(spec.PlatformID), spec.MonitorTaskID).Scan(&busy); err != nil { + return false, response.ErrInternal(50124, "desktop_task_throttle_lookup_failed", "failed to inspect phase2 monitor desktop platform slot") + } + return !busy, nil +} + +func monitorDesktopTaskPlatformSlotBusySQL() string { + return ` + SELECT EXISTS ( + SELECT 1 + FROM desktop_tasks + WHERE tenant_id = $1 + AND workspace_id = $2 + AND kind = 'monitor' + AND target_client_id = $3 + AND platform_id = $4 + AND status = 'in_progress' + AND ( + $5::bigint <= 0 + OR monitor_task_id IS NULL + OR monitor_task_id <> $5 + ) + ) + ` +} + +func monitorDesktopTaskPlatformSlotLockKey(clientID uuid.UUID, platformID string) int64 { + return int64(monitoringDailyStableHash("monitor_desktop_platform_slot", clientID.String(), normalizeMonitoringPlatformID(platformID))) +} + func (s *MonitoringService) loadMonitorDesktopTaskTargets( ctx context.Context, tenantID, workspaceID, userID int64, @@ -721,6 +790,7 @@ func (s *MonitoringService) requestPhase2MonitorInterrupts( ctx context.Context, targetClientID uuid.UUID, excludedMonitorTaskIDs []int64, + excludedPlatformIDs []string, interruptGeneration int, ) ([]monitorDesktopInterruptTarget, error) { if s == nil || s.businessPool == nil { @@ -728,6 +798,7 @@ func (s *MonitoringService) requestPhase2MonitorInterrupts( } queryArgs := []any{targetClientID, interruptGeneration} + nextParam := 3 query := ` UPDATE desktop_tasks SET control_flags = ( @@ -752,8 +823,14 @@ func (s *MonitoringService) requestPhase2MonitorInterrupts( AND monitor_task_id IS NOT NULL ` if len(excludedMonitorTaskIDs) > 0 { - query += ` AND NOT (monitor_task_id = ANY($3::bigint[]))` + query += fmt.Sprintf(` AND NOT (monitor_task_id = ANY($%d::bigint[]))`, nextParam) queryArgs = append(queryArgs, excludedMonitorTaskIDs) + nextParam++ + } + excludedPlatformIDs = reconcileEnabledMonitoringPlatforms(excludedPlatformIDs) + if len(excludedPlatformIDs) > 0 { + query += fmt.Sprintf(` AND NOT (platform_id = ANY($%d::text[]))`, nextParam) + queryArgs = append(queryArgs, excludedPlatformIDs) } query += ` RETURNING desktop_id::text, monitor_task_id, target_client_id::text, workspace_id, interrupt_generation @@ -981,18 +1058,18 @@ func monitoringSchedulerGroupKey(questionHash string, questionID int64, question } func marshalMonitorDesktopTaskPayload(spec monitorDesktopTaskSpec) ([]byte, error) { - title := strings.TrimSpace(spec.QuestionText) - platformName := platformDisplayName(spec.PlatformID) - if title != "" { - title = fmt.Sprintf("%s · %s", platformName, title) - } else { - title = fmt.Sprintf("监控任务 · %s", platformName) + questionText := strings.TrimSpace(spec.QuestionText) + if questionText == "" { + return nil, response.ErrInternal(50041, "missing_question_text_snapshot", "monitor desktop task question text is empty") } + title := questionText + platformName := platformDisplayName(spec.PlatformID) + title = fmt.Sprintf("%s · %s", platformName, title) payload, err := json.Marshal(map[string]any{ "title": title, "business_date": spec.BusinessDate, "scheduler_group_key": spec.SchedulerGroupKey, - "question_text": spec.QuestionText, + "question_text": questionText, "question_id": spec.QuestionID, "question_hash": spec.QuestionHash, "ai_platform_id": spec.PlatformID, diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go index 8fc350e..34ab2f2 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -1,10 +1,13 @@ package app import ( + "encoding/json" + "strings" "testing" "github.com/google/uuid" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestSelectMonitorDesktopTaskTargetsPrefersPresentAccountClient(t *testing.T) { @@ -147,3 +150,60 @@ func TestSelectMonitorDesktopTaskTargetsPrefersEarlierOnlineClientForSameAccount ClientID: earlierClientID, }, targets["doubao"]) } + +func TestMarshalMonitorDesktopTaskPayloadRequiresQuestionText(t *testing.T) { + payload, err := marshalMonitorDesktopTaskPayload(monitorDesktopTaskSpec{ + PlatformID: "qwen", + QuestionText: " ", + }) + + assert.Nil(t, payload) + require.Error(t, err) + assert.Contains(t, err.Error(), "missing_question_text_snapshot") +} + +func TestMarshalMonitorDesktopTaskPayloadUsesQuestionTextNotFallbackTitle(t *testing.T) { + payload, err := marshalMonitorDesktopTaskPayload(monitorDesktopTaskSpec{ + PlatformID: "qwen", + BusinessDate: "2026-06-22", + SchedulerGroupKey: "qid:145", + QuestionID: 145, + QuestionText: " 幽灵门轨道长度定制 ", + QuestionHash: "v1:test", + }) + require.NoError(t, err) + + var decoded map[string]any + require.NoError(t, json.Unmarshal(payload, &decoded)) + assert.Equal(t, "幽灵门轨道长度定制", decoded["question_text"]) + assert.Equal(t, "通义千问 · 幽灵门轨道长度定制", decoded["title"]) +} + +func TestMonitorDesktopTaskPlatformSlotBusySQLThrottlesPerClientPlatformInProgress(t *testing.T) { + query := strings.Join(strings.Fields(monitorDesktopTaskPlatformSlotBusySQL()), " ") + + for _, fragment := range []string{ + "FROM desktop_tasks", + "kind = 'monitor'", + "target_client_id = $3", + "platform_id = $4", + "status = 'in_progress'", + "monitor_task_id <> $5", + } { + assert.Contains(t, query, fragment) + } +} + +func TestRequestPhase2MonitorInterruptsAcceptsExcludedPlatforms(t *testing.T) { + clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000701") + targets, err := (&MonitoringService{}).requestPhase2MonitorInterrupts( + t.Context(), + clientID, + []int64{101, 102}, + []string{"doubao", "qwen"}, + 7, + ) + + require.NoError(t, err) + assert.Empty(t, targets) +} diff --git a/server/internal/tenant/app/monitoring_service.go b/server/internal/tenant/app/monitoring_service.go index 2ecf4ae..2dd7f1d 100644 --- a/server/internal/tenant/app/monitoring_service.go +++ b/server/internal/tenant/app/monitoring_service.go @@ -361,6 +361,14 @@ type monitoringConfiguredQuestion struct { QuestionHash []byte } +func monitoringQuestionTextSnapshot(question monitoringConfiguredQuestion) (string, error) { + questionText := strings.TrimSpace(question.QuestionText) + if questionText == "" { + return "", response.ErrInternal(50041, "missing_question_text_snapshot", "configured monitoring question text is empty") + } + return questionText, nil +} + var defaultMonitoringPlatforms = []monitoringPlatformMetadata{ {ID: "yuanbao", Name: "混元 / 元宝"}, {ID: "kimi", Name: "Kimi"}, @@ -1018,8 +1026,10 @@ func (s *MonitoringService) CollectNow( phase2DeferredTasks := make([]*repository.DesktopTask, 0) if options.Preempt && phase2DispatchReady { excludedMonitorTaskIDs := make([]int64, 0, len(phase2TaskSpecs)) + excludedPlatformIDs := make([]string, 0, len(phase2TaskSpecs)) for _, spec := range phase2TaskSpecs { excludedMonitorTaskIDs = append(excludedMonitorTaskIDs, spec.MonitorTaskID) + excludedPlatformIDs = append(excludedPlatformIDs, spec.PlatformID) } phase2TargetClientIDs := desktopTaskClientIDs(phase2PublishedTasks) if len(phase2TargetClientIDs) == 0 { @@ -1031,7 +1041,7 @@ func (s *MonitoringService) CollectNow( return nil, deferErr } phase2DeferredTasks = append(phase2DeferredTasks, deferredTasks...) - interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, interruptGeneration) + interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, excludedPlatformIDs, interruptGeneration) if interruptErr != nil { return nil, interruptErr } @@ -1534,6 +1544,10 @@ func (s *MonitoringService) loadConfiguredQuestions(ctx context.Context, tenantI if scanErr := rows.Scan(&item.ID, &item.KeywordID, &item.QuestionText); scanErr != nil { return nil, response.ErrInternal(50041, "scan_failed", "failed to parse configured monitoring questions") } + item.QuestionText = strings.TrimSpace(item.QuestionText) + if item.QuestionText == "" { + return nil, response.ErrInternal(50041, "missing_question_text_snapshot", "configured monitoring question text is empty") + } item.QuestionHash = seededQuestionHash(item.QuestionText) items = append(items, item) } @@ -1631,6 +1645,10 @@ func (s *MonitoringService) syncMonitoringQuestionSnapshots(ctx context.Context, } for _, question := range questions { + questionText, textErr := monitoringQuestionTextSnapshot(question) + if textErr != nil { + return textErr + } if _, err := tx.Exec(ctx, ` UPDATE monitoring_question_config_snapshots SET superseded_at = COALESCE(superseded_at, NOW()), @@ -1661,7 +1679,7 @@ func (s *MonitoringService) syncMonitoringQuestionSnapshots(ctx context.Context, deleted_at = NULL, projected_at = NOW(), updated_at = NOW() - `, tenantID, brandID, question.ID, question.KeywordID, question.QuestionHash, question.QuestionText); err != nil { + `, tenantID, brandID, question.ID, question.KeywordID, question.QuestionHash, questionText); err != nil { return response.ErrInternal(50041, "snapshot_upsert_failed", "failed to persist monitoring question snapshot") } } @@ -1692,6 +1710,10 @@ func (s *MonitoringService) ensureCollectNowTasks( interruptTaskIDs := make([]int64, 0) for _, question := range questions { + questionText, textErr := monitoringQuestionTextSnapshot(question) + if textErr != nil { + return 0, 0, 0, 0, nil, textErr + } for _, platform := range platforms { platformID := normalizeMonitoringPlatformID(platform.ID) if platformID == "" { @@ -1704,6 +1726,7 @@ func (s *MonitoringService) ensureCollectNowTasks( tag, err := tx.Exec(ctx, ` UPDATE monitoring_collect_tasks SET question_hash = $7, + question_text_snapshot = $12, status = 'pending', planned_at = NOW(), trigger_source = 'manual', @@ -1737,7 +1760,7 @@ func (s *MonitoringService) ensureCollectNowTasks( AND run_mode = $6 AND business_date = $8::date AND status IN ('pending', 'expired', 'failed', 'completed', 'skipped') - `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, executionOwner) + `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, executionOwner, questionText) if err != nil { return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks") } @@ -1746,6 +1769,7 @@ func (s *MonitoringService) ensureCollectNowTasks( rows, err := tx.Query(ctx, ` UPDATE monitoring_collect_tasks SET question_hash = $7, + question_text_snapshot = $11, trigger_source = 'manual', dispatch_priority = 5000, dispatch_lane = 'high', @@ -1770,7 +1794,7 @@ func (s *MonitoringService) ensureCollectNowTasks( AND status = 'leased' AND COALESCE(execution_owner, 'legacy') = 'legacy' RETURNING id - `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration) + `, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, questionText) if err != nil { return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks") } @@ -1790,13 +1814,13 @@ func (s *MonitoringService) ensureCollectNowTasks( tag, err = tx.Exec(ctx, ` INSERT INTO monitoring_collect_tasks ( - tenant_id, brand_id, question_id, question_hash, ai_platform_id, + tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id, collector_type, trigger_source, run_mode, business_date, planned_at, status, dispatch_priority, dispatch_lane, target_client_id, dispatch_after, interrupt_generation, ingest_shard_key, execution_owner ) VALUES ( - $1, $2, $3, $4, $5, + $1, $2, $3, $4, $13, $5, $6, 'manual', $7, $8::date, NOW(), 'pending', 5000, 'high', $9, NOW(), $10, $11, $12 @@ -1805,7 +1829,7 @@ func (s *MonitoringService) ensureCollectNowTasks( tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date ) DO NOTHING - `, tenantID, brandID, question.ID, question.QuestionHash, platformID, monitoringCollectorType, runMode, dateText, platformTargetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner) + `, tenantID, brandID, question.ID, question.QuestionHash, platformID, monitoringCollectorType, runMode, dateText, platformTargetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner, questionText) if err != nil { return 0, 0, 0, 0, nil, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks") } diff --git a/server/migrations_monitoring/20260622093000_add_collect_task_question_text_snapshot.down.sql b/server/migrations_monitoring/20260622093000_add_collect_task_question_text_snapshot.down.sql new file mode 100644 index 0000000..99637cb --- /dev/null +++ b/server/migrations_monitoring/20260622093000_add_collect_task_question_text_snapshot.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE monitoring_collect_tasks + DROP COLUMN IF EXISTS question_text_snapshot; diff --git a/server/migrations_monitoring/20260622093000_add_collect_task_question_text_snapshot.up.sql b/server/migrations_monitoring/20260622093000_add_collect_task_question_text_snapshot.up.sql new file mode 100644 index 0000000..9debc40 --- /dev/null +++ b/server/migrations_monitoring/20260622093000_add_collect_task_question_text_snapshot.up.sql @@ -0,0 +1,28 @@ +ALTER TABLE monitoring_collect_tasks + ADD COLUMN IF NOT EXISTS question_text_snapshot TEXT NOT NULL DEFAULT ''; + +UPDATE monitoring_collect_tasks t +SET question_text_snapshot = s.question_text_snapshot +FROM ( + SELECT DISTINCT ON (tenant_id, brand_id, question_id, question_hash) + tenant_id, + brand_id, + question_id, + question_hash, + question_text_snapshot + FROM monitoring_question_config_snapshots + WHERE btrim(question_text_snapshot) <> '' + ORDER BY + tenant_id, + brand_id, + question_id, + question_hash, + CASE WHEN deleted_at IS NULL THEN 0 ELSE 1 END, + projected_at DESC, + id DESC +) s +WHERE t.tenant_id = s.tenant_id + AND t.brand_id = s.brand_id + AND t.question_id = s.question_id + AND t.question_hash = s.question_hash + AND btrim(t.question_text_snapshot) = '';