From 04bd3e42e09ce7b41739cb5a701c683b9654dcba Mon Sep 17 00:00:00 2001 From: liangxu Date: Tue, 23 Jun 2026 23:12:25 +0800 Subject: [PATCH] fix: harden desktop monitoring recovery --- .../src/main/adapters/doubao.test.ts | 64 ++++ .../src/main/adapters/doubao.ts | 241 ++++++++++-- apps/desktop-client/src/main/bootstrap.ts | 77 +++- .../src/main/monitor-scheduler.ts | 21 ++ .../desktop-client/src/main/playwright-cdp.ts | 234 ++++++++++-- .../src/main/publish-scheduler.ts | 18 + .../src/main/runtime-controller.ts | 342 +++++++++++++++++- .../desktop-client/src/main/runtime-events.ts | 10 +- apps/desktop-client/src/preload/bridge.ts | 16 +- apps/desktop-client/src/renderer/env.d.ts | 8 +- .../tenant/app/desktop_task_service.go | 125 +++++-- .../tenant/app/desktop_task_service_test.go | 22 +- .../tenant/app/monitoring_callback_service.go | 228 ++++++++++-- .../app/monitoring_phase2_desktop_tasks.go | 43 ++- .../monitoring_phase2_desktop_tasks_test.go | 29 +- .../internal/tenant/app/monitoring_service.go | 15 +- 16 files changed, 1329 insertions(+), 164 deletions(-) diff --git a/apps/desktop-client/src/main/adapters/doubao.test.ts b/apps/desktop-client/src/main/adapters/doubao.test.ts index 6472572..a47e849 100644 --- a/apps/desktop-client/src/main/adapters/doubao.test.ts +++ b/apps/desktop-client/src/main/adapters/doubao.test.ts @@ -12,6 +12,7 @@ const { isNonCitationAssetUrl, parseDoubaoStreamBody, resolveDoubaoSubmissionAnswer, + shouldRetryDoubaoPageAttempt, } = __doubaoTestUtils describe('doubao adapter helpers', () => { @@ -28,6 +29,69 @@ describe('doubao adapter helpers', () => { expect(isDoubaoRecoverablePageError('doubao_login_expired')).toBe(false) }) + it('retries once when Doubao is stuck on the history surface without an answer', () => { + const streamSummary = parseDoubaoStreamBody( + ['event: SSE_REPLY_END', 'data: {"message_id":"msg_1"}', '', ''].join('\n'), + ) + + expect( + shouldRetryDoubaoPageAttempt( + { + ok: false, + error: 'doubao_query_timeout', + detail: '历史对话 主对话 折叠门联动配件明细', + url: 'https://www.doubao.com/chat/local_4995117315817271', + title: '豆包', + modelLabel: '豆包', + modeLabel: '思考', + thinkingModeApplied: true, + deepThinkEnabled: true, + domAnswer: '折叠门联动配件明细', + domReasoning: null, + domLinks: [], + submitted: true, + questionAnchorFound: true, + historySurfaceDetected: true, + }, + streamSummary, + ), + ).toBe(true) + }) + + it('does not retry a recoverable Doubao page error after SSE already yielded an answer', () => { + const streamSummary = parseDoubaoStreamBody( + [ + 'event: STREAM_CHUNK', + 'data: {"text_block":{"text":"这是一段完整的豆包回答。"}}', + '', + '', + ].join('\n'), + ) + + expect( + shouldRetryDoubaoPageAttempt( + { + ok: false, + error: 'doubao_query_timeout', + detail: 'timeout', + url: 'https://www.doubao.com/chat/local_4995117315817271', + title: '豆包', + modelLabel: '豆包', + modeLabel: '思考', + thinkingModeApplied: true, + deepThinkEnabled: true, + domAnswer: null, + domReasoning: null, + domLinks: [], + submitted: true, + questionAnchorFound: true, + historySurfaceDetected: true, + }, + streamSummary, + ), + ).toBe(false) + }) + it('maps the new expert/deep-thinking mode label to expert mode', () => { expect(doubaoModeKindFromLabelText('专家')).toBe('expert') expect(doubaoModeKindFromLabelText('专家 深度思考 研究级智能模型')).toBe('expert') diff --git a/apps/desktop-client/src/main/adapters/doubao.ts b/apps/desktop-client/src/main/adapters/doubao.ts index 3a91859..6967ca3 100644 --- a/apps/desktop-client/src/main/adapters/doubao.ts +++ b/apps/desktop-client/src/main/adapters/doubao.ts @@ -9,7 +9,7 @@ const DOUBAO_PAGE_READY_TIMEOUT_MS = 20_000 const DOUBAO_QUERY_TIMEOUT_MS = 120_000 const DOUBAO_CAPTURE_LIMIT = 64 const DOUBAO_CAPTURE_BODY_LIMIT = 8_000_000 -const DOUBAO_CAPTURE_FLUSH_TIMEOUT_MS = DOUBAO_QUERY_TIMEOUT_MS +const DOUBAO_CAPTURE_FLUSH_TIMEOUT_MS = 10_000 const DOUBAO_STREAM_URL_FRAGMENT = '/chat/completion' const DOUBAO_NON_CITATION_ASSET_DOMAINS = new Set(['byteimg.com', 'bytednsdoc.com']) @@ -116,10 +116,22 @@ type DoubaoPageQueryFailureResult = { domAnswer: string | null domReasoning: string | null domLinks: DoubaoDomLink[] + submitted: boolean | null + questionAnchorFound: boolean | null + historySurfaceDetected: boolean | null } type DoubaoPageQueryResult = DoubaoPageQuerySuccessResult | DoubaoPageQueryFailureResult +type DoubaoQueryAttemptResult = { + attempt: number + pageResult: DoubaoPageQueryResult + captures: DoubaoCaptureRecord[] + streamSummary: DoubaoStreamSummary + captureDebug: JsonValue[] + pendingCaptureCount: number +} + function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value) } @@ -1039,7 +1051,12 @@ function isDoubaoExecutionContextNavigationError(error: unknown): boolean { } function isDoubaoRecoverablePageError(error: string): boolean { - return error === 'doubao_query_timeout' || error === 'doubao_page_navigation_interrupted' + return ( + error === 'doubao_query_timeout' || + error === 'doubao_page_navigation_interrupted' || + error === 'doubao_submission_not_observed' || + error === 'doubao_history_surface_timeout' + ) } function buildDoubaoPageNavigationResult( @@ -1059,9 +1076,30 @@ function buildDoubaoPageNavigationResult( domAnswer: null, domReasoning: null, domLinks: [], + submitted: null, + questionAnchorFound: null, + historySurfaceDetected: null, } } +function shouldRetryDoubaoPageAttempt( + pageResult: DoubaoPageQueryResult, + streamSummary: DoubaoStreamSummary, +): boolean { + if (pageResult.ok || !isDoubaoRecoverablePageError(pageResult.error)) { + return false + } + if (isDoubaoSSESubmissionCandidate(streamSummary.answer)) { + return false + } + return ( + pageResult.error === 'doubao_submission_not_observed' || + pageResult.error === 'doubao_history_surface_timeout' || + pageResult.historySurfaceDetected === true || + pageResult.questionAnchorFound === false + ) +} + function buildDoubaoUnknownResult( code: string, summary: string, @@ -1597,6 +1635,8 @@ const doubaoQueryInPage = async ( const pollIntervalMs = 1_500 const incompleteAnswerGraceMs = 30_000 const minimumAnswerSettleMs = 10_000 + const submissionObserveTimeoutMs = 12_000 + const staleHistoryTimeoutMs = 18_000 const wait = (timeMs: number) => new Promise((resolve) => { @@ -2684,6 +2724,39 @@ const doubaoQueryInPage = async ( line, )) + const looksLikeHistoryTitleList = (text: string | null): boolean => { + if (!text) { + return false + } + const normalizedText = text.replace(/\s+/g, ' ').trim() + if ( + /搜索…\s*Ctrl\s*K\s*豆包|新对话\s*Ctrl\s*Shift\s*K|历史对话\s*主对话/.test( + normalizedText, + ) + ) { + return true + } + const lines = text + .split(/\n+/) + .map((line) => line.trim()) + .filter(Boolean) + if (lines.length < 6) { + return false + } + const shortTitleLines = lines.filter( + (line) => + line.length <= 36 && + !/[。！？!?]/.test(line) && + /门|锁|轨|五金|导轨|合页|配件|工厂|厂家|供货|定制|折叠|幽灵|口袋/.test(line), + ).length + const navHits = lines.filter((line) => + /^(搜索|豆包|新对话|新办公任务|AI 创作|AI创作|云盘|收起|发现智能体|历史对话|主对话)$/.test( + line, + ), + ).length + return shortTitleLines >= 6 && (navHits >= 2 || shortTitleLines / lines.length >= 0.65) + } + // Tokens that only appear on Doubao's empty-chat surface (prompt templates, nav, feature chips). const EMPTY_CHAT_TOKENS = [ '快速', @@ -2721,6 +2794,9 @@ const doubaoQueryInPage = async ( return hits >= 3 || (hits >= 2 && lines.length <= 6) } + const looksLikeNonAnswerSurface = (text: string | null): boolean => + looksLikeEmptyChatSurface(text) || looksLikeHistoryTitleList(text) + const stripKnownNoise = (text: string | null): string | null => { if (!text) { return null @@ -2870,7 +2946,7 @@ const doubaoQueryInPage = async ( if (!text && links.length === 0) { continue } - if (rawText && looksLikeEmptyChatSurface(rawText)) { + if (rawText && looksLikeNonAnswerSurface(rawText)) { continue } if (text && text.length < 12 && links.length === 0) { @@ -3033,6 +3109,11 @@ const doubaoQueryInPage = async ( applied: null, }, snapshotOverride: ConversationSnapshot | null = null, + diagnostics: { + submitted?: boolean | null + questionAnchorFound?: boolean | null + historySurfaceDetected?: boolean | null + } = {}, ): DoubaoPageQueryFailureResult => { const snapshot = snapshotOverride ?? snapshotConversation(composerTop) const composer = findComposer() @@ -3057,6 +3138,9 @@ const doubaoQueryInPage = async ( domAnswer: snapshot.answerText, domReasoning: snapshot.reasoningText, domLinks: snapshot.links, + submitted: diagnostics.submitted ?? null, + questionAnchorFound: diagnostics.questionAnchorFound ?? null, + historySurfaceDetected: diagnostics.historySurfaceDetected ?? null, } } @@ -3113,6 +3197,8 @@ const doubaoQueryInPage = async ( } const initialSnapshot = snapshotConversation(composerTop) + let questionAnchorFoundAfterSubmit = Boolean(findQuestionAnchor()) + let historySurfaceDetected = looksLikeHistoryTitleList(bodyText()) let bestSnapshot = initialSnapshot let bestContentSnapshot: ConversationSnapshot | null = hasObservedContent(initialSnapshot) ? initialSnapshot @@ -3121,10 +3207,13 @@ const doubaoQueryInPage = async ( let stableCount = submitQuestion ? 0 : hasObservedContent(initialSnapshot) ? 1 : 0 let sawChange = !submitQuestion && hasObservedContent(initialSnapshot) let firstChangedContentAt: number | null = sawChange ? Date.now() : null + const submittedAt = Date.now() const deadline = Date.now() + timeoutMs while (Date.now() <= deadline) { const snapshot = snapshotConversation(composerTop) + questionAnchorFoundAfterSubmit = questionAnchorFoundAfterSubmit || Boolean(findQuestionAnchor()) + historySurfaceDetected = historySurfaceDetected || looksLikeHistoryTitleList(bodyText()) bestSnapshot = pickBetterSnapshot(bestSnapshot, snapshot) if (hasObservedContent(snapshot)) { bestContentSnapshot = pickBetterSnapshot(bestContentSnapshot, snapshot) @@ -3152,6 +3241,53 @@ const doubaoQueryInPage = async ( composerTop, answerMode, bestContentSnapshot ?? bestSnapshot, + { + submitted: submitQuestion, + questionAnchorFound: questionAnchorFoundAfterSubmit, + historySurfaceDetected, + }, + ) + } + + const elapsedSinceSubmit = Date.now() - submittedAt + if ( + submitQuestion && + !questionAnchorFoundAfterSubmit && + !busy && + elapsedSinceSubmit >= submissionObserveTimeoutMs + ) { + return fail( + 'doubao_submission_not_observed', + bodyText(), + composerTop, + answerMode, + bestContentSnapshot ?? bestSnapshot, + { + submitted: true, + questionAnchorFound: false, + historySurfaceDetected, + }, + ) + } + if ( + submitQuestion && + questionAnchorFoundAfterSubmit && + historySurfaceDetected && + !hasContent && + !busy && + elapsedSinceSubmit >= staleHistoryTimeoutMs + ) { + return fail( + 'doubao_history_surface_timeout', + bodyText(), + composerTop, + answerMode, + bestContentSnapshot ?? bestSnapshot, + { + submitted: true, + questionAnchorFound: true, + historySurfaceDetected: true, + }, ) } @@ -3196,6 +3332,11 @@ const doubaoQueryInPage = async ( composerTop, answerMode, bestContentSnapshot ?? bestSnapshot, + { + submitted: submitQuestion, + questionAnchorFound: questionAnchorFoundAfterSubmit, + historySurfaceDetected, + }, ) } @@ -3214,40 +3355,66 @@ export const doubaoAdapter: MonitorAdapter = { const questionText = extractQuestionText(payload) const page = context.playwright.page + const runAttempt = async (attempt: number): Promise => { + context.reportProgress(attempt === 1 ? 'doubao.query' : 'doubao.query_retry') + const capture = attachDoubaoResponseCapture(page) + let pageResult: DoubaoPageQueryResult + try { + try { + pageResult = await page.evaluate(doubaoQueryInPage, { + questionText, + timeoutMs: DOUBAO_QUERY_TIMEOUT_MS, + readyTimeoutMs: DOUBAO_PAGE_READY_TIMEOUT_MS, + enableDeepThink: true, + }) + } catch (error) { + if (!isDoubaoExecutionContextNavigationError(error)) { + throw error + } + await page + .waitForLoadState('domcontentloaded', { timeout: DOUBAO_PAGE_READY_TIMEOUT_MS }) + .catch(() => undefined) + await page.waitForLoadState('networkidle', { timeout: 3_000 }).catch(() => undefined) + pageResult = buildDoubaoPageNavigationResult(page, formatUnknownError(error)) + } + await sleep(600, context.signal).catch(() => undefined) + await capture.flush(DOUBAO_CAPTURE_FLUSH_TIMEOUT_MS) + } finally { + capture.dispose() + } + const streamSummary = parseDoubaoCaptures(capture.captures) + return { + attempt, + pageResult, + captures: [...capture.captures], + streamSummary, + captureDebug: summarizeCapturesForDebug(capture.captures), + pendingCaptureCount: capture.pendingCount(), + } + } + context.reportProgress('doubao.page_ready') await ensurePageOnDoubaoChat(page, context.signal) - const capture = attachDoubaoResponseCapture(page) - let pageResult: DoubaoPageQueryResult - - try { - context.reportProgress('doubao.query') - try { - pageResult = await page.evaluate(doubaoQueryInPage, { - questionText, - timeoutMs: DOUBAO_QUERY_TIMEOUT_MS, - readyTimeoutMs: DOUBAO_PAGE_READY_TIMEOUT_MS, - enableDeepThink: true, - }) - } catch (error) { - if (!isDoubaoExecutionContextNavigationError(error)) { - throw error - } - await page - .waitForLoadState('domcontentloaded', { timeout: DOUBAO_PAGE_READY_TIMEOUT_MS }) - .catch(() => undefined) - await page.waitForLoadState('networkidle', { timeout: 3_000 }).catch(() => undefined) - pageResult = buildDoubaoPageNavigationResult(page, formatUnknownError(error)) - } - await sleep(600, context.signal).catch(() => undefined) - await capture.flush(DOUBAO_CAPTURE_FLUSH_TIMEOUT_MS) - } finally { - capture.dispose() + let attemptResult = await runAttempt(1) + if ( + attemptResult.pageResult.ok === false && + shouldRetryDoubaoPageAttempt(attemptResult.pageResult, attemptResult.streamSummary) + ) { + context.reportProgress('doubao.recover_new_chat') + await resetDoubaoPageAfterRecoverableFailure( + page, + attemptResult.pageResult.error, + context.signal, + ) + await ensurePageOnDoubaoChat(page, context.signal) + attemptResult = await runAttempt(2) } - const streamSummary = parseDoubaoCaptures(capture.captures) - const captureDebug = summarizeCapturesForDebug(capture.captures) - const pendingCaptureCount = capture.pendingCount() + const pageResult = attemptResult.pageResult + const streamSummary = attemptResult.streamSummary + const captureDebug = attemptResult.captureDebug + const pendingCaptureCount = attemptResult.pendingCaptureCount const domSearchResults = dedupeSourceItems( pageResult.domLinks .map((item) => @@ -3316,6 +3483,11 @@ export const doubaoAdapter: MonitorAdapter = { reply_end_event_count: streamSummary.reply_end_event_count, answer_finish_event_count: streamSummary.answer_finish_event_count, stream_event_count: streamSummary.event_count, + submitted: failed.submitted, + question_anchor_found: failed.questionAnchorFound, + history_surface_detected: failed.historySurfaceDetected, + query_attempt: attemptResult.attempt, + captures: captureDebug, }), } } @@ -3426,11 +3598,12 @@ export const doubaoAdapter: MonitorAdapter = { platform: 'doubao', page_url: pageResult.url, page_title: pageResult.title, + query_attempt: attemptResult.attempt, model_label: pageResult.modelLabel, mode_label: pageResult.modeLabel, thinking_mode_applied: pageResult.thinkingModeApplied, deep_think_enabled: pageResult.deepThinkEnabled, - capture_count: capture.captures.length, + capture_count: attemptResult.captures.length, capture_pending_after_flush: pendingCaptureCount, event_count: streamSummary.event_count, full_message_event_count: streamSummary.full_message_event_count, @@ -3464,6 +3637,8 @@ export const __doubaoTestUtils = { isNonCitationAssetDomain, isNonCitationAssetUrl, isDoubaoRecoverablePageError, + isDoubaoDOMSubmissionCandidate, + shouldRetryDoubaoPageAttempt, parseDoubaoStreamBody, resolveDoubaoSubmissionAnswer, } diff --git a/apps/desktop-client/src/main/bootstrap.ts b/apps/desktop-client/src/main/bootstrap.ts index bf2b78e..470406f 100644 --- a/apps/desktop-client/src/main/bootstrap.ts +++ b/apps/desktop-client/src/main/bootstrap.ts @@ -53,7 +53,7 @@ import { installObservedGlobalFetch, registerObservedRequestRendererTarget, } from './network-observer' -import { getPlaywrightCDPPort, startHiddenPlaywrightReaper } from './playwright-cdp' +import { getPlaywrightCDPPort, preparePlaywrightCDPPort, startHiddenPlaywrightReaper } from './playwright-cdp' import { initProcessMetricsSampler } from './process-metrics' import { registerRendererDevtoolsProxyTarget } from './renderer-devtools-proxy' import { @@ -112,8 +112,8 @@ app.commandLine.appendSwitch( ) app.commandLine.appendSwitch('disable-quic') app.commandLine.appendSwitch('disable-background-networking') -// CDP is required by the hidden Playwright execution pool in packaged builds. -app.commandLine.appendSwitch('remote-debugging-port', String(getPlaywrightCDPPort())) +// CDP is required by the hidden Playwright execution pool. Electron only reads +// this switch during early startup, so select the port before app.whenReady(). app.userAgentFallback = STANDARD_USER_AGENT // Windows 10+ routes tray.displayBalloon through the Toast Notification API, @@ -151,6 +151,7 @@ console.info('[desktop-main] boot', { chrome: process.versions.chrome, node: process.versions.node, ua: STANDARD_USER_AGENT, + cdpPort: getPlaywrightCDPPort(), }) process.on('unhandledRejection', (reason) => { @@ -166,6 +167,7 @@ let loginWindow: ElectronBrowserWindow | null = null let settingsWindow: ElectronBrowserWindow | null = null let mainRendererContents: ElectronWebContents | null = null let quitReleaseInFlight = false +let cdpRecoveryRelaunchInFlight = false let quitAfterSaasLogoutPromise: Promise | null = null let mainWindowCreatePromise: Promise | null = null let loginWindowCreatePromise: Promise | null = null @@ -176,6 +178,10 @@ let canHandleDesktopDeepLinks = false const pendingDesktopDeepLinks: string[] = [] const forceCloseWindows = new WeakSet() const appWindowModes = new WeakMap() +const maxCDPRecoveryRelaunchesPerWindow = 3 +const cdpRecoveryRelaunchWindowMs = 10 * 60_000 +const cdpRecoveryRelaunchDelayMs = 1_500 +const cdpFatalArg = '--geo-cdp-fatal' type WindowMode = 'login' | 'main' type RendererWindowMode = WindowMode | 'settings' @@ -547,6 +553,50 @@ function quitAfterClearingSaasSession(): void { }) } +function requestPlaywrightCDPRecoveryRelaunch(message: string): void { + if (cdpRecoveryRelaunchInFlight) { + return + } + + const now = Date.now() + const recentRelaunches = process.argv + .filter((arg) => arg.startsWith('--geo-cdp-recovery-at=')) + .map((arg) => Number.parseInt(arg.split('=')[1] ?? '', 10)) + .filter((timestamp) => Number.isFinite(timestamp) && now - timestamp < cdpRecoveryRelaunchWindowMs) + + if (recentRelaunches.length >= maxCDPRecoveryRelaunchesPerWindow) { + console.error('[desktop-main] playwright cdp recovery relaunch suppressed', { + message, + recentRelaunches: recentRelaunches.length, + }) + showTrayBalloon( + '浏览器自动化通道异常', + '客户端已停止采集。请重启客户端或检查安全软件/端口占用。', + ) + return + } + + cdpRecoveryRelaunchInFlight = true + console.warn('[desktop-main] scheduling playwright cdp recovery relaunch', { + message, + cdpPort: getPlaywrightCDPPort(), + }) + showTrayBalloon('正在恢复浏览器自动化通道', '客户端将自动重启后台进程，登录状态会保留。') + + setTimeout(() => { + quitReleaseInFlight = true + app.relaunch({ + args: [ + ...process.argv.slice(1).filter((arg) => !arg.startsWith('--geo-cdp-recovery-at=')), + ...recentRelaunches.map((timestamp) => `--geo-cdp-recovery-at=${timestamp}`), + `--geo-cdp-recovery-at=${now}`, + ...(recentRelaunches.length + 1 >= maxCDPRecoveryRelaunchesPerWindow ? [cdpFatalArg] : []), + ], + }) + app.exit(0) + }, cdpRecoveryRelaunchDelayMs) +} + function windowFromIpcEvent(event: IpcMainInvokeEvent): ElectronBrowserWindow | null { const window = BrowserWindow.fromWebContents(event.sender) if (!window || window.isDestroyed()) { @@ -1197,9 +1247,15 @@ if (!hasSingleInstanceLock) { console.info('[desktop-main] another instance is already running; exiting') app.exit(0) } else { - app - .whenReady() + preparePlaywrightCDPPort() + .then((selectedPlaywrightCDPPort) => { + app.commandLine.appendSwitch('remote-debugging-port', String(selectedPlaywrightCDPPort)) + return app.whenReady() + }) .then(async () => { + console.info('[desktop-main] playwright cdp port selected', { + cdpPort: getPlaywrightCDPPort(), + }) suppressNativeWindowMenu() registerDesktopProtocolClient() currentWindowMode = readPersistedWindowMode() @@ -1215,6 +1271,15 @@ if (!hasSingleInstanceLock) { startHiddenPlaywrightReaper() onRuntimeInvalidated((event) => { syncDesktopHealthIndicator(currentMainWindow()) + if (event.reason === 'playwright-cdp-fatal') { + showTrayBalloon( + '浏览器自动化组件不可用', + event.message ?? '客户端已停止采集，请重启客户端或检查安全软件/端口占用。', + ) + } + if (event.reason === 'playwright-cdp-recovery') { + requestPlaywrightCDPRecoveryRelaunch(event.message ?? 'playwright_cdp_endpoint_unavailable') + } if (!mainRendererContents || mainRendererContents.isDestroyed()) { return } @@ -1234,7 +1299,7 @@ if (!hasSingleInstanceLock) { flushPendingDesktopDeepLinks() }) .catch((error) => { - console.error('[desktop-main] app.whenReady failed', error) + console.error('[desktop-main] startup failed', error) }) app.on('activate', () => { diff --git a/apps/desktop-client/src/main/monitor-scheduler.ts b/apps/desktop-client/src/main/monitor-scheduler.ts index 408e96a..8007a75 100644 --- a/apps/desktop-client/src/main/monitor-scheduler.ts +++ b/apps/desktop-client/src/main/monitor-scheduler.ts @@ -64,6 +64,7 @@ interface MonitorSchedulerSelectionOptions { maxConcurrency: number activePlatforms: ReadonlySet activeQuestionKeys: ReadonlySet + blockedPlatforms?: ReadonlySet activeCount: number } @@ -434,6 +435,22 @@ export function noteMonitorTaskLeaseMiss(taskId: string): void { }) } +export function noteMonitorTaskDeferred(taskId: string, delayMs: number): void { + const now = Date.now() + const availableAt = now + Math.max(0, delayMs) + + mutatePersistedState((draft) => { + const existing = draft.tasks[taskId] + if (!existing) { + return + } + + existing.state = 'queued' + existing.availableAt = Math.max(existing.availableAt, availableAt) + existing.lastSeenAt = now + }) +} + export function noteMonitorTaskCompleted(taskId: string): void { const now = Date.now() @@ -488,6 +505,10 @@ export function selectNextMonitorTask( }) for (const candidate of candidates) { + if (options.blockedPlatforms?.has(candidate.platform)) { + continue + } + if (options.activePlatforms.has(candidate.platform)) { continue } diff --git a/apps/desktop-client/src/main/playwright-cdp.ts b/apps/desktop-client/src/main/playwright-cdp.ts index 77c65a3..66c104d 100644 --- a/apps/desktop-client/src/main/playwright-cdp.ts +++ b/apps/desktop-client/src/main/playwright-cdp.ts @@ -1,3 +1,5 @@ +import { createServer } from 'node:net' + import type { BrowserWindow as ElectronBrowserWindow, Session } from 'electron/main' import { BrowserWindow, nativeTheme } from 'electron/main' import type { Browser as PlaywrightBrowser, Page as PlaywrightPage } from 'playwright-core' @@ -13,9 +15,23 @@ const hiddenPlaywrightReaperIntervalMs = 60_000 const defaultConnectTimeoutMs = 10_000 const defaultPageTimeoutMs = 45_000 const hiddenBootstrapMetaName = 'geo-hidden-playwright-token' -const cdpPort = Number.parseInt(process.env.GEO_ELECTRON_CDP_PORT ?? '9339', 10) -const cdpEndpointURL = `http://127.0.0.1:${cdpPort}` -const cdpTargetsListURL = `${cdpEndpointURL}/json/list` +const preferredCDPPort = normalizeCDPPort(process.env.GEO_ELECTRON_CDP_PORT, 9339) +const dynamicCDPPortMin = normalizeCDPPort(process.env.GEO_ELECTRON_CDP_PORT_MIN, 9339) +const dynamicCDPPortMax = normalizeCDPPort(process.env.GEO_ELECTRON_CDP_PORT_MAX, 20000) +const cdpCircuitOpenTTLms = 60_000 +const cdpCircuitProbeTimeoutMs = 2_000 + +let cdpPort = preferredCDPPort +let cdpEndpointURL = endpointURLForPort(cdpPort) +let cdpTargetsListURL = `${cdpEndpointURL}/json/list` + +interface PlaywrightCDPHealthState { + status: 'unknown' | 'healthy' | 'unavailable' + checkedAt: number | null + unavailableSince: number | null + nextRetryAt: number | null + lastError: string | null +} interface HiddenPlaywrightRecord { key: string @@ -41,11 +57,95 @@ const records = new Map() let reaperHandle: ReturnType | null = null let connectPromise: Promise | null = null let connectedBrowser: PlaywrightBrowser | null = null +let cdpHealthState: PlaywrightCDPHealthState = { + status: 'unknown', + checkedAt: null, + unavailableSince: null, + nextRetryAt: null, + lastError: null, +} export function getPlaywrightCDPPort(): number { return cdpPort } +export async function preparePlaywrightCDPPort(): Promise { + const port = await findAvailableCDPPort() + setPlaywrightCDPPort(port) + return port +} + +export function isPlaywrightCDPUnavailableError(error: unknown): boolean { + if (!(error instanceof Error)) { + return false + } + return ( + error.message === 'playwright_cdp_endpoint_unavailable' || + error.message === 'playwright_cdp_port_unavailable' || + error.message.includes('playwright_cdp_endpoint_unavailable') || + error.message.includes('connectOverCDP') + ) +} + +export async function ensurePlaywrightCDPReady(options: { force?: boolean } = {}): Promise { + const now = Date.now() + if ( + !options.force && + cdpHealthState.status === 'healthy' && + cdpHealthState.checkedAt && + now - cdpHealthState.checkedAt < 15_000 + ) { + return + } + if ( + !options.force && + cdpHealthState.status === 'unavailable' && + cdpHealthState.nextRetryAt && + cdpHealthState.nextRetryAt > now + ) { + throw new Error(cdpHealthState.lastError || 'playwright_cdp_endpoint_unavailable') + } + + try { + await waitForCDPEndpoint(cdpCircuitProbeTimeoutMs) + markPlaywrightCDPHealthy() + } catch (error) { + markPlaywrightCDPUnavailable(error) + throw error + } +} + +export function isPlaywrightCDPExecutionAvailable(now = Date.now()): boolean { + if (connectedBrowser?.isConnected()) { + return true + } + if (cdpHealthState.status !== 'unavailable') { + return true + } + return Boolean(cdpHealthState.nextRetryAt && cdpHealthState.nextRetryAt <= now) +} + +export function getHiddenPlaywrightSnapshot() { + return { + cdpPort, + cdpEndpointURL, + connected: Boolean(connectedBrowser?.isConnected()), + pageCount: records.size, + idleTTLms: hiddenPlaywrightIdleTTLms, + health: { ...cdpHealthState }, + records: [...records.values()].map((record) => ({ + accountId: record.accountId, + title: record.title, + targetURL: record.targetURL, + retainCount: record.retainCount, + activatedAt: record.activatedAt, + lastReleasedAt: record.lastReleasedAt, + windowDestroyed: record.window.isDestroyed(), + pageClosed: record.page.isClosed(), + })), + } +} + export function startHiddenPlaywrightReaper(): void { if (reaperHandle) { return @@ -155,26 +255,6 @@ export function reapIdleHiddenPlaywrightPages( return reclaimed } -export function getHiddenPlaywrightSnapshot() { - return { - cdpPort, - cdpEndpointURL, - connected: Boolean(connectedBrowser?.isConnected()), - pageCount: records.size, - idleTTLms: hiddenPlaywrightIdleTTLms, - records: [...records.values()].map((record) => ({ - accountId: record.accountId, - title: record.title, - targetURL: record.targetURL, - retainCount: record.retainCount, - activatedAt: record.activatedAt, - lastReleasedAt: record.lastReleasedAt, - windowDestroyed: record.window.isDestroyed(), - pageClosed: record.page.isClosed(), - })), - } -} - function createLease(record: HiddenPlaywrightRecord): HiddenPlaywrightLease { return { accountId: record.accountId, @@ -271,6 +351,7 @@ async function connectBrowserOverCDP(): Promise { connectPromise = (async () => { await waitForCDPEndpoint() + markPlaywrightCDPHealthy() const { chromium } = await import('playwright-core') await pruneStaleCDPTargets() @@ -292,6 +373,9 @@ async function connectBrowserOverCDP(): Promise { try { return await connectPromise + } catch (error) { + markPlaywrightCDPUnavailable(error) + throw error } finally { if (!connectedBrowser?.isConnected()) { connectPromise = null @@ -314,7 +398,7 @@ async function waitForCDPEndpoint(timeoutMs = defaultConnectTimeoutMs): Promise< await wait(200) } - throw new Error('playwright_cdp_endpoint_unavailable') + throw new Error(`playwright_cdp_endpoint_unavailable:${cdpEndpointURL}`) } async function openBrowserOverCDP( @@ -330,6 +414,7 @@ async function openBrowserOverCDP( }) connectedBrowser = browser + markPlaywrightCDPHealthy() return browser } @@ -468,3 +553,104 @@ function wait(timeMs: number): Promise { setTimeout(resolve, timeMs) }) } + +function normalizeCDPPort(value: string | undefined, fallback: number): number { + const parsed = Number.parseInt(value ?? '', 10) + if (!Number.isFinite(parsed) || parsed <= 0 || parsed > 65_535) { + return fallback + } + return parsed +} + +function endpointURLForPort(port: number): string { + return `http://127.0.0.1:${port}` +} + +function setPlaywrightCDPPort(port: number): void { + cdpPort = port + cdpEndpointURL = endpointURLForPort(port) + cdpTargetsListURL = `${cdpEndpointURL}/json/list` +} + +async function findAvailableCDPPort(): Promise { + if (await canBindPort(preferredCDPPort)) { + return preferredCDPPort + } + + const min = Math.min(dynamicCDPPortMin, dynamicCDPPortMax) + const max = Math.max(dynamicCDPPortMin, dynamicCDPPortMax) + const start = Math.max(min, preferredCDPPort) + for (let port = start; port <= max; port += 1) { + if (port === preferredCDPPort) { + continue + } + if (await canBindPort(port)) { + return port + } + } + + for (let port = min; port < start; port += 1) { + if (port === preferredCDPPort) { + continue + } + if (await canBindPort(port)) { + return port + } + } + + const reserved = await reserveEphemeralPort() + return reserved +} + +function canBindPort(port: number): Promise { + return new Promise((resolve) => { + const server = createServer() + server.once('error', () => { + resolve(false) + }) + server.listen({ host: '127.0.0.1', port }, () => { + server.close(() => { + resolve(true) + }) + }) + }) +} + +function reserveEphemeralPort(): Promise { + return new Promise((resolve, reject) => { + const server = createServer() + server.once('error', reject) + server.listen({ host: '127.0.0.1', port: 0 }, () => { + const address = server.address() + const port = typeof address === 'object' && address ? address.port : 0 + server.close(() => { + if (port > 0) { + resolve(port) + } else { + reject(new Error('playwright_cdp_port_unavailable')) + } + }) + }) + }) +} + +function markPlaywrightCDPHealthy(): void { + cdpHealthState = { + status: 'healthy', + checkedAt: Date.now(), + unavailableSince: null, + nextRetryAt: null, + lastError: null, + } +} + +function markPlaywrightCDPUnavailable(error: unknown): void { + const now = Date.now() + cdpHealthState = { + status: 'unavailable', + checkedAt: now, + unavailableSince: cdpHealthState.unavailableSince ?? now, + nextRetryAt: now + cdpCircuitOpenTTLms, + lastError: error instanceof Error ? error.message : String(error), + } +} diff --git a/apps/desktop-client/src/main/publish-scheduler.ts b/apps/desktop-client/src/main/publish-scheduler.ts index d3622a7..d545112 100644 --- a/apps/desktop-client/src/main/publish-scheduler.ts +++ b/apps/desktop-client/src/main/publish-scheduler.ts @@ -32,6 +32,7 @@ interface PublishSchedulerSelectionOptions { now?: number maxConcurrency: number activePlatforms?: ReadonlySet + blockedPlatforms?: ReadonlySet activeCount: number } @@ -97,6 +98,10 @@ export function selectNextPublishTask( }) for (const candidate of candidates) { + if (options.blockedPlatforms?.has(candidate.platform)) { + continue + } + if (activePlatforms.has(candidate.platform)) { continue } @@ -143,6 +148,19 @@ export function notePublishTaskLeaseMiss(taskId: string): void { tasks.delete(taskId) } +export function notePublishTaskDeferred(taskId: string): void { + const existing = tasks.get(taskId) + if (!existing) { + return + } + + tasks.set(taskId, { + ...existing, + state: 'queued', + updatedAt: Date.now(), + }) +} + export function notePublishTaskCompleted(taskId: string, platform?: string): void { const existing = tasks.get(taskId) const resolvedPlatform = (platform ?? existing?.platform ?? '').trim() diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index afa907f..4a3670d 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -68,10 +68,17 @@ import { noteMonitorTaskActivated, noteMonitorTaskCanceled, noteMonitorTaskCompleted, + noteMonitorTaskDeferred, noteMonitorTaskLeased, selectNextMonitorTask, } from './monitor-scheduler' -import { retainHiddenPlaywrightPage, startHiddenPlaywrightReaper } from './playwright-cdp' +import { + ensurePlaywrightCDPReady, + isPlaywrightCDPExecutionAvailable, + isPlaywrightCDPUnavailableError, + retainHiddenPlaywrightPage, + startHiddenPlaywrightReaper, +} from './playwright-cdp' import { getProcessMetricsSnapshot } from './process-metrics' import { enqueuePublishTask as enqueuePublishSchedulerTask, @@ -80,6 +87,7 @@ import { notePublishTaskActivated, notePublishTaskCanceled, notePublishTaskCompleted, + notePublishTaskDeferred, notePublishTaskLeaseMiss, selectNextPublishTask, } from './publish-scheduler' @@ -102,6 +110,7 @@ import { clearSessionHandle, createSessionHandle, hasLocalSessionPartition, + type SessionHandle, } from './session-registry' import { cancelDesktopTask, @@ -148,6 +157,8 @@ const heartbeatIntervalMs = 15_000 const pullIntervalMs = 60_000 const publishPullIntervalMs = 30_000 const monitorPullFallbackIntervalMs = 10_000 +const playwrightCDPPreflightDeferMs = 15_000 +const playwrightCDPRecoveryActivityCooldownMs = 60_000 const accountSyncIntervalMs = 30_000 const pumpWatchdogIntervalMs = 30_000 const leaseExtendIntervalMs = 60_000 @@ -163,6 +174,8 @@ const maximumRuntimeTotalConcurrency = 4 const BAIJIAHAO_RISK_CONTROL_PROMPT = '您触发平台风控，请手动发稿一篇并完成验证，方可继续使用' const accountHealthReportDebounceMs = 1_000 const accountHealthReportMaxBatchSize = 100 +const maxPlaywrightCDPSoftRecoveries = 3 +const playwrightCDPFatalMarker = '--geo-cdp-fatal' interface RuntimeTaskRecord { id: string @@ -312,6 +325,18 @@ const state: RuntimeState = { activitySeq: 0, } let accountHealthReportBridgeUnsubscribe: (() => void) | null = null +let lastPlaywrightCDPRecoveryActivityAt = 0 +let playwrightCDPRecoveryAttemptCount = process.argv.filter((arg) => + arg.startsWith('--geo-cdp-recovery-at='), +).length +let playwrightCDPFatal = process.argv.includes(playwrightCDPFatalMarker) + +class PlaywrightCDPInfrastructureError extends Error { + constructor(message: string) { + super(message) + this.name = 'PlaywrightCDPInfrastructureError' + } +} function accountIdentityFromDesktopAccount(account: DesktopAccountInfo): PublishAccountIdentity { return { @@ -703,6 +728,17 @@ function startRuntime(): void { if (!canRunLive(state.session) || state.running) { return } + if (playwrightCDPFatal) { + const message = '浏览器自动化通道不可用，客户端已停止采集。请重启客户端或检查安全软件/端口占用。' + state.lastError = message + setSchedulerPhase('paused') + setSchedulerError(message) + recordActivity('danger', '浏览器自动化组件不可用', message) + emitRuntimeInvalidated('playwright-cdp-fatal', { + message, + }) + return + } initScheduler() initMonitorScheduler() @@ -1850,17 +1886,40 @@ function asSubmitUncertainExecution(result: AdapterExecutionResult): AdapterExec } } +async function retainPlaywrightPageOrThrow( + task: RuntimeTaskRecord, + payload: Record, + sessionHandle: SessionHandle, +) { + try { + await ensurePlaywrightCDPReady() + return await retainHiddenPlaywrightPage({ + accountId: task.accountId || task.id, + session: sessionHandle.session, + targetURL: resolvePlaywrightTargetURL(task.platform, payload), + title: `${task.title} · Hidden Playwright`, + }) + } catch (error) { + if (isPlaywrightCDPUnavailableError(error)) { + throw new PlaywrightCDPInfrastructureError(errorMessage(error)) + } + throw error + } +} + function pumpExecutionLoop(): void { - if (!state.running) { + if (!state.running || playwrightCDPFatal) { return } syncSchedulerSurface() + const blockedPlaywrightPlatforms = blockedPlaywrightExecutionPlatforms() if (canStartAnotherPublishExecution()) { const selected = selectNextPublishTask({ maxConcurrency: resolveAdaptiveTotalConcurrency(), activePlatforms: activePublishPlatforms(), + blockedPlatforms: blockedPlaywrightPlatforms, activeCount: activePublishCount(), }) if (selected) { @@ -1883,6 +1942,7 @@ function pumpExecutionLoop(): void { maxConcurrency: resolveAdaptiveMonitorConcurrency(), activePlatforms: activeMonitorPlatforms(), activeQuestionKeys: activeMonitorQuestionKeys(), + blockedPlatforms: blockedPlaywrightPlatforms, activeCount: activeMonitorCount(), }) if (selected) { @@ -1912,6 +1972,10 @@ async function leaseSpecificTask( return } + if (!(await ensurePlaywrightCDPAdmissionForRequest(expectedKind, request))) { + return + } + state.leaseInFlightKinds.add(expectedKind) try { @@ -1967,7 +2031,11 @@ async function leaseSpecificTask( async function pullNextTask(kind: 'publish' | 'monitor'): Promise { if (kind === 'monitor') { - if (!canRunLive(state.session) || isLeaseInFlight('monitor')) { + if (!canRunLive(state.session) || isLeaseInFlight('monitor') || playwrightCDPFatal) { + return + } + + if (!(await ensurePlaywrightCDPAdmissionForFallback('monitor'))) { return } @@ -2002,7 +2070,11 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise { return } - if (!canRunLive(state.session) || isLeaseInFlight('publish')) { + if (!canRunLive(state.session) || isLeaseInFlight('publish') || playwrightCDPFatal) { + return + } + + if (!(await ensurePlaywrightCDPAdmissionForFallback('publish'))) { return } @@ -2046,7 +2118,7 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise { async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise { const leaseLimit = resolveLegacyMonitoringLeaseLimit() - if (!canRunLive(state.session) || isLeaseInFlight('monitor') || leaseLimit <= 0) { + if (!canRunLive(state.session) || isLeaseInFlight('monitor') || leaseLimit <= 0 || playwrightCDPFatal) { return } @@ -2270,6 +2342,15 @@ async function executeLeasedMonitoringTask( `${taskRecord.title} ${execution.summary}`, ) } catch (error) { + if (isPlaywrightCDPInfrastructureError(error)) { + await deferMonitoringLeaseAfterInfrastructureFailure( + taskRecord, + monitoringTaskID, + taskRecord.leaseToken, + ) + return + } + if (isMonitorAbortReason(taskRecord, error, 'stale_business_date')) { let dropSummary = staleMonitoringDropSummary(taskRecord) try { @@ -2531,6 +2612,11 @@ async function executeLeasedDesktopMonitorTask( `${taskRecord.title} ${execution.summary}`, ) } catch (error) { + if (isPlaywrightCDPInfrastructureError(error)) { + await deferDesktopMonitorTaskAfterInfrastructureFailure(taskRecord, leased, monitoringTaskID) + return + } + if (isMonitorAbortReason(taskRecord, error, 'stale_business_date')) { let dropSummary = staleMonitoringDropSummary(taskRecord) try { @@ -3024,12 +3110,7 @@ async function executeTaskAdapter( : retainHotView(task.accountId || task.id) const playwrightLease = adapter.executionMode === 'playwright' - ? await retainHiddenPlaywrightPage({ - accountId: task.accountId || task.id, - session: sessionHandle.session, - targetURL: resolvePlaywrightTargetURL(task.platform, payload), - title: `${task.title} · Hidden Playwright`, - }) + ? await retainPlaywrightPageOrThrow(task, payload, sessionHandle) : null try { @@ -3098,12 +3179,7 @@ async function executeTaskAdapter( : retainHotView(task.accountId || task.id) const playwrightLease = adapter.executionMode === 'playwright' - ? await retainHiddenPlaywrightPage({ - accountId: task.accountId || task.id, - session: sessionHandle.session, - targetURL: resolvePlaywrightTargetURL(task.platform, payload), - title: `${task.title} · Hidden Playwright`, - }) + ? await retainPlaywrightPageOrThrow(task, payload, sessionHandle) : null try { @@ -3534,6 +3610,238 @@ function isLeaseInFlight(kind: 'publish' | 'monitor'): boolean { return state.leaseInFlightKinds.has(kind) } +function blockedPlaywrightExecutionPlatforms(): Set { + if (isPlaywrightCDPExecutionAvailable()) { + return new Set() + } + + return new Set([ + ...state.accounts + .map((account) => account.platform) + .filter((platform) => adapterRequiresPlaywright('monitor', platform)), + ...getPublishSchedulerSnapshot() + .tasks.map((task) => task.platform) + .filter((platform) => adapterRequiresPlaywright('publish', platform)), + ...getMonitorSchedulerSnapshot() + .tasks.map((task) => task.platform) + .filter((platform) => adapterRequiresPlaywright('monitor', platform)), + ]) +} + +async function ensurePlaywrightCDPAdmissionForRequest( + kind: 'publish' | 'monitor', + request: PendingTaskRequest, +): Promise { + if (!request.platform || !adapterRequiresPlaywright(kind, request.platform)) { + return true + } + + const ready = await ensurePlaywrightCDPAdmission() + if (!ready) { + deferPendingTaskRequest(kind, request, playwrightCDPPreflightDeferMs) + } + return ready +} + +async function ensurePlaywrightCDPAdmissionForFallback(kind: 'publish' | 'monitor'): Promise { + if (kind === 'publish') { + const queuedPlaywrightPublish = getPublishSchedulerSnapshot().tasks.some( + (task) => task.state === 'queued' && adapterRequiresPlaywright('publish', task.platform), + ) + if (!queuedPlaywrightPublish) { + return true + } + return ensurePlaywrightCDPAdmission() + } + + const hasNonPlaywrightMonitorAccount = state.accounts.some( + (account) => + isAIPlatformId(account.platform) && + !adapterRequiresPlaywright('monitor', account.platform) && + getProjectedAccountHealth({ + accountId: account.id, + platform: account.platform, + health: account.health, + verifiedAt: account.verified_at, + }).health === 'live', + ) + if (hasNonPlaywrightMonitorAccount) { + return true + } + + return ensurePlaywrightCDPAdmission() +} + +async function ensurePlaywrightCDPAdmission(): Promise { + try { + await ensurePlaywrightCDPReady() + return true + } catch (error) { + handlePlaywrightCDPUnavailable(error) + return false + } +} + +function isPlaywrightCDPInfrastructureError(error: unknown): error is PlaywrightCDPInfrastructureError { + return error instanceof PlaywrightCDPInfrastructureError || isPlaywrightCDPUnavailableError(error) +} + +function deferPendingTaskRequest( + kind: 'publish' | 'monitor', + request: PendingTaskRequest, + delayMs: number, +): void { + if (kind === 'publish') { + notePublishTaskDeferred(request.taskId) + state.publishFallbackBackoffUntil = Date.now() + Math.max(delayMs, publishPullIntervalMs) + return + } + + noteMonitorTaskDeferred(request.taskId, delayMs) + state.monitorFallbackBackoffUntil = Date.now() + Math.max(delayMs, monitorPullFallbackIntervalMs) +} + +function adapterRequiresPlaywright(kind: 'publish' | 'monitor', platform: string): boolean { + const adapter = kind === 'publish' ? selectPublishAdapter(platform) : selectMonitorAdapter(platform) + return adapter?.executionMode === 'playwright' +} + +function handlePlaywrightCDPUnavailable(error: unknown, taskId?: string): void { + const message = errorMessage(error) || 'playwright_cdp_endpoint_unavailable' + state.lastError = message + setSchedulerError(message) + state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs + state.publishFallbackBackoffUntil = Date.now() + publishPullIntervalMs + + if (playwrightCDPRecoveryAttemptCount >= maxPlaywrightCDPSoftRecoveries) { + markPlaywrightCDPFatal(message, taskId) + return + } + playwrightCDPRecoveryAttemptCount += 1 + + const now = Date.now() + if (now - lastPlaywrightCDPRecoveryActivityAt >= playwrightCDPRecoveryActivityCooldownMs) { + lastPlaywrightCDPRecoveryActivityAt = now + recordActivity( + 'warn', + '浏览器自动化通道异常', + '已暂停领取 Playwright 平台任务，客户端将自动恢复 CDP 后继续采集。', + ) + } + + emitRuntimeInvalidated('playwright-cdp-recovery', { + taskId, + message, + }) + syncSchedulerSurface() +} + +function markPlaywrightCDPFatal(message: string, taskId?: string): void { + if (playwrightCDPFatal) { + return + } + playwrightCDPFatal = true + + const fatalMessage = + message || + '浏览器自动化通道不可用，客户端已停止采集。请重启客户端或检查安全软件/端口占用。' + state.lastError = fatalMessage + setSchedulerPhase('paused') + setSchedulerError(fatalMessage) + setTransportDispatchState('idle') + state.dispatchWsConnected = false + state.dispatchWsClient?.stop() + state.dispatchWsClient = null + state.leaseInFlightKinds.clear() + recordActivity('danger', '浏览器自动化组件不可用', fatalMessage) + + void offlineDesktopClient().catch((offlineError) => { + console.warn('[desktop-runtime] failed to mark client offline after cdp fatal', { + message: errorMessage(offlineError), + }) + }) + + emitRuntimeInvalidated('playwright-cdp-fatal', { + taskId, + message: fatalMessage, + }) + syncSchedulerSurface() +} + +async function deferMonitoringLeaseAfterInfrastructureFailure( + task: RuntimeTaskRecord, + monitoringTaskID: number, + leaseToken: string, + reason = 'playwright_cdp_unavailable', +): Promise { + try { + await cancelMonitoringTask(monitoringTaskID, { + lease_token: leaseToken, + reason, + }) + } catch (cancelError) { + recordActivity( + 'warn', + '监控租约释放失败', + `${task.title} 基础设施异常后释放租约失败：${errorMessage(cancelError)}`, + ) + } + + const existing = state.tasks.get(task.id) + if (existing) { + existing.status = 'queued' + existing.summary = '浏览器自动化通道异常，已释放租约并等待客户端自动恢复后重试。' + existing.error = { + code: reason, + message: 'playwright cdp is unavailable; task deferred without marking collection failed', + } + existing.updatedAt = Date.now() + existing.attemptId = null + existing.leaseToken = null + state.tasks.set(task.id, existing) + } + noteLeaseReleased('completed', task.id) + noteMonitorTaskDeferred(task.id, playwrightCDPPreflightDeferMs) + handlePlaywrightCDPUnavailable(new Error(reason), task.id) +} + +async function deferDesktopMonitorTaskAfterInfrastructureFailure( + task: RuntimeTaskRecord, + leased: LeaseDesktopTaskResponse, + monitoringTaskID: number, + reason = 'playwright_cdp_unavailable', +): Promise { + try { + await cancelMonitoringTask(monitoringTaskID, { + lease_token: leased.lease_token as string, + reason, + }) + } catch (cancelError) { + recordActivity( + 'warn', + '监控租约释放失败', + `${task.title} 基础设施异常后释放 phase2 租约失败：${errorMessage(cancelError)}`, + ) + } + + const existing = state.tasks.get(task.id) + if (existing) { + existing.status = 'queued' + existing.summary = '浏览器自动化通道异常，任务已回队等待客户端自动恢复。' + existing.error = { + code: reason, + message: 'playwright cdp is unavailable; desktop monitor task requeued', + } + existing.updatedAt = Date.now() + existing.attemptId = null + existing.leaseToken = null + state.tasks.set(task.id, existing) + } + noteLeaseReleased('completed', task.id) + noteMonitorTaskDeferred(task.id, playwrightCDPPreflightDeferMs) + handlePlaywrightCDPUnavailable(new Error(reason), task.id) +} + function activePublishCount(): number { return [...state.activeExecutions.values()].filter((item) => item.kind === 'publish').length } diff --git a/apps/desktop-client/src/main/runtime-events.ts b/apps/desktop-client/src/main/runtime-events.ts index 6feb4f8..4e9d59c 100644 --- a/apps/desktop-client/src/main/runtime-events.ts +++ b/apps/desktop-client/src/main/runtime-events.ts @@ -1,17 +1,23 @@ -type RuntimeInvalidationReason = 'account-health' | 'publish-task-lease' | 'publish-task-progress' +type RuntimeInvalidationReason = + | 'account-health' + | 'publish-task-lease' + | 'publish-task-progress' + | 'playwright-cdp-recovery' + | 'playwright-cdp-fatal' type RuntimeInvalidationListener = (event: { reason: RuntimeInvalidationReason at: number accountId?: string taskId?: string + message?: string }) => void const listeners = new Set() export function emitRuntimeInvalidated( reason: RuntimeInvalidationReason, - details: { accountId?: string; taskId?: string } = {}, + details: { accountId?: string; taskId?: string; message?: string } = {}, ): void { const event = { reason, diff --git a/apps/desktop-client/src/preload/bridge.ts b/apps/desktop-client/src/preload/bridge.ts index 743a00e..fce63f7 100644 --- a/apps/desktop-client/src/preload/bridge.ts +++ b/apps/desktop-client/src/preload/bridge.ts @@ -249,19 +249,31 @@ const desktopBridge = { ipcRenderer.invoke('desktop:set-window-mode', mode, request) as Promise, onRuntimeInvalidated: ( listener: (event: { - reason: 'account-health' | 'publish-task-lease' | 'publish-task-progress' + reason: + | 'account-health' + | 'publish-task-lease' + | 'publish-task-progress' + | 'playwright-cdp-recovery' + | 'playwright-cdp-fatal' at: number accountId?: string taskId?: string + message?: string }) => void, ) => { const wrapped = ( _event: IpcRendererEvent, payload: { - reason: 'account-health' | 'publish-task-lease' | 'publish-task-progress' + reason: + | 'account-health' + | 'publish-task-lease' + | 'publish-task-progress' + | 'playwright-cdp-recovery' + | 'playwright-cdp-fatal' at: number accountId?: string taskId?: string + message?: string }, ) => { listener(payload) diff --git a/apps/desktop-client/src/renderer/env.d.ts b/apps/desktop-client/src/renderer/env.d.ts index 9b88438..30ee1e9 100644 --- a/apps/desktop-client/src/renderer/env.d.ts +++ b/apps/desktop-client/src/renderer/env.d.ts @@ -142,10 +142,16 @@ declare global { ): Promise onRuntimeInvalidated( listener: (event: { - reason: 'account-health' | 'publish-task-lease' | 'publish-task-progress' + reason: + | 'account-health' + | 'publish-task-lease' + | 'publish-task-progress' + | 'playwright-cdp-recovery' + | 'playwright-cdp-fatal' at: number accountId?: string taskId?: string + message?: string }) => void, ): () => void onNetworkObserved(listener: (event: DesktopObservedRequest) => void): () => void diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 9364993..4d39b4e 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -1901,7 +1901,8 @@ type PublishLeaseRecoveryResult struct { } type MonitorLeaseRecoveryResult struct { - Failed int `json:"failed"` + Requeued int `json:"requeued"` + Failed int `json:"failed"` } // resolvePublishRecoveryOutcome decides what to do with a publish task whose lease was lost. @@ -1923,7 +1924,7 @@ func resolvePublishRecoveryOutcome(submitStarted bool, attempts int) (status str } func resolveMonitorRecoveryOutcome(mode desktopTaskRecoveryMode, attempts int) (status string, payloadKind string) { - if mode == desktopTaskRecoveryModeLeaseExpiry || attempts >= 2 { + if mode != desktopTaskRecoveryModeLeaseExpiry && attempts >= 2 { return "failed", "monitor_final" } return "queued", "monitor" @@ -2026,12 +2027,21 @@ func (s *DesktopTaskService) recoverClientTasks( continue } } + } else if item.MonitorTaskID.Valid { + requeuedMonitorTaskIDs, requeueErr := s.requeueMonitorCollectTasksForDesktopRecovery(ctx, []int64{item.MonitorTaskID.Int64}, monitorErrorForTask) + if requeueErr != nil { + return requeueErr + } + if len(requeuedMonitorTaskIDs) == 0 { + continue + } } if _, execErr := tx.Exec(ctx, ` - UPDATE desktop_tasks - SET status = $2, - error = $3, - active_attempt_id = NULL, + UPDATE desktop_tasks + SET status = $2, + error = $3, + result = CASE WHEN $2 = 'queued' THEN NULL ELSE result END, + active_attempt_id = NULL, lease_token_hash = NULL, lease_expires_at = NULL, interrupted_at = COALESCE(interrupted_at, NOW()), @@ -2355,7 +2365,7 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim limit = 1000 } - errorJSON, reason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor_final") + errorJSON, reason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor") if err != nil { return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload") } @@ -2390,7 +2400,7 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim s.logWarn("expired desktop monitor recovery scan failed", scanErr) return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks") } - item.Status = "failed" + item.Status = "queued" item.ErrorJSON = errorJSON recovered = append(recovered, item) if item.MonitorTaskID.Valid { @@ -2402,18 +2412,18 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks") } - failedMonitorTaskIDs, err := s.failMonitorCollectTasksForDesktopRecovery(ctx, monitorTaskIDs, errorJSON) + requeuedMonitorTaskIDs, err := s.requeueMonitorCollectTasksForDesktopRecovery(ctx, monitorTaskIDs, errorJSON) if err != nil { return result, err } - failedMonitorTaskSet := int64Set(failedMonitorTaskIDs) + requeuedMonitorTaskSet := int64Set(requeuedMonitorTaskIDs) recoverableDesktopIDs := make([]uuid.UUID, 0, len(recovered)) recoverable := make([]recoveredDesktopTaskLease, 0, len(recovered)) for _, item := range recovered { if !item.MonitorTaskID.Valid { continue } - if _, ok := failedMonitorTaskSet[item.MonitorTaskID.Int64]; !ok { + if _, ok := requeuedMonitorTaskSet[item.MonitorTaskID.Int64]; !ok { continue } recoverableDesktopIDs = append(recoverableDesktopIDs, item.TaskID) @@ -2423,7 +2433,7 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim return result, nil } - updateRows, err := tx.Query(ctx, failExpiredMonitorDesktopTasksSQL(), recoverableDesktopIDs, errorJSON, reason) + updateRows, err := tx.Query(ctx, requeueExpiredMonitorDesktopTasksSQL(), recoverableDesktopIDs, errorJSON, reason) if err != nil { s.logWarn("expired desktop monitor recovery update failed", err) return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks") @@ -2462,7 +2472,7 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{ TaskID: item.TaskID, AttemptID: attemptID, - FinalStatus: literalStringPtr("failed"), + FinalStatus: literalStringPtr("aborted"), Error: errorJSON, }); finishErr != nil { s.logWarn("expired desktop monitor recovery attempt finish failed", finishErr, zap.String("task_id", item.TaskID.String())) @@ -2473,7 +2483,7 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim return result, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery") } - result.Failed = len(updatedDesktopIDs) + result.Requeued = len(updatedDesktopIDs) for _, item := range recoverable { if _, ok := updatedDesktopIDs[item.TaskID]; !ok { @@ -2484,12 +2494,12 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim s.logWarn("expired desktop monitor recovery reload after commit failed", getErr, zap.String("task_id", item.TaskID.String())) continue } - s.publishTaskEvent(ctx, task, "task_completed") + s.publishTaskEvent(ctx, task, "task_available") } - if result.Failed > 0 && s.logger != nil { + if result.Requeued > 0 && s.logger != nil { s.logger.Info("expired desktop monitor tasks recovered", - zap.Int("failed", result.Failed), + zap.Int("requeued", result.Requeued), ) } @@ -2519,16 +2529,18 @@ func expiredMonitorDesktopTasksCandidateSQL() string { ` } -func failExpiredMonitorDesktopTasksSQL() string { +func requeueExpiredMonitorDesktopTasksSQL() string { return ` UPDATE desktop_tasks - SET status = 'failed', + SET status = 'queued', error = $2, + result = NULL, active_attempt_id = NULL, lease_token_hash = NULL, lease_expires_at = NULL, interrupted_at = COALESCE(interrupted_at, NOW()), interrupt_reason = COALESCE(interrupt_reason, $3), + enqueued_at = NOW(), updated_at = NOW() WHERE desktop_id = ANY($1::uuid[]) AND kind = 'monitor' @@ -2537,6 +2549,69 @@ func failExpiredMonitorDesktopTasksSQL() string { ` } +func (s *DesktopTaskService) requeueMonitorCollectTasksForDesktopRecovery(ctx context.Context, monitorTaskIDs []int64, errorJSON []byte) ([]int64, error) { + if s == nil || s.monitoringPool == nil || len(monitorTaskIDs) == 0 { + return nil, nil + } + monitorTaskIDs = uniqueInt64s(monitorTaskIDs) + message := desktopTaskRecoveryMessage(errorJSON) + requestPayloadJSON, err := json.Marshal(map[string]any{ + "runtime_error": json.RawMessage(errorJSON), + }) + if err != nil { + return nil, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload") + } + rows, err := s.monitoringPool.Query(ctx, ` + WITH input AS ( + SELECT unnest($1::bigint[]) AS id + ), + updated AS ( + 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, + completed_at = NULL, + skip_reason = NULL, + dispatch_after = NOW() + interval '30 seconds', + error_message = NULLIF($2, ''), + 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 IN (SELECT id FROM input) + AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks' + AND callback_received_at IS NULL + AND status IN ('pending', 'leased', 'expired') + RETURNING id + ) + SELECT id FROM updated + `, monitorTaskIDs, message, requestPayloadJSON) + if err != nil { + return nil, response.ErrInternal(50041, "task_update_failed", "failed to requeue expired monitor desktop tasks") + } + defer rows.Close() + + requeuedIDs := make([]int64, 0, len(monitorTaskIDs)) + for rows.Next() { + var id int64 + if scanErr := rows.Scan(&id); scanErr != nil { + return nil, response.ErrInternal(50041, "task_update_failed", "failed to parse requeued monitor desktop tasks") + } + requeuedIDs = append(requeuedIDs, id) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "task_update_failed", "failed to iterate requeued monitor desktop tasks") + } + return requeuedIDs, nil +} + func (s *DesktopTaskService) failMonitorCollectTasksForDesktopRecovery(ctx context.Context, monitorTaskIDs []int64, errorJSON []byte) ([]int64, error) { if s == nil || s.monitoringPool == nil || len(monitorTaskIDs) == 0 { return nil, nil @@ -2586,8 +2661,8 @@ func (s *DesktopTaskService) failMonitorCollectTasksForDesktopRecovery(ctx conte AND t.status = 'failed' AND ( t.error_message = $2 - OR t.request_payload_json #>> '{runtime_error,source}' = 'desktop_task_lease_expiry' - OR t.request_payload_json #>> '{runtime_error,reason}' = 'lease_expired' + OR t.request_payload_json #>> '{runtime_error,source}' = 'desktop_task_recovery' + OR t.request_payload_json #>> '{runtime_error,source}' = 'desktop_client_offline' ) ) SELECT id FROM updated @@ -2595,7 +2670,7 @@ func (s *DesktopTaskService) failMonitorCollectTasksForDesktopRecovery(ctx conte SELECT id FROM already_failed `, monitorTaskIDs, message, requestPayloadJSON) if err != nil { - return nil, response.ErrInternal(50041, "task_update_failed", "failed to fail expired monitor desktop tasks") + return nil, response.ErrInternal(50041, "task_update_failed", "failed to fail recovered monitor desktop tasks") } defer rows.Close() @@ -2700,6 +2775,8 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by message = "desktop publish task may have already been submitted to the platform before the lease was lost; kept as unknown for manual reconcile to avoid duplicate publishing" } else if isMonitorFinal { message = "desktop monitor task lost its lease; task has been marked failed to avoid repeated platform retries" + } else if kind == "monitor" { + message = "desktop monitor task lost its lease; task has been re-queued" } switch mode { @@ -2735,8 +2812,10 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by message = fmt.Sprintf("desktop task lease expired before publish completion; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts) } else if isPublishUncertain { message = "desktop task lease expired while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing" - } else if isMonitorFinal || kind == "monitor" { + } else if isMonitorFinal { message = "desktop task lease expired before monitor completion; task has been marked failed" + } else if kind == "monitor" { + message = "desktop task lease expired before monitor completion; task has been re-queued" } else { message = "desktop task lease expired before publish completion; task has been re-queued for retry" } diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index 25758e3..92d5bbb 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -101,21 +101,21 @@ func TestDesktopTaskRecoveryPayloadPublishFinalFailsAfterMaxAttempts(t *testing. } } -func TestDesktopTaskRecoveryPayloadMonitorLeaseExpiryFails(t *testing.T) { +func TestDesktopTaskRecoveryPayloadMonitorLeaseExpiryRequeues(t *testing.T) { t.Parallel() - payload, reason, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor_final") + payload, reason, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor") if err != nil { t.Fatalf("desktopTaskRecoveryPayload returned error: %v", err) } if reason != "lease_expired" { t.Fatalf("reason = %q, want lease_expired", reason) } - if !strings.Contains(message, "marked failed") { - t.Fatalf("message = %q, want marked failed", message) + if !strings.Contains(message, "re-queued") { + t.Fatalf("message = %q, want re-queued", message) } - if strings.Contains(message, "re-queued") { - t.Fatalf("message must not requeue expired monitor tasks: %q", message) + if strings.Contains(message, "marked failed") { + t.Fatalf("message must not fail expired monitor tasks: %q", message) } if !strings.Contains(string(payload), "desktop_task_lease_expiry") { t.Fatalf("payload missing source: %s", payload) @@ -176,7 +176,8 @@ func TestResolveMonitorRecoveryOutcome(t *testing.T) { {"disconnect first attempt requeues", desktopTaskRecoveryModeDisconnect, 1, "queued", "monitor"}, {"startup repeated recovery fails", desktopTaskRecoveryModeStartup, 2, "failed", "monitor_final"}, {"disconnect repeated recovery fails", desktopTaskRecoveryModeDisconnect, 2, "failed", "monitor_final"}, - {"lease expiry fails immediately", desktopTaskRecoveryModeLeaseExpiry, 1, "failed", "monitor_final"}, + {"lease expiry requeues first attempt", desktopTaskRecoveryModeLeaseExpiry, 1, "queued", "monitor"}, + {"lease expiry requeues repeated attempt", desktopTaskRecoveryModeLeaseExpiry, 2, "queued", "monitor"}, } for _, tc := range cases { @@ -221,16 +222,17 @@ func TestExpiredMonitorDesktopTasksCandidateSQLSelectsOnlyExpiredInProgressMonit } } -func TestFailExpiredMonitorDesktopTasksSQLMarksTerminalFailure(t *testing.T) { +func TestRequeueExpiredMonitorDesktopTasksSQLReturnsTaskToQueue(t *testing.T) { t.Parallel() - normalized := normalizeSQLForDesktopTaskTest(failExpiredMonitorDesktopTasksSQL()) + normalized := normalizeSQLForDesktopTaskTest(requeueExpiredMonitorDesktopTasksSQL()) for _, fragment := range []string{ - "SET status = 'failed'", + "SET status = 'queued'", "active_attempt_id = NULL", "lease_token_hash = NULL", "lease_expires_at = NULL", "interrupt_reason = COALESCE(interrupt_reason, $3)", + "enqueued_at = NOW()", "WHERE desktop_id = ANY($1::uuid[])", "AND kind = 'monitor'", "AND status = 'in_progress'", diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index 7ea2262..3783d5b 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -40,6 +40,7 @@ const ( // 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 + monitoringRuntimeUnknownMaxDispatchAttempts = 3 ) var monitoringCitationMarkerPattern = regexp.MustCompile(`(?i)\[\s*citation\s*:\s*\d+\s*\]`) @@ -194,6 +195,15 @@ type MonitoringCancelTaskResponse struct { TaskStatus string `json:"task_status"` } +func isMonitoringInfrastructureCancelReason(reason *string) bool { + switch strings.TrimSpace(optionalStringValue(reason)) { + case "desktop_infra_unavailable", "playwright_cdp_unavailable", "playwright_cdp_recovery": + return true + default: + return false + } +} + type monitoringInstallationIdentity struct { ID string TenantID int64 @@ -250,6 +260,7 @@ type monitoringCollectTask struct { TriggerSource string QuestionText string TargetClientID sql.NullString + DispatchAttempts int } type monitoringDesktopExecutionLease struct { @@ -258,6 +269,7 @@ type monitoringDesktopExecutionLease struct { TargetClientID uuid.UUID LeaseTokenHash []byte InterruptGeneration int + ActiveAttemptID pgtype.UUID } type monitoringLeasedTaskRow struct { @@ -945,7 +957,24 @@ func (s *MonitoringCallbackService) skipTaskForInstallation( } skipReason := normalizeSkipReason(req.SkipReason) - skipOutcome := monitoringSkipOutcomeForReason(skipReason) + skipOutcome := monitoringSkipOutcomeForReason(skipReason, task.DispatchAttempts, task.TriggerSource) + errorMessage := strings.TrimSpace(optionalStringValue(req.ErrorMessage)) + requestPayloadJSON := []byte(`{}`) + if skipReason == "runtime_unknown" { + var marshalErr error + requestPayloadJSON, marshalErr = json.Marshal(map[string]any{ + "runtime_unknown": map[string]any{ + "reason": skipReason, + "message": errorMessage, + "dispatch_attempts": task.DispatchAttempts, + "max_attempts": monitoringRuntimeUnknownMaxDispatchAttempts, + "retryable": skipOutcome.Requeue, + }, + }) + if marshalErr != nil { + return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode monitoring task skip payload") + } + } tx, err := s.monitoringPool.Begin(ctx) if err != nil { @@ -953,16 +982,48 @@ func (s *MonitoringCallbackService) skipTaskForInstallation( } defer tx.Rollback(ctx) - if _, err := tx.Exec(ctx, ` - UPDATE monitoring_collect_tasks - SET status = $2, - callback_received_at = NOW(), - completed_at = $3, - skip_reason = $4, - error_message = $5, - updated_at = NOW() - WHERE id = $1 - `, task.ID, skipOutcome.TaskStatus, completedAt, nullableStringPtr(skipOutcome.StoredSkipReason), nullableString(req.ErrorMessage)); err != nil { + if skipOutcome.Requeue { + if _, err := tx.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, + completed_at = NULL, + skip_reason = NULL, + dispatch_after = NOW() + $2::interval, + error_message = NULLIF($3, ''), + 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, formatPgInterval(skipOutcome.RetryAfter), errorMessage, requestPayloadJSON); err != nil { + return nil, response.ErrInternal(50041, "task_update_failed", "failed to requeue monitoring task after runtime unknown") + } + } else if _, err := tx.Exec(ctx, ` + UPDATE monitoring_collect_tasks + SET status = $2, + callback_received_at = NOW(), + completed_at = $3, + skip_reason = $4, + error_message = $5, + request_payload_json = ( + CASE + WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object' + THEN '{}'::jsonb + ELSE request_payload_json + END + ) || $6::jsonb, + updated_at = NOW() + WHERE id = $1 + `, task.ID, skipOutcome.TaskStatus, completedAt, nullableStringPtr(skipOutcome.StoredSkipReason), nullableString(req.ErrorMessage), requestPayloadJSON); err != nil { return nil, response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task skip status") } @@ -976,29 +1037,56 @@ func (s *MonitoringCallbackService) skipTaskForInstallation( _ = s.touchInstallation(ctx, installation.ID) - completedAtText := completedAt.Format(time.RFC3339) + var completedAtText *string + if !skipOutcome.Requeue { + text := completedAt.Format(time.RFC3339) + completedAtText = &text + } return &MonitoringSkipTaskResponse{ TaskID: task.ID, TaskStatus: skipOutcome.TaskStatus, SkipReason: skipReason, - CompletedAt: &completedAtText, + CompletedAt: completedAtText, }, nil } type monitoringSkipOutcome struct { TaskStatus string StoredSkipReason string + Requeue bool + RetryAfter time.Duration } -func monitoringSkipOutcomeForReason(skipReason string) monitoringSkipOutcome { +func monitoringSkipOutcomeForReason(skipReason string, dispatchAttempts int, triggerSource string) monitoringSkipOutcome { switch strings.TrimSpace(skipReason) { case "runtime_unknown": + if strings.EqualFold(strings.TrimSpace(triggerSource), "manual") { + return monitoringSkipOutcome{TaskStatus: "failed"} + } + if dispatchAttempts < monitoringRuntimeUnknownMaxDispatchAttempts { + return monitoringSkipOutcome{ + TaskStatus: "pending", + Requeue: true, + RetryAfter: monitoringRuntimeUnknownRetryAfter(dispatchAttempts), + } + } return monitoringSkipOutcome{TaskStatus: "failed"} default: return monitoringSkipOutcome{TaskStatus: "skipped", StoredSkipReason: strings.TrimSpace(skipReason)} } } +func monitoringRuntimeUnknownRetryAfter(dispatchAttempts int) time.Duration { + switch { + case dispatchAttempts <= 0: + return 30 * time.Second + case dispatchAttempts == 1: + return 2 * time.Minute + default: + return 5 * time.Minute + } +} + func nullableStringPtr(value string) *string { trimmed := strings.TrimSpace(value) if trimmed == "" { @@ -1045,15 +1133,23 @@ func (s *MonitoringCallbackService) CancelTaskForDesktopClient( 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") + if isMonitoringInfrastructureCancelReason(req.Reason) { + requeuedTask, requeueErr := s.requeueDesktopMonitorTaskAfterInfrastructureCancel(ctx, desktopExecution, errorJSON) + if requeueErr != nil { + return nil, requeueErr } - return nil, response.ErrInternal(50041, "desktop_task_cancel_failed", "failed to cancel phase2 monitor desktop task lease") + desktopTaskCanceled = requeuedTask + } else { + 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 } - desktopTaskCanceled = canceledTask } else if task.Status != "leased" { return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow lease cancel") } @@ -1066,11 +1162,11 @@ func (s *MonitoringCallbackService) CancelTaskForDesktopClient( leased_at = NULL, lease_expires_at = NULL, callback_received_at = NULL, - dispatch_after = NOW(), + dispatch_after = CASE WHEN $3::boolean THEN NOW() + interval '30 seconds' ELSE NOW() END, error_message = $2, updated_at = NOW() WHERE id = $1 - `, task.ID, nullableString(req.Reason)); err != nil { + `, task.ID, nullableString(req.Reason), isMonitoringInfrastructureCancelReason(req.Reason)); err != nil { return nil, response.ErrInternal(50041, "task_update_failed", "failed to cancel monitoring task lease") } @@ -1106,6 +1202,72 @@ func (s *MonitoringCallbackService) CancelTaskForDesktopClient( }, nil } +func (s *MonitoringCallbackService) requeueDesktopMonitorTaskAfterInfrastructureCancel( + ctx context.Context, + lease *monitoringDesktopExecutionLease, + errorJSON []byte, +) (*repository.DesktopTask, error) { + if s == nil || s.businessPool == nil || lease == nil { + return nil, response.ErrInternal(50041, "desktop_task_requeue_failed", "failed to requeue phase2 monitor desktop task lease") + } + + tx, err := s.businessPool.Begin(ctx) + if err != nil { + return nil, response.ErrInternal(50041, "desktop_task_requeue_begin_failed", "failed to start phase2 monitor desktop task requeue") + } + defer tx.Rollback(ctx) + + repo := repository.NewDesktopTaskRepository(tx) + row := tx.QueryRow(ctx, ` + UPDATE desktop_tasks + SET status = 'queued', + error = $3, + result = NULL, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + interrupted_at = COALESCE(interrupted_at, NOW()), + interrupt_reason = COALESCE(interrupt_reason, 'desktop_infra_unavailable'), + enqueued_at = NOW(), + updated_at = NOW() + WHERE desktop_id = $1 + AND lease_token_hash = $2 + AND kind = 'monitor' + AND status = 'in_progress' + RETURNING `+desktopTaskRepositoryReturningColumns, + lease.DesktopTaskID, + lease.LeaseTokenHash, + errorJSON, + ) + task, scanErr := scanRepositoryDesktopTask(row) + if scanErr != nil { + if errors.Is(scanErr, 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_requeue_failed", "failed to requeue phase2 monitor desktop task lease") + } + + if lease.ActiveAttemptID.Valid { + attemptID, convErr := uuid.FromBytes(lease.ActiveAttemptID.Bytes[:]) + if convErr != nil { + return nil, response.ErrInternal(50041, "desktop_task_attempt_decode_failed", "failed to decode phase2 monitor desktop task attempt") + } + if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{ + TaskID: task.DesktopID, + AttemptID: attemptID, + FinalStatus: literalStringPtr("aborted"), + Error: errorJSON, + }); finishErr != nil { + return nil, response.ErrInternal(50041, "desktop_task_attempt_finish_failed", "failed to finish phase2 monitor desktop task attempt") + } + } + + if err := tx.Commit(ctx); err != nil { + return nil, response.ErrInternal(50041, "desktop_task_requeue_commit_failed", "failed to commit phase2 monitor desktop task requeue") + } + return task, 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") @@ -1393,6 +1555,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI completed_at, skip_reason, trigger_source, + COALESCE(dispatch_attempts, 0) AS dispatch_attempts, COALESCE(target_client_id::text, '') AS target_client_id, COALESCE(NULLIF(t.question_text_snapshot, ''), ( SELECT question_text_snapshot @@ -1430,6 +1593,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI &item.CompletedAt, &item.SkipReason, &item.TriggerSource, + &item.DispatchAttempts, &item.TargetClientID, &item.QuestionText, ); err != nil { @@ -1464,6 +1628,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas completed_at, skip_reason, trigger_source, + COALESCE(dispatch_attempts, 0) AS dispatch_attempts, COALESCE(target_client_id::text, '') AS target_client_id, COALESCE(NULLIF(t.question_text_snapshot, ''), ( SELECT question_text_snapshot @@ -1500,6 +1665,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas &item.CompletedAt, &item.SkipReason, &item.TriggerSource, + &item.DispatchAttempts, &item.TargetClientID, &item.QuestionText, ); err != nil { @@ -1518,23 +1684,25 @@ func (s *MonitoringCallbackService) loadActiveDesktopMonitorExecutionLease( ) (*monitoringDesktopExecutionLease, error) { item := &monitoringDesktopExecutionLease{} if err := s.businessPool.QueryRow(ctx, ` - SELECT - desktop_id, - status, - target_client_id, - lease_token_hash, - interrupt_generation + SELECT + desktop_id, + status, + target_client_id, + lease_token_hash, + active_attempt_id, + interrupt_generation FROM desktop_tasks WHERE kind = 'monitor' AND monitor_task_id = $1 AND status IN ('queued', 'in_progress') ORDER BY created_at DESC, id DESC LIMIT 1 - `, monitorTaskID).Scan( + `, monitorTaskID).Scan( &item.DesktopTaskID, &item.Status, &item.TargetClientID, &item.LeaseTokenHash, + &item.ActiveAttemptID, &item.InterruptGeneration, ); err != nil { if err == pgx.ErrNoRows { diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 7e4e0b4..0d616ea 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -225,12 +225,12 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( repo := repository.NewDesktopTaskRepository(tx) publishableTasks := make([]*repository.DesktopTask, 0, len(specs)) - fallbackLegacyTaskIDs := make([]int64, 0) + deferredTaskIDs := make([]int64, 0) for _, spec := range specs { 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) + deferredTaskIDs = append(deferredTaskIDs, spec.MonitorTaskID) continue } spec.TargetAccountID = target.AccountID @@ -242,7 +242,7 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( return nil, nil, upsertErr } if fallbackLegacy { - fallbackLegacyTaskIDs = append(fallbackLegacyTaskIDs, spec.MonitorTaskID) + deferredTaskIDs = append(deferredTaskIDs, spec.MonitorTaskID) continue } if shouldPublish && task != nil { @@ -250,10 +250,16 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( } } + if len(deferredTaskIDs) > 0 { + if err := deferPhase2MonitorCollectTasks(ctx, tx, deferredTaskIDs, defaultMonitoringDailyDesktopRetryCooldown); err != nil { + return nil, nil, err + } + } + if err := tx.Commit(ctx); err != nil { return nil, nil, response.ErrInternal(50116, "desktop_task_commit_failed", "failed to commit phase2 monitor desktop dispatch") } - return publishableTasks, fallbackLegacyTaskIDs, nil + return publishableTasks, deferredTaskIDs, nil } func monitorDesktopTaskSpecsNeedTargetLookup(specs []monitorDesktopTaskSpec) bool { @@ -738,6 +744,35 @@ func (s *MonitoringService) fallbackMonitorDesktopTasksToLegacy(ctx context.Cont return nil } +func deferPhase2MonitorCollectTasks(ctx context.Context, tx pgx.Tx, taskIDs []int64, delay time.Duration) error { + if tx == nil || len(taskIDs) == 0 { + return nil + } + if delay <= 0 { + delay = defaultMonitoringDailyDesktopRetryCooldown + } + if _, err := tx.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, + completed_at = NULL, + skip_reason = NULL, + dispatch_after = NOW() + $2::interval, + error_message = NULL, + updated_at = NOW() + WHERE id = ANY($1::bigint[]) + AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks' + AND callback_received_at IS NULL + `, uniqueInt64s(taskIDs), formatPgInterval(delay)); err != nil { + return response.ErrInternal(50041, "task_defer_failed", "failed to defer phase2 monitor tasks until a desktop client is online") + } + return nil +} + func (s *MonitoringService) deferQueuedPhase2MonitorTasks( ctx context.Context, targetClientID uuid.UUID, 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 2578ddc..987993c 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "strings" "testing" + "time" "github.com/google/uuid" "github.com/stretchr/testify/assert" @@ -193,15 +194,37 @@ func TestCompleteLinkedDesktopMonitorTaskOnlyFinalizesInProgress(t *testing.T) { assert.NotContains(t, query, "'unknown'") } -func TestMonitoringSkipOutcomeForRuntimeUnknownFailsTask(t *testing.T) { - outcome := monitoringSkipOutcomeForReason("runtime_unknown") +func TestMonitoringSkipOutcomeForRuntimeUnknownRequeuesAutomaticTask(t *testing.T) { + outcome := monitoringSkipOutcomeForReason("runtime_unknown", 1, "automatic") + + assert.Equal(t, "pending", outcome.TaskStatus) + assert.True(t, outcome.Requeue) + assert.Equal(t, 2*time.Minute, outcome.RetryAfter) + assert.Empty(t, outcome.StoredSkipReason) +} + +func TestMonitoringSkipOutcomeForRuntimeUnknownFailsAfterMaxAttempts(t *testing.T) { + outcome := monitoringSkipOutcomeForReason( + "runtime_unknown", + monitoringRuntimeUnknownMaxDispatchAttempts, + "automatic", + ) assert.Equal(t, "failed", outcome.TaskStatus) + assert.False(t, outcome.Requeue) + assert.Empty(t, outcome.StoredSkipReason) +} + +func TestMonitoringSkipOutcomeForManualRuntimeUnknownFailsTask(t *testing.T) { + outcome := monitoringSkipOutcomeForReason("runtime_unknown", 0, "manual") + + assert.Equal(t, "failed", outcome.TaskStatus) + assert.False(t, outcome.Requeue) assert.Empty(t, outcome.StoredSkipReason) } func TestMonitoringSkipOutcomeKeepsOrdinarySkip(t *testing.T) { - outcome := monitoringSkipOutcomeForReason("stale_business_date") + outcome := monitoringSkipOutcomeForReason("stale_business_date", 0, "automatic") assert.Equal(t, "skipped", outcome.TaskStatus) assert.Equal(t, "stale_business_date", outcome.StoredSkipReason) diff --git a/server/internal/tenant/app/monitoring_service.go b/server/internal/tenant/app/monitoring_service.go index 44734b4..6576987 100644 --- a/server/internal/tenant/app/monitoring_service.go +++ b/server/internal/tenant/app/monitoring_service.go @@ -993,24 +993,21 @@ func (s *MonitoringService) CollectNow( phase2DispatchReady := executionOwner == "desktop_tasks" if len(phase2TaskSpecs) > 0 { phase2TaskCount := len(phase2TaskSpecs) - publishedTasks, fallbackLegacyTaskIDs, dispatchErr := s.dispatchMonitorDesktopTasks(ctx, phase2TaskSpecs) + publishedTasks, deferredTaskIDs, dispatchErr := s.dispatchMonitorDesktopTasks(ctx, phase2TaskSpecs) if dispatchErr != nil { - if fallbackErr := s.fallbackMonitorDesktopTasksToLegacy(ctx, monitorDesktopTaskSpecIDs(phase2TaskSpecs)); fallbackErr != nil { - return nil, fallbackErr - } phase2DispatchReady = false phase2TaskSpecs = nil if s.logger != nil { - s.logger.Warn("phase2 monitor desktop dispatch failed, falling back to legacy execution", + s.logger.Warn("phase2 monitor desktop dispatch failed; pending tasks will retry when a desktop client is online", zap.Error(dispatchErr), zap.Int("phase2_task_count", phase2TaskCount), ) } } else { - if len(fallbackLegacyTaskIDs) > 0 { - if fallbackErr := s.fallbackMonitorDesktopTasksToLegacy(ctx, fallbackLegacyTaskIDs); fallbackErr != nil { - return nil, fallbackErr - } + if len(deferredTaskIDs) > 0 && s.logger != nil { + s.logger.Info("phase2 monitor desktop tasks deferred until a desktop client is online", + zap.Int("deferred_task_count", len(deferredTaskIDs)), + ) } phase2PublishedTasks = publishedTasks }