From 8f7a83bba9a425dac1e48a05409e3441304d47eb Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 22 Jun 2026 16:37:41 +0800 Subject: [PATCH] fix: queue monitor tasks and recover doubao dom answers --- .../src/main/adapters/doubao.test.ts | 34 +++++++- .../src/main/adapters/doubao.ts | 46 +++++++++-- .../tenant/app/desktop_task_service.go | 79 +++++++++++++++++-- .../tenant/app/desktop_task_service_test.go | 6 ++ .../app/monitoring_phase2_desktop_tasks.go | 47 +---------- .../monitoring_phase2_desktop_tasks_test.go | 15 ---- 6 files changed, 154 insertions(+), 73 deletions(-) diff --git a/apps/desktop-client/src/main/adapters/doubao.test.ts b/apps/desktop-client/src/main/adapters/doubao.test.ts index 2e9c087..6472572 100644 --- a/apps/desktop-client/src/main/adapters/doubao.test.ts +++ b/apps/desktop-client/src/main/adapters/doubao.test.ts @@ -251,10 +251,42 @@ describe('doubao adapter helpers', () => { }) }) - it('does not submit a success answer when the SSE stream has no answer text', () => { + it('uses DOM answer when Doubao finishes the stream without SSE answer text', () => { expect( resolveDoubaoSubmissionAnswer({ answer: null, + domAnswer: + '口袋门墙体预埋件通常需要确认墙体厚度、轨道承重、缓冲阻尼和检修口位置。选购时建议优先核对门扇重量、轨道长度和五金兼容性。', + allowDomAnswer: true, + }), + ).toEqual({ + answer: + '口袋门墙体预埋件通常需要确认墙体厚度、轨道承重、缓冲阻尼和检修口位置。选购时建议优先核对门扇重量、轨道长度和五金兼容性。', + source: 'dom', + }) + }) + + it('does not submit DOM answer before a finish signal allows DOM recovery', () => { + expect( + resolveDoubaoSubmissionAnswer({ + answer: null, + domAnswer: + '口袋门墙体预埋件通常需要确认墙体厚度、轨道承重、缓冲阻尼和检修口位置。选购时建议优先核对门扇重量、轨道长度和五金兼容性。', + allowDomAnswer: false, + }), + ).toEqual({ + answer: null, + source: null, + }) + }) + + it('does not submit a weak history-title DOM fragment as a success answer', () => { + expect( + resolveDoubaoSubmissionAnswer({ + answer: null, + domAnswer: + '口袋门墙体预埋件全解析 卧室口袋门阻尼缓冲导轨选购安装指南 口袋门防夹手缓冲方案 隐形门合页选购指南', + allowDomAnswer: true, }), ).toEqual({ answer: null, diff --git a/apps/desktop-client/src/main/adapters/doubao.ts b/apps/desktop-client/src/main/adapters/doubao.ts index 611031a..3a91859 100644 --- a/apps/desktop-client/src/main/adapters/doubao.ts +++ b/apps/desktop-client/src/main/adapters/doubao.ts @@ -469,14 +469,45 @@ function selectBestDoubaoText(primary: string | null, secondary: string | null): return secondaryScore > primaryScore ? secondaryText : primaryText } -function resolveDoubaoSubmissionAnswer(input: { answer: string | null }): { +function isDoubaoSSESubmissionCandidate(value: string | null): boolean { + const answer = normalizeOptionalString(value) + if (!answer || containsDoubaoMojibakeSignal(answer)) { + return false + } + return true +} + +function isDoubaoDOMSubmissionCandidate(value: string | null): boolean { + const answer = normalizeOptionalString(value) + if (!answer || containsDoubaoMojibakeSignal(answer)) { + return false + } + if (answer.length < 48) { + return false + } + const punctuationCount = countDoubaoMatches(answer, /[。！？；：，、,.!?;:]/g) + if (punctuationCount < 2 && !doubaoHasStructuredAnswerMarkers(answer) && !/\n/.test(answer)) { + return false + } + return doubaoTextQualityScore(answer) >= 180 +} + +function resolveDoubaoSubmissionAnswer(input: { answer: string | null - source: 'sse' | null + domAnswer?: string | null + allowDomAnswer?: boolean +}): { + answer: string | null + source: 'sse' | 'dom' | null } { const answer = normalizeOptionalString(input.answer) - if (answer && !containsDoubaoMojibakeSignal(answer)) { + if (isDoubaoSSESubmissionCandidate(answer)) { return { answer, source: 'sse' } } + const domAnswer = normalizeOptionalString(input.domAnswer) + if (input.allowDomAnswer === true && isDoubaoDOMSubmissionCandidate(domAnswer)) { + return { answer: domAnswer, source: 'dom' } + } return { answer: null, source: null } } @@ -2505,7 +2536,8 @@ const doubaoQueryInPage = async ( return } seen.add(href) - const visibleText = normalize(element.innerText) ?? normalize(element.textContent) ?? fallbackText + const visibleText = + normalize(element.innerText) ?? normalize(element.textContent) ?? fallbackText links.push({ url: href, title: @@ -2643,9 +2675,7 @@ const doubaoQueryInPage = async ( /搜索\s*\d+\s*个关键词|参考\s*\d+\s*篇资料|查看\d+篇资料|参考资料|资料来源/.test(line) const isReferenceQueryLine = (line: string): boolean => - /[“"「].+[”"」]/.test(line) || - /、/.test(line) || - /关键词|搜索词|查询词/.test(line) + /[“"「].+[”"」]/.test(line) || /、/.test(line) || /关键词|搜索词|查询词/.test(line) const isReferenceItemLine = (line: string): boolean => /^\d+[.、]\s*\S+/.test(line) && @@ -3240,6 +3270,8 @@ export const doubaoAdapter: MonitorAdapter = { pageResult.ok === false && isDoubaoRecoverablePageError(pageResult.error) const submission = resolveDoubaoSubmissionAnswer({ answer, + domAnswer: pageResult.domAnswer, + allowDomAnswer: streamSawAnswerFinish || pageResult.ok === true, }) const hasRecoveredAnswer = Boolean(submission.answer) const hasRecoveredContent = hasRecoveredAnswer || searchResults.length > 0 diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index d121ce3..b079a57 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -25,7 +25,10 @@ import ( "github.com/geo-platform/tenant-api/internal/tenant/repository" ) -const desktopPublishMaxAttempts = 3 +const ( + desktopPublishMaxAttempts = 3 + maxConcurrentMonitorPlatformsPerDesktopClient = 2 +) var errDesktopTaskLeaseDeferred = errors.New("desktop task lease deferred") @@ -520,7 +523,13 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask( } hasAccountIDs := len(accountIDs) > 0 - row := s.pool.QueryRow(ctx, leaseNextQueuedMonitorTaskSQL(), + tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + row := tx.QueryRow(ctx, leaseNextQueuedMonitorTaskSQL(), client.TenantID, client.WorkspaceID, client.ID, @@ -528,8 +537,16 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask( accountIDs, params.AttemptID, params.LeaseTokenHash, + maxConcurrentMonitorPlatformsPerDesktopClient, ) - return scanRepositoryDesktopTask(row) + task, err := scanRepositoryDesktopTask(row) + if err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to commit desktop monitor task lease") + } + return task, nil } func leaseNextQueuedMonitorTaskSQL() string { @@ -564,6 +581,16 @@ func leaseNextQueuedMonitorTaskSQL() string { AND recent.updated_at >= now() - interval '30 seconds' AND recent.desktop_id <> dt.desktop_id ) + AND ( + SELECT COUNT(DISTINCT active_platform.platform_id) + FROM desktop_tasks AS active_platform + WHERE active_platform.tenant_id = dt.tenant_id + AND active_platform.workspace_id = dt.workspace_id + AND active_platform.kind = 'monitor' + AND active_platform.target_client_id = $3 + AND active_platform.status = 'in_progress' + AND active_platform.desktop_id <> dt.desktop_id + ) < $8::integer AND ( dt.target_client_id = $3 OR ($4::boolean AND dt.target_account_id = ANY($5::uuid[])) @@ -658,7 +685,13 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID( } hasAccountIDs := len(accountIDs) > 0 - row := s.pool.QueryRow(ctx, leaseQueuedDesktopTaskByIDSQL(), + tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + row := tx.QueryRow(ctx, leaseQueuedDesktopTaskByIDSQL(), desktopID, client.WorkspaceID, client.TenantID, @@ -668,8 +701,16 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID( params.AttemptID, params.LeaseTokenHash, desktopPublishMaxAttempts, + maxConcurrentMonitorPlatformsPerDesktopClient, ) - return scanRepositoryDesktopTask(row) + task, err := scanRepositoryDesktopTask(row) + if err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to commit desktop task lease") + } + return task, nil } func leaseQueuedDesktopTaskByIDSQL() string { @@ -722,6 +763,19 @@ func leaseQueuedDesktopTaskByIDSQL() string { AND recent.desktop_id <> dt.desktop_id ) ) + AND ( + dt.kind <> 'monitor' + OR ( + SELECT COUNT(DISTINCT active_platform.platform_id) + FROM desktop_tasks AS active_platform + WHERE active_platform.tenant_id = dt.tenant_id + AND active_platform.workspace_id = dt.workspace_id + AND active_platform.kind = 'monitor' + AND active_platform.target_client_id = $4 + AND active_platform.status = 'in_progress' + AND active_platform.desktop_id <> dt.desktop_id + ) < $10::integer + ) AND ( ( dt.kind = 'monitor' @@ -753,6 +807,21 @@ func leaseQueuedDesktopTaskByIDSQL() string { RETURNING ` + desktopTaskRepositoryReturningColumns } +func beginDesktopMonitorLeaseTx(ctx context.Context, pool *pgxpool.Pool, clientID uuid.UUID) (pgx.Tx, error) { + if pool == nil { + return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "desktop task pool is not available") + } + tx, err := pool.Begin(ctx) + if err != nil { + return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to start desktop monitor task lease") + } + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskClientSlotLockKey(clientID)); err != nil { + _ = tx.Rollback(ctx) + return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to lock desktop monitor task lease") + } + return tx, nil +} + func (s *DesktopTaskService) loadMonitorLeaseAccountIDs(ctx context.Context, client *repository.DesktopClient) ([]uuid.UUID, error) { if s == nil || client == nil || s.pool == nil { return nil, nil diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index 20f08bd..174a0c9 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -249,6 +249,9 @@ func TestLeaseNextQueuedMonitorTaskSQLSerializesPerClientPlatform(t *testing.T) "active.platform_id = dt.platform_id", "active.status = 'in_progress'", "active.desktop_id <> dt.desktop_id", + "COUNT(DISTINCT active_platform.platform_id)", + "active_platform.target_client_id = $3", + "< $8::integer", "recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')", "recent.updated_at >= now() - interval '30 seconds'", } { @@ -268,6 +271,9 @@ func TestLeaseQueuedDesktopTaskByIDSQLSerializesMonitorPerClientPlatform(t *test "active.platform_id = dt.platform_id", "active.status = 'in_progress'", "active.desktop_id <> dt.desktop_id", + "COUNT(DISTINCT active_platform.platform_id)", + "active_platform.target_client_id = $4", + "< $10::integer", "recent.status IN ('succeeded', 'failed', 'unknown', 'aborted')", "recent.updated_at >= now() - interval '30 seconds'", } { diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index c257568..9f9b9b7 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -246,14 +246,6 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( spec.TargetClientID = target.ClientID } - available, throttleErr := acquireMonitorDesktopTaskPlatformSlot(ctx, tx, spec) - if throttleErr != nil { - return nil, nil, throttleErr - } - if !available { - continue - } - task, shouldPublish, fallbackLegacy, upsertErr := s.upsertMonitorDesktopTask(ctx, tx, repo, spec) if upsertErr != nil { return nil, nil, upsertErr @@ -282,43 +274,8 @@ func monitorDesktopTaskSpecsNeedTargetLookup(specs []monitorDesktopTaskSpec) boo 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 monitorDesktopTaskClientSlotLockKey(clientID uuid.UUID) int64 { + return int64(monitoringDailyStableHash("monitor_desktop_client_slot", clientID.String())) } func (s *MonitoringService) loadMonitorDesktopTaskTargets( 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 707aa4b..dd6ffc5 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -207,21 +207,6 @@ func TestMonitoringSkipOutcomeKeepsOrdinarySkip(t *testing.T) { assert.Equal(t, "stale_business_date", outcome.StoredSkipReason) } -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(