diff --git a/apps/admin-web/src/i18n/messages/en-US.ts b/apps/admin-web/src/i18n/messages/en-US.ts index c43cebc..12fe394 100644 --- a/apps/admin-web/src/i18n/messages/en-US.ts +++ b/apps/admin-web/src/i18n/messages/en-US.ts @@ -413,6 +413,7 @@ const enUS = { citationRate: "Citation Rate", citations: "Citations", sourcePlatform: "Source", + channelName: "Channel", domain: "Domain", share: "Share", contentName: "Content", diff --git a/apps/admin-web/src/i18n/messages/zh-CN.ts b/apps/admin-web/src/i18n/messages/zh-CN.ts index 8b11f9b..5d9cc0c 100644 --- a/apps/admin-web/src/i18n/messages/zh-CN.ts +++ b/apps/admin-web/src/i18n/messages/zh-CN.ts @@ -413,6 +413,7 @@ const zhCN = { citationRate: "引用率", citations: "引用次数", sourcePlatform: "来源平台", + channelName: "渠道名称", domain: "来源域名", share: "占比", contentName: "内容名称", diff --git a/apps/admin-web/src/lib/api.ts b/apps/admin-web/src/lib/api.ts index 59724db..61e5fb6 100644 --- a/apps/admin-web/src/lib/api.ts +++ b/apps/admin-web/src/lib/api.ts @@ -941,11 +941,14 @@ export const monitoringApi = { { params }, ); }, - collectNow(brandId: number, payload?: { keyword_id?: number | null }) { - return apiClient.post( - `/api/tenant/monitoring/brands/${brandId}/collect-now`, - payload ?? {}, - ); + collectNow( + brandId: number, + payload?: { keyword_id?: number | null; platform_ids?: string[] }, + ) { + return apiClient.post< + MonitoringCollectNowResponse, + { keyword_id?: number | null; platform_ids?: string[] } + >(`/api/tenant/monitoring/brands/${brandId}/collect-now`, payload ?? {}); }, }; diff --git a/apps/admin-web/src/views/TrackingQuestionDetailView.vue b/apps/admin-web/src/views/TrackingQuestionDetailView.vue index 3e171a3..4f70de0 100644 --- a/apps/admin-web/src/views/TrackingQuestionDetailView.vue +++ b/apps/admin-web/src/views/TrackingQuestionDetailView.vue @@ -4,7 +4,6 @@ import { useQuery } from "@tanstack/vue-query"; import dayjs from "dayjs"; import type { MonitoringQuestionCitationAnalysisItem, - MonitoringQuestionContentCitation, MonitoringQuestionDetailCitation, MonitoringQuestionDetailPlatform, } from "@geo/shared-types"; @@ -98,11 +97,6 @@ const activeCitationAnalysis = computed item.ai_platform_id === activePlatformId.value); }); -const activeContentCitations = computed(() => { - const items = detailQuery.data.value?.content_citations ?? []; - return items.filter((item) => item.ai_platform_id === activePlatformId.value); -}); - const latestSampledAt = computed(() => { const platforms = detailQuery.data.value?.platforms ?? []; const sampledAtValues = platforms @@ -126,6 +120,27 @@ const collectedAtDisplay = computed(() => { return sampledAt ? formatDateTime(sampledAt) : "--"; }); +type TrackingAnalyzedContentCitation = { + key: string; + contentName: string; + channelName: string; + citationCount: number; + citationRate: number | null; + citedURL: string | null; +}; + +const sanitizedAnswerText = computed(() => { + return stripAnswerCitationMarkers(activePlatform.value?.answer_text); +}); + +const activeInlineContentCitations = computed(() => { + return analyzeInlineContentCitations(activePlatform.value); +}); + +const shouldShowInlineContentCitations = computed(() => { + return activeInlineContentCitations.value.length > 0; +}); + function goBack(): void { void router.push({ name: "tracking", @@ -168,6 +183,74 @@ function resolveCitationTitle(citation: MonitoringQuestionDetailCitation): strin return citation.cited_title || citation.article_title || citation.cited_url; } +function createQwenSourceGroupPattern(): RegExp { + return /\[\[source_group_web_(\d+)\]\]/g; +} + +function stripAnswerCitationMarkers(answerText: string | null | undefined): string | null { + const normalized = String(answerText ?? "").trim(); + if (!normalized) { + return null; + } + + const cleaned = normalized + .replace(createQwenSourceGroupPattern(), "") + .replace(/[ \t]+([,.;:!?，。！？；：])/g, "$1") + .replace(/[ \t]{2,}/g, " ") + .replace(/\n{3,}/g, "\n\n") + .trim(); + + return cleaned || null; +} + +function analyzeInlineContentCitations( + platform: MonitoringQuestionDetailPlatform | null, +): TrackingAnalyzedContentCitation[] { + const answerText = String(platform?.answer_text ?? ""); + if (!answerText) { + return []; + } + + const matches = Array.from(answerText.matchAll(createQwenSourceGroupPattern())); + if (!matches.length) { + return []; + } + + const citationCounts = new Map(); + for (const match of matches) { + const sourceGroupIndex = Number(match[1]); + if (!Number.isFinite(sourceGroupIndex) || sourceGroupIndex <= 0) { + continue; + } + citationCounts.set(sourceGroupIndex, (citationCounts.get(sourceGroupIndex) ?? 0) + 1); + } + + const totalCitationCount = Array.from(citationCounts.values()).reduce((sum, count) => sum + count, 0); + if (!totalCitationCount) { + return []; + } + + return Array.from(citationCounts.entries()) + .sort((left, right) => { + if (right[1] !== left[1]) { + return right[1] - left[1]; + } + return left[0] - right[0]; + }) + .map(([sourceGroupIndex, citationCount]) => { + const citation = platform?.citations[sourceGroupIndex - 1] ?? null; + + return { + key: `${sourceGroupIndex}-${citation?.cited_url ?? `source_group_web_${sourceGroupIndex}`}`, + contentName: citation ? resolveCitationTitle(citation) : `引用内容 #${sourceGroupIndex}`, + channelName: citation?.site_name ?? "--", + citationCount, + citationRate: citationCount / totalCitationCount, + citedURL: citation?.cited_url ?? null, + }; + }); +} + function normalizeQueryValue(value: unknown): string { const raw = Array.isArray(value) ? value[0] : value; return String(raw ?? "").trim(); @@ -268,8 +351,8 @@ function parsePositiveNumber(value: unknown): number | null {
@@ -314,7 +397,12 @@ function parsePositiveNumber(value: unknown): number | null { -
+
@@ -351,7 +439,7 @@ function parsePositiveNumber(value: unknown): number | null {
-
+
@@ -360,34 +448,40 @@ function parsePositiveNumber(value: unknown): number | null {
-
-
+
+
{{ t("tracking.columns.contentName") }} - {{ t("tracking.columns.publishPlatform") }} + {{ t("tracking.columns.channelName") }} {{ t("tracking.columns.citations") }} {{ t("tracking.columns.share") }}
- {{ item.article_title }} + + {{ item.contentName }} + + {{ item.contentName }}
- {{ item.publish_platform }} + {{ item.channelName }}
- {{ item.citation_count }} + {{ item.citationCount }}
- {{ formatPercent(item.citation_rate) }} + {{ formatPercent(item.citationRate) }}
-
@@ -528,6 +622,10 @@ function parsePositiveNumber(value: unknown): number | null { /* removed min-height, grid-auto-rows handles this universally */ } +.tracking-question-card--span-2 { + grid-column: 1 / -1; +} + .tracking-question-card__header { margin-bottom: 20px; flex-shrink: 0; diff --git a/apps/admin-web/src/views/TrackingView.vue b/apps/admin-web/src/views/TrackingView.vue index 71d9273..9338077 100644 --- a/apps/admin-web/src/views/TrackingView.vue +++ b/apps/admin-web/src/views/TrackingView.vue @@ -261,6 +261,8 @@ const collectNowMutation = useMutation({ return monitoringApi.collectNow(selectedBrandId.value, { keyword_id: selectedKeywordId.value, + platform_ids: + selectedPlatformId.value !== "all" ? [selectedPlatformId.value] : undefined, }); }, onSuccess: (collectNowResult) => { diff --git a/apps/desktop-client/src/main/adapters/doubao.ts b/apps/desktop-client/src/main/adapters/doubao.ts index 44885b9..2408ffe 100644 --- a/apps/desktop-client/src/main/adapters/doubao.ts +++ b/apps/desktop-client/src/main/adapters/doubao.ts @@ -1,10 +1,16 @@ -import type { JsonValue, MonitoringSourceItem } from "@geo/shared-types"; +import { readdirSync, readFileSync, statSync } from "node:fs"; +import { join } from "node:path"; -import { ensureViewLoaded, normalizeText } from "./common"; +import type { JsonValue, MonitoringSourceItem } from "@geo/shared-types"; +import type { Session } from "electron/main"; +import type { Page as PlaywrightPage } from "playwright-core"; + +import { normalizeText } from "./common"; import type { MonitorAdapter } from "./base"; const DOUBAO_APP_ID = "497858"; const DOUBAO_BOT_ID = "7338286299411103781"; +const DOUBAO_HOME_URL = "https://www.doubao.com/"; const DOUBAO_CHAT_URL = "https://www.doubao.com/chat/completion"; const DOUBAO_REFERER = "https://www.doubao.com/chat/"; @@ -63,6 +69,7 @@ type DoubaoStreamEvent = { type DoubaoStreamSummary = { answer: string | null; + reasoning: string | null; provider_request_id: string | null; request_id: string | null; conversation_id: string | null; @@ -72,6 +79,12 @@ type DoubaoStreamSummary = { error_message: string | null; }; +const ANSWER_EVENT_WHITELIST = new Set([ + "STREAM_CHUNK", + "STREAM_MSG_NOTIFY", + "FULL_MSG_NOTIFY", +]); + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -326,13 +339,30 @@ function collectSources( } } -function collectAnswerFragments(payload: unknown, result: string[] = []): string[] { +type AnswerFragment = { + text: string; + kind: "answer" | "reasoning"; +}; + +type AnswerWalkContext = { + inReasoning: boolean; +}; + +function isReasoningIcon(value: unknown): boolean { + return typeof value === "string" && value.includes("Deep_Think"); +} + +function collectAnswerFragments( + payload: unknown, + result: AnswerFragment[] = [], + context: AnswerWalkContext = { inReasoning: false }, +): AnswerFragment[] { if (payload == null) { return result; } if (Array.isArray(payload)) { for (const item of payload) { - collectAnswerFragments(item, result); + collectAnswerFragments(item, result, context); } return result; } @@ -340,19 +370,32 @@ function collectAnswerFragments(payload: unknown, result: string[] = []): string return result; } - const textBlockText = readNestedString(payload, ["text_block", "text"]); - if (textBlockText) { - result.push(textBlockText); + let nextContext = context; + if (typeof payload.block_type === "number" && payload.block_type !== 10000) { + // Non-answer blocks (search queries, status cards, etc.) — skip entirely. + return result; + } + if (isReasoningIcon((payload as { icon_url?: unknown }).icon_url)) { + nextContext = { inReasoning: true }; + } + + const textBlock = payload.text_block; + if (isRecord(textBlock)) { + const text = normalizeOptionalString(textBlock.text); + if (text) { + const reasoning = context.inReasoning || isReasoningIcon(textBlock.icon_url); + result.push({ text, kind: reasoning ? "reasoning" : "answer" }); + } } const deltaText = normalizeOptionalString(payload.delta); if (deltaText) { - result.push(deltaText); + result.push({ text: deltaText, kind: nextContext.inReasoning ? "reasoning" : "answer" }); } const nestedDeltaText = readNestedString(payload, ["delta", "text"]); if (nestedDeltaText) { - result.push(nestedDeltaText); + result.push({ text: nestedDeltaText, kind: nextContext.inReasoning ? "reasoning" : "answer" }); } const assistantText = @@ -360,12 +403,13 @@ function collectAnswerFragments(payload: unknown, result: string[] = []): string normalizeOptionalString(payload.message_type) === "assistant") && normalizeOptionalString(payload.text); if (assistantText) { - result.push(assistantText); + result.push({ text: assistantText, kind: nextContext.inReasoning ? "reasoning" : "answer" }); } - for (const value of Object.values(payload)) { + for (const [key, value] of Object.entries(payload)) { + if (key === "text_block") continue; if (Array.isArray(value) || isRecord(value)) { - collectAnswerFragments(value, result); + collectAnswerFragments(value, result, nextContext); } } @@ -423,6 +467,7 @@ function formatDoubaoStreamError(payload: unknown): string { async function parseDoubaoStream(response: Response): Promise { const summary: DoubaoStreamSummary = { answer: null, + reasoning: null, provider_request_id: null, request_id: null, conversation_id: null, @@ -466,9 +511,15 @@ async function parseDoubaoStream(response: Response): Promise = [ + "fp", + "ms_token", + "device_id", + "web_id", + "tea_uuid", +]; +const DOUBAO_RUNTIME_STATE_TIMEOUT_MS = 15_000; +const DOUBAO_RUNTIME_STATE_POLL_INTERVAL_MS = 500; +const DOUBAO_RUNTIME_STATE_DISK_SCAN_FILE_LIMIT = 8; +const DOUBAO_RUNTIME_STATE_DISK_SCAN_MAX_BYTES = 512_000; +const DOUBAO_XMST_PATTERN = /xmst[\x00-\x20\x7f-\xff]{0,8}([A-Za-z0-9_-]{20,}={0,2})/g; +const DOUBAO_TEA_TOKENS_PATTERN = /__tea_cache_tokens_497858(?:u)?[\x00-\x20\x7f-\xff]{0,8}(\{[^{}]{0,500}\})/g; +const DOUBAO_SAMANTHA_WEB_ID_PATTERN = /samantha_web_web_id[\x00-\x20\x7f-\xff]{0,8}(\{[^{}]{0,500}\})/g; +const DOUBAO_SLARDAR_FLOW_PATTERN = /SLARDARflow_web[\x00-\x20\x7f-\xff]{0,16}(JTdC[A-Za-z0-9+/=_-]{40,}JTdE)/g; + +function parseRuntimeState(value: unknown): { + state: DoubaoRuntimeState | null; + missing: Array; + readError: string | null; +} { if (!isRecord(value)) { - return null; + return { state: null, missing: [...DOUBAO_RUNTIME_STATE_FIELDS], readError: null }; } + const readError = normalizeOptionalString(value.error); const fp = normalizeOptionalString(value.fp); const msToken = normalizeOptionalString(value.ms_token); const deviceID = normalizeOptionalString(value.device_id); const webID = normalizeOptionalString(value.web_id); const teaUUID = normalizeOptionalString(value.tea_uuid); - if (!fp || !msToken || !deviceID || !webID || !teaUUID) { - return null; + + const missing: Array = []; + if (!fp) missing.push("fp"); + if (!msToken) missing.push("ms_token"); + if (!deviceID) missing.push("device_id"); + if (!webID) missing.push("web_id"); + if (!teaUUID) missing.push("tea_uuid"); + + if (missing.length > 0 || !fp || !msToken || !deviceID || !webID || !teaUUID) { + return { state: null, missing, readError }; } return { - fp, - ms_token: msToken, - device_id: deviceID, - web_id: webID, - tea_uuid: teaUUID, + state: { + fp, + ms_token: msToken, + device_id: deviceID, + web_id: webID, + tea_uuid: teaUUID, + }, + missing, + readError, + }; +} + +async function sleep(ms: number, signal?: AbortSignal): Promise { + await new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new Error("adapter_aborted")); + return; + } + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(timer); + reject(new Error("adapter_aborted")); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +async function ensurePageOnDoubaoChat( + page: PlaywrightPage, + signal: AbortSignal, +): Promise { + if (signal.aborted) { + throw new Error("adapter_aborted"); + } + const currentURL = safePlaywrightPageURL(page); + if (!currentURL.startsWith("https://www.doubao.com/")) { + await page.goto(DOUBAO_HOME_URL, { waitUntil: "domcontentloaded" }); + if (signal.aborted) { + throw new Error("adapter_aborted"); + } + } + if (!safePlaywrightPageURL(page).startsWith("https://www.doubao.com/chat")) { + await page.goto(DOUBAO_REFERER, { waitUntil: "domcontentloaded" }); + } + await page.waitForLoadState("domcontentloaded"); + await page.waitForLoadState("networkidle", { timeout: 3_000 }).catch(() => undefined); + if (signal.aborted) { + throw new Error("adapter_aborted"); + } +} + +function safePlaywrightPageURL(page: PlaywrightPage): string { + try { + return page.url(); + } catch { + return ""; + } +} + +const DOUBAO_READ_RUNTIME_STATE_SCRIPT = () => { + try { + const safeParse = (value) => { + if (!value) return null; + try { + return JSON.parse(value); + } catch { + return value; + } + }; + const readStorage = (name) => { + try { + return window.localStorage.getItem(name) || window.sessionStorage.getItem(name); + } catch { + return null; + } + }; + const decodeFlowState = (value) => { + if (typeof value !== "string" || !value.trim()) { + return null; + } + let current = value.trim(); + try { + current = window.atob(current); + } catch { + // Keep original text when it is not base64-encoded. + } + for (let attempt = 0; attempt < 3; attempt += 1) { + try { + const next = decodeURIComponent(current); + if (next === current) { + break; + } + current = next; + } catch { + break; + } + } + return safeParse(current); + }; + const samanthaState = safeParse(window.localStorage.getItem("samantha_web_web_id")); + const teaState = safeParse(window.localStorage.getItem("__tea_cache_tokens_497858")); + const teaSDKState = safeParse(window.localStorage.getItem("__tea_sdk_ab_version_497858")); + const flowState = decodeFlowState(readStorage("SLARDARflow_web")); + const pick = (source, paths) => { + for (const path of paths) { + let current = source; + let ok = true; + for (const key of path) { + if (!current || typeof current !== "object" || !(key in current)) { + ok = false; + break; + } + current = current[key]; + } + if (ok && typeof current === "string" && current.trim()) { + return current.trim(); + } + } + return null; + }; + const readCookie = (name) => { + const segments = document.cookie.split(";"); + for (const segment of segments) { + const [rawName, ...rest] = segment.trim().split("="); + if (rawName === name) { + const joined = rest.join("=").trim(); + return joined || null; + } + } + return null; + }; + return { + fp: readCookie("s_v_web_id"), + ms_token: readStorage("xmst") || readStorage("msToken"), + device_id: + pick(flowState, [["deviceId"], ["device_id"], ["id"]]) || + pick(samanthaState, [["device_id"], ["deviceId"], ["web_id"], ["webId"], ["id"]]), + web_id: + pick(teaState, [["web_id"], ["webId"], ["id"]]) || + pick(samanthaState, [["web_id"], ["webId"], ["device_id"], ["deviceId"], ["id"]]), + tea_uuid: + pick(teaState, [["user_unique_id"], ["userUniqueId"], ["tea_uuid"], ["teaUuid"], ["uuid"], ["web_id"]]) || + pick(teaSDKState, [["uuid"], ["user_unique_id"], ["userUniqueId"]]), + }; + } catch (error) { + return { + error: error instanceof Error ? error.message : "runtime_state_read_failed", + }; + } +}; + +function lastPatternMatch(text: string, pattern: RegExp): string | null { + pattern.lastIndex = 0; + let last: string | null = null; + let matched: RegExpExecArray | null = null; + while ((matched = pattern.exec(text)) !== null) { + const candidate = normalizeOptionalString(matched[1]); + if (candidate) { + last = candidate; + } + } + pattern.lastIndex = 0; + return last; +} + +function readFileTailAsLatin1(filePath: string, maxBytes = DOUBAO_RUNTIME_STATE_DISK_SCAN_MAX_BYTES): string { + const buffer = readFileSync(filePath); + if (buffer.length <= maxBytes) { + return buffer.toString("latin1"); + } + return buffer.subarray(buffer.length - maxBytes).toString("latin1"); +} + +function parseDoubaoRecord(value: string | null): Record | null { + if (!value) { + return null; + } + const parsed = safeParseJSON(value); + return isRecord(parsed) ? parsed : null; +} + +function decodeDoubaoFlowState(value: string | null): Record | null { + const raw = normalizeOptionalString(value); + if (!raw) { + return null; + } + try { + let decoded = Buffer.from(raw, "base64").toString("utf8"); + for (let attempt = 0; attempt < 3; attempt += 1) { + try { + const next = decodeURIComponent(decoded); + if (next === decoded) { + break; + } + decoded = next; + } catch { + break; + } + } + const parsed = safeParseJSON(decoded); + return isRecord(parsed) ? parsed : null; + } catch { + return null; + } +} + +async function flushDoubaoSessionPersistence(session: Session): Promise { + const maybeFlush = (session as Session & { + flushStorageData?: () => Promise | void; + }).flushStorageData; + if (typeof maybeFlush === "function") { + try { + await Promise.resolve(maybeFlush.call(session)); + } catch { + // Ignore persistence flush errors and continue with the best available state. + } + } + try { + await Promise.resolve(session.cookies.flushStore()); + } catch { + // Ignore cookie flush errors and continue with the best available state. + } +} + +async function readDoubaoCookieValue( + session: Session, + name: string, +): Promise { + let cookies: Awaited> = []; + try { + cookies = await Promise.resolve(session.cookies.get({ url: DOUBAO_HOME_URL })); + } catch { + cookies = []; + } + for (const cookie of cookies) { + if (cookie.name !== name) { + continue; + } + const value = normalizeOptionalString(cookie.value); + if (value) { + return value; + } + } + return null; +} + +function readDoubaoPersistedStorageState(storagePath: string | null): Partial { + if (!storagePath) { + return {}; + } + + const leveldbPath = join(storagePath, "Local Storage", "leveldb"); + let candidates: string[] = []; + try { + candidates = readdirSync(leveldbPath) + .map((entry) => join(leveldbPath, entry)) + .filter((entry) => /\.(?:log|ldb)$/i.test(entry)) + .sort((left, right) => statSync(right).mtimeMs - statSync(left).mtimeMs) + .slice(0, DOUBAO_RUNTIME_STATE_DISK_SCAN_FILE_LIMIT); + } catch { + return {}; + } + + let xmst: string | null = null; + let teaState: Record | null = null; + let samanthaState: Record | null = null; + let flowState: Record | null = null; + + for (const filePath of candidates) { + let text = ""; + try { + text = readFileTailAsLatin1(filePath); + } catch { + continue; + } + if (!text) { + continue; + } + xmst ||= lastPatternMatch(text, DOUBAO_XMST_PATTERN); + teaState ||= parseDoubaoRecord(lastPatternMatch(text, DOUBAO_TEA_TOKENS_PATTERN)); + samanthaState ||= parseDoubaoRecord(lastPatternMatch(text, DOUBAO_SAMANTHA_WEB_ID_PATTERN)); + flowState ||= decodeDoubaoFlowState(lastPatternMatch(text, DOUBAO_SLARDAR_FLOW_PATTERN)); + if (xmst && teaState && samanthaState && flowState) { + break; + } + } + + const deviceID = + getDirectString(flowState ?? {}, ["deviceId", "device_id", "id"]) ?? + getDirectString(samanthaState ?? {}, ["device_id", "deviceId", "web_id", "webId", "id"]); + const webID = + getDirectString(teaState ?? {}, ["web_id", "webId", "id"]) ?? + getDirectString(samanthaState ?? {}, ["web_id", "webId", "device_id", "deviceId", "id"]); + const teaUUID = + getDirectString(teaState ?? {}, ["user_unique_id", "userUniqueId", "tea_uuid", "teaUuid", "uuid", "web_id", "webId"]) ?? + webID; + + return { + ms_token: xmst ?? undefined, + device_id: deviceID ?? undefined, + web_id: webID ?? undefined, + tea_uuid: teaUUID ?? undefined, + }; +} + +function mergeDoubaoRuntimeStateSources(...sources: Array | null | undefined>): Record { + const merged: Record = {}; + for (const field of DOUBAO_RUNTIME_STATE_FIELDS) { + for (const source of sources) { + const value = normalizeOptionalString(source?.[field]); + if (value) { + merged[field] = value; + break; + } + } + } + for (const source of sources) { + const error = normalizeOptionalString(source?.error); + if (error) { + merged.error = error; + break; + } + } + return merged; +} + +async function readDoubaoRuntimeStateFallback(session: Session): Promise> { + const [fp, persisted] = await Promise.all([ + readDoubaoCookieValue(session, "s_v_web_id"), + Promise.resolve(readDoubaoPersistedStorageState(session.storagePath)), + ]); + return { + ...(fp ? { fp } : {}), + ...persisted, }; } async function loadDoubaoRuntimeState(context: Parameters[0]): Promise { - if (!context.view) { - throw new Error("doubao_view_required"); + if (!context.playwright) { + throw new Error("doubao_playwright_required"); } - const view = context.view; + const page = context.playwright.page; context.reportProgress("doubao.bootstrap_view"); - await ensureViewLoaded(view, DOUBAO_REFERER, context.signal); + await ensurePageOnDoubaoChat(page, context.signal); + await flushDoubaoSessionPersistence(context.session); context.reportProgress("doubao.read_runtime_state"); - const state = await view.webContents.executeJavaScript( - `(() => { - try { - const safeParse = (value) => { - if (!value) return null; - try { - return JSON.parse(value); - } catch { - return value; - } - }; - const samanthaState = safeParse(window.localStorage.getItem("samantha_web_web_id")); - const teaState = safeParse(window.localStorage.getItem("__tea_cache_tokens_497858")); - const pick = (source, paths) => { - for (const path of paths) { - let current = source; - let ok = true; - for (const key of path) { - if (!current || typeof current !== "object" || !(key in current)) { - ok = false; - break; - } - current = current[key]; - } - if (ok && typeof current === "string" && current.trim()) { - return current.trim(); - } - } - return null; - }; - const readCookie = (name) => { - const segments = document.cookie.split(";"); - for (const segment of segments) { - const [rawName, ...rest] = segment.trim().split("="); - if (rawName === name) { - const joined = rest.join("=").trim(); - return joined || null; - } - } - return null; - }; - return { - fp: readCookie("s_v_web_id"), - ms_token: window.localStorage.getItem("xmst"), - device_id: pick(samanthaState, [["web_id"], ["device_id"], ["deviceId"], ["id"]]), - web_id: pick(teaState, [["web_id"], ["webId"], ["id"]]), - tea_uuid: pick(teaState, [["web_id"], ["tea_uuid"], ["teaUuid"], ["user_unique_id"], ["uuid"]]), - }; - } catch (error) { - return { - error: error instanceof Error ? error.message : "runtime_state_read_failed", - }; - } - })();`, - true, - ); - const runtimeState = parseRuntimeState(state); - if (!runtimeState) { - throw new Error("doubao_runtime_state_missing"); + const deadline = Date.now() + DOUBAO_RUNTIME_STATE_TIMEOUT_MS; + let lastMissing: Array = [...DOUBAO_RUNTIME_STATE_FIELDS]; + let lastReadError: string | null = null; + const persistedFallback = await readDoubaoRuntimeStateFallback(context.session).catch(() => ({})); + + while (true) { + if (context.signal?.aborted) { + throw new Error("adapter_aborted"); + } + + const raw = await page.evaluate(DOUBAO_READ_RUNTIME_STATE_SCRIPT); + const merged = mergeDoubaoRuntimeStateSources(raw, persistedFallback); + const { state, missing, readError } = parseRuntimeState(merged); + if (state) { + return state; + } + lastMissing = missing; + lastReadError = readError; + + if (Date.now() >= deadline) { + break; + } + await sleep(DOUBAO_RUNTIME_STATE_POLL_INTERVAL_MS, context.signal); } - return runtimeState; + + const detail = lastReadError + ? `read_error=${lastReadError}` + : `missing=${lastMissing.join(",") || "unknown"}`; + throw new Error(`doubao_runtime_state_missing:${detail}`); +} + +const DOUBAO_COMPLETION_FETCH_SCRIPT = async ( + { url, body, referer }: { url: string; body: Record; referer: string }, +) => { + try { + const res = await fetch(url, { + method: "POST", + credentials: "include", + referrer: referer, + headers: { + Accept: "text/event-stream", + "Content-Type": "application/json", + }, + body: JSON.stringify(body), + }); + const text = await res.text(); + return { ok: res.ok, status: res.status, body: text }; + } catch (error) { + return { + ok: false, + status: 0, + body: "", + error: error instanceof Error ? error.message : "doubao_completion_fetch_failed", + }; + } +}; + +type DoubaoCompletionResult = { + ok: boolean; + status: number; + body: string; + error?: string; +}; + +async function fetchDoubaoCompletion( + page: PlaywrightPage, + url: string, + body: Record, +): Promise { + let result: unknown; + try { + result = await page.evaluate(DOUBAO_COMPLETION_FETCH_SCRIPT, { + url, + body, + referer: DOUBAO_REFERER, + }); + } catch (error) { + return { + ok: false, + status: 0, + body: "", + error: error instanceof Error ? error.message : "doubao_completion_fetch_eval_failed", + }; + } + + if (!isRecord(result)) { + return { ok: false, status: 0, body: "", error: "doubao_completion_fetch_malformed" }; + } + return { + ok: Boolean(result.ok), + status: typeof result.status === "number" ? result.status : 0, + body: typeof result.body === "string" ? result.body : "", + error: normalizeOptionalString(result.error) ?? undefined, + }; } function extractQuestionText(payload: Record): string { @@ -716,50 +1162,70 @@ function buildAdapterError( export const doubaoAdapter: MonitorAdapter = { provider: "doubao", - executionMode: "view", + executionMode: "playwright", async query(context, payload) { - if (!context.view) { + if (!context.playwright) { return { status: "failed", - summary: "豆包监测缺少页面执行上下文。", + summary: "豆包监测缺少 Playwright 执行上下文。", error: buildAdapterError( - "doubao_view_required", - "doubao_view_required", + "doubao_playwright_required", + "doubao_playwright_required", ), }; } const questionText = extractQuestionText(payload); - const runtimeState = await loadDoubaoRuntimeState(context); + let runtimeState: DoubaoRuntimeState; + try { + runtimeState = await loadDoubaoRuntimeState(context); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (message.startsWith("doubao_runtime_state_missing")) { + const detail = message.slice("doubao_runtime_state_missing:".length) || "unknown"; + return { + status: "failed", + summary: `豆包运行时上下文未就绪（${detail}），请稍后重试或在桌面端打开豆包页面完成初始化。`, + error: buildAdapterError( + "doubao_runtime_state_missing", + `doubao_runtime_state_missing:${detail}`, + ), + }; + } + throw error; + } const requestBody = buildDoubaoRequestBody(questionText, runtimeState); context.reportProgress("doubao.query"); - const response = await context.session.fetch(buildDoubaoRequestUrl(runtimeState), { - method: "POST", - headers: { - Accept: "text/event-stream", - "Content-Type": "application/json", - referer: DOUBAO_REFERER, - }, - body: JSON.stringify(requestBody), - }); + const fetchResult = await fetchDoubaoCompletion( + context.playwright.page, + buildDoubaoRequestUrl(runtimeState), + requestBody, + ); - if (!response.ok) { - const bodyText = await response.text().catch(() => ""); - const detail = normalizeText(bodyText)?.slice(0, 240); + if (fetchResult.error) { return { status: "failed", - summary: detail ? `豆包请求失败：${detail}` : `豆包请求失败（${response.status}）。`, + summary: `豆包请求发送失败：${fetchResult.error}`, + error: buildAdapterError("doubao_request_failed", fetchResult.error), + }; + } + + if (!fetchResult.ok) { + const detail = normalizeText(fetchResult.body)?.slice(0, 240); + return { + status: "failed", + summary: detail ? `豆包请求失败：${detail}` : `豆包请求失败（${fetchResult.status}）。`, error: buildAdapterError( "doubao_request_failed", - detail || `doubao_request_failed_${response.status}`, - { http_status: response.status }, + detail || `doubao_request_failed_${fetchResult.status}`, + { http_status: fetchResult.status }, ), }; } context.reportProgress("doubao.parse_stream"); - const streamSummary = await parseDoubaoStream(response); + const streamSummary = await parseDoubaoStream(new Response(fetchResult.body)); if (streamSummary.error_message) { return { status: "failed", @@ -786,6 +1252,7 @@ export const doubaoAdapter: MonitorAdapter = { request_id: streamSummary.request_id ?? streamSummary.conversation_id, conversation_id: streamSummary.conversation_id, answer: streamSummary.answer, + reasoning: streamSummary.reasoning, citation_count: streamSummary.citations.length, search_result_count: streamSummary.search_results.length, event_count: streamSummary.event_count, diff --git a/apps/desktop-client/src/main/bootstrap.ts b/apps/desktop-client/src/main/bootstrap.ts index 76c3617..b0a8a5d 100644 --- a/apps/desktop-client/src/main/bootstrap.ts +++ b/apps/desktop-client/src/main/bootstrap.ts @@ -1,4 +1,5 @@ import { hostname } from "node:os"; +import { readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { BrowserWindow, WebContentsView, app, ipcMain, nativeTheme } from "electron/main"; @@ -11,6 +12,7 @@ import type { DesktopAccountInfo, DesktopRuntimeSessionSyncRequest } from "@geo/ import { initAccountHealth } from "./account-health"; import { bindPublishAccount, openPublishAccountConsole } from "./account-binder"; +import { installObservedGlobalFetch, registerObservedRequestRendererTarget } from "./network-observer"; import { initProcessMetricsSampler } from "./process-metrics"; import { getPlaywrightCDPPort, startHiddenPlaywrightReaper } from "./playwright-cdp"; import { registerRendererDevtoolsProxyTarget } from "./renderer-devtools-proxy"; @@ -38,6 +40,7 @@ app.commandLine.appendSwitch("disable-quic"); app.commandLine.appendSwitch("disable-background-networking"); app.commandLine.appendSwitch("remote-debugging-port", String(getPlaywrightCDPPort())); app.userAgentFallback = STANDARD_USER_AGENT; +installObservedGlobalFetch(); console.info("[desktop-main] boot", { electron: process.versions.electron, @@ -58,6 +61,107 @@ let mainWindow: ElectronBrowserWindow | null = null; let mainRendererContents: ElectronWebContents | null = null; let quitReleaseInFlight = false; +type WindowMode = "login" | "main"; + +const WINDOW_SIZES: Record = { + login: { width: 340, height: 540, resizable: false }, + main: { width: 1320, height: 860, resizable: true }, +}; + +function windowModePath(): string { + return join(app.getPath("userData"), "window-mode.json"); +} + +function readPersistedWindowMode(): WindowMode { + try { + const raw = readFileSync(windowModePath(), "utf8"); + const data = JSON.parse(raw) as { mode?: unknown }; + if (data?.mode === "main" || data?.mode === "login") { + return data.mode; + } + } catch { + // First run, missing file, or corrupt — fall through to default. + } + return "login"; +} + +function persistWindowMode(mode: WindowMode): void { + try { + writeFileSync(windowModePath(), JSON.stringify({ mode }), "utf8"); + } catch (err) { + console.warn("[desktop-main] persist window mode failed", err); + } +} + +let currentWindowMode: WindowMode = "login"; +let windowModeApplied = false; + +function applyWindowMode(window: ElectronBrowserWindow, mode: WindowMode): void { + const isFirstApply = !windowModeApplied; + const sameMode = windowModeApplied && currentWindowMode === mode; + + if (!sameMode) { + currentWindowMode = mode; + persistWindowMode(mode); + + const target = WINDOW_SIZES[mode]; + window.setResizable(true); + window.setMinimumSize(1, 1); + window.setMaximumSize(0, 0); + window.setSize(target.width, target.height, false); + window.center(); + + if (mode === "login") { + window.setMinimumSize(target.width, target.height); + window.setMaximumSize(target.width, target.height); + window.setResizable(false); + } else { + window.setMinimumSize(800, 600); + window.setResizable(true); + } + } + + windowModeApplied = true; + + if (isFirstApply || !window.isVisible()) { + window.show(); + window.focus(); + } +} + +function currentMainWindow(): ElectronBrowserWindow | null { + if (!mainWindow) { + return null; + } + if (mainWindow.isDestroyed()) { + mainWindow = null; + windowModeApplied = false; + return null; + } + return mainWindow; +} + +async function ensureMainWindow(): Promise { + const existing = currentMainWindow(); + if (existing) { + return existing; + } + const window = await createMainWindow(); + mainWindow = window; + return window; +} + +async function revealMainWindow(): Promise { + const window = await ensureMainWindow(); + applyWindowMode(window, currentWindowMode); +} + +function revealMainWindowSafely(source: string): void { + void revealMainWindow().catch((error) => { + console.error("[desktop-main] reveal main window failed", { source, error }); + }); +} + function rendererURL(): string | null { return process.env.ELECTRON_RENDERER_URL ?? null; } @@ -83,6 +187,7 @@ async function mountRendererView(window: ElectronBrowserWindow): Promise { view.webContents.setWindowOpenHandler(() => ({ action: "deny" })); registerRendererDevtoolsProxyTarget(view.webContents); + registerObservedRequestRendererTarget(view.webContents); mainRendererContents = view.webContents; window.contentView.addChildView(view); syncViewBounds(window, view); @@ -104,12 +209,30 @@ async function mountRendererView(window: ElectronBrowserWindow): Promise { } async function createMainWindow(): Promise { + const initial = WINDOW_SIZES[currentWindowMode]; const window = new BrowserWindow({ - width: 1320, - height: 860, + width: initial.width, + height: initial.height, + show: false, title: "GEO Rankly Desktop", - backgroundColor: nativeTheme.shouldUseDarkColors ? "#15191a" : "#f4f1ea", + titleBarStyle: "hiddenInset", + trafficLightPosition: { x: 12, y: 14 }, + backgroundColor: nativeTheme.shouldUseDarkColors ? "#15191a" : "#eaf2ff", }); + + const fallbackReveal = setTimeout(() => { + if (!windowModeApplied && !window.isDestroyed()) { + applyWindowMode(window, currentWindowMode); + } + }, 1500); + window.once("closed", () => { + clearTimeout(fallbackReveal); + if (mainWindow === window) { + mainWindow = null; + windowModeApplied = false; + } + }); + await mountRendererView(window); return window; } @@ -187,18 +310,23 @@ function registerBridgeHandlers(): void { await releaseRuntimeSession({ revoke: Boolean(revoke) }); return null; }); + ipcMain.handle("desktop:set-window-mode", (_event, mode: WindowMode) => { + const window = currentMainWindow(); + if (window && (mode === "login" || mode === "main")) { + applyWindowMode(window, mode); + } + return null; + }); } if (!initSingleInstance(() => { - if (mainWindow) { - mainWindow.show(); - mainWindow.focus(); - } + revealMainWindowSafely("single-instance"); })) { app.quit(); } app.whenReady().then(async () => { + currentWindowMode = readPersistedWindowMode(); registerBridgeHandlers(); await initSessionRegistry(); initAccountHealth(); @@ -213,25 +341,16 @@ app.whenReady().then(async () => { } mainRendererContents.send("desktop:runtime-invalidated", event); }); - mainWindow = await createMainWindow(); + await ensureMainWindow(); initTray(() => { - if (!mainWindow) { - return; - } - mainWindow.show(); - mainWindow.focus(); + revealMainWindowSafely("tray"); }); }).catch((error) => { console.error("[desktop-main] app.whenReady failed", error); }); -app.on("activate", async () => { - if (mainWindow) { - mainWindow.show(); - mainWindow.focus(); - return; - } - mainWindow = await createMainWindow(); +app.on("activate", () => { + revealMainWindowSafely("activate"); }); app.on("window-all-closed", () => { diff --git a/apps/desktop-client/src/main/execution-devtools.ts b/apps/desktop-client/src/main/execution-devtools.ts new file mode 100644 index 0000000..fa99ae6 --- /dev/null +++ b/apps/desktop-client/src/main/execution-devtools.ts @@ -0,0 +1,57 @@ +import type { WebContents } from "electron/main"; + +function normalizeBooleanEnv(value: string | undefined): boolean | null { + if (!value) { + return null; + } + + const normalized = value.trim().toLowerCase(); + if (["1", "true", "yes", "on"].includes(normalized)) { + return true; + } + if (["0", "false", "no", "off"].includes(normalized)) { + return false; + } + return null; +} + +export function shouldAutoOpenExecutionDevtools(): boolean { + const explicit = normalizeBooleanEnv(process.env.GEO_DESKTOP_EXECUTION_DEVTOOLS); + if (explicit !== null) { + return explicit; + } + + return Boolean(process.env.ELECTRON_RENDERER_URL); +} + +export function shouldAutoOpenBackgroundExecutionDevtools(): boolean { + const explicit = normalizeBooleanEnv(process.env.GEO_DESKTOP_EXECUTION_DEVTOOLS); + if (explicit !== null) { + return explicit; + } + + return false; +} + +export function maybeOpenExecutionDevtools( + webContents: WebContents, + label: string, + options: { background?: boolean } = {}, +): void { + const allowOpen = options.background + ? shouldAutoOpenBackgroundExecutionDevtools() + : shouldAutoOpenExecutionDevtools(); + if (!allowOpen) { + return; + } + if (webContents.isDestroyed() || webContents.isDevToolsOpened()) { + return; + } + + try { + webContents.openDevTools({ mode: "detach" }); + console.info("[desktop-devtools] opened execution devtools", { label }); + } catch (error) { + console.warn("[desktop-devtools] failed to open execution devtools", { label, error }); + } +} diff --git a/apps/desktop-client/src/main/monitor-scheduler.ts b/apps/desktop-client/src/main/monitor-scheduler.ts index af1ae12..7ce3a02 100644 --- a/apps/desktop-client/src/main/monitor-scheduler.ts +++ b/apps/desktop-client/src/main/monitor-scheduler.ts @@ -7,6 +7,7 @@ import type { DesktopTaskEventMessage, DesktopTaskInfo, JsonValue } from "@geo/s export type MonitorSchedulerRouting = "rabbitmq-primary" | "db-recovery"; type MonitorTaskState = "queued" | "leasing" | "active"; +export type MonitorTaskSource = "desktop_task" | "monitoring_lease"; export interface MonitorSchedulerTaskRecord { taskId: string; @@ -15,7 +16,11 @@ export interface MonitorSchedulerTaskRecord { clientId: string; platform: string; routing: MonitorSchedulerRouting; + source: MonitorTaskSource; state: MonitorTaskState; + priority: number; + lane: string | null; + laneWeight: number; title: string | null; businessDate: string | null; questionKey: string | null; @@ -51,6 +56,7 @@ export interface MonitorSchedulerSnapshot { export interface MonitorSchedulerSelection { taskId: string; routing: MonitorSchedulerRouting; + source: MonitorTaskSource; } interface MonitorSchedulerSelectionOptions { @@ -68,7 +74,7 @@ interface MonitorTaskMetadata { questionText: string | null; } -const schedulerStateVersion = 1; +const schedulerStateVersion = 5; const schedulerQuestionCooldownMs = 45_000; const schedulerPlatformCooldownMs = 5_000; const schedulerRestartRecoveryDelayMs = 90_000; @@ -141,7 +147,15 @@ function normalizePersistedState(input: PersistedMonitorSchedulerState | null | return next; } - next.version = typeof input.version === "number" ? input.version : schedulerStateVersion; + // The monitor scheduler is only a local optimization layer. When its on-disk + // shape changes, start from a clean slate and let the server-side queue/resume + // APIs repopulate state. Keeping old records risks reviving already-finished + // monitor desktop_tasks and blocking legacy fallback pulls. + if (input.version !== schedulerStateVersion) { + return next; + } + + next.version = schedulerStateVersion; next.lastHydratedAt = typeof input.lastHydratedAt === "number" ? input.lastHydratedAt : 0; if (input.tasks && typeof input.tasks === "object") { @@ -162,7 +176,11 @@ function normalizePersistedState(input: PersistedMonitorSchedulerState | null | clientId: typeof task.clientId === "string" ? task.clientId : "", platform: task.platform, routing: task.routing === "db-recovery" ? "db-recovery" : "rabbitmq-primary", + source: task.source === "monitoring_lease" ? "monitoring_lease" : "desktop_task", state: task.state === "active" || task.state === "leasing" ? task.state : "queued", + priority: typeof task.priority === "number" ? task.priority : 100, + lane: normalizeOptionalString(task.lane), + laneWeight: typeof task.laneWeight === "number" ? task.laneWeight : laneWeightFromLane(task.lane), title: typeof task.title === "string" ? task.title : null, businessDate: normalizeBusinessDate(task.businessDate), questionKey: normalizeOptionalString(task.questionKey), @@ -234,6 +252,11 @@ export function initMonitorScheduler(): void { const now = Date.now(); const next = clonePersistedState(readPersistedState()); prunePersistedState(next, now); + for (const [taskId, task] of Object.entries(next.tasks)) { + if (task.source === "monitoring_lease") { + delete next.tasks[taskId]; + } + } next.lastHydratedAt = now; writePersistedState(next); } @@ -254,7 +277,11 @@ export function enqueueMonitorTaskFromEvent( clientId: event.target_client_id, platform: event.platform, routing, + source: "desktop_task", state: existing?.state === "active" ? "active" : "queued", + priority: normalizePriority(event.priority, existing?.priority), + lane: normalizeLane(event.lane, existing?.lane), + laneWeight: laneWeightFromLane(event.lane ?? existing?.lane ?? null), title: metadata.title ?? existing?.title ?? null, businessDate: metadata.businessDate ?? existing?.businessDate ?? null, questionKey: metadata.questionKey ?? existing?.questionKey ?? null, @@ -286,7 +313,11 @@ export function noteMonitorTaskLeased( clientId: task.target_client_id, platform: task.platform, routing, + source: "desktop_task", state: "active", + priority: normalizePriority(payloadNumber(task.payload ?? null, "priority"), existing?.priority), + lane: normalizeLane(firstString(task.payload ?? null, ["lane"]), existing?.lane), + laneWeight: laneWeightFromLane(firstString(task.payload ?? null, ["lane"]) ?? existing?.lane ?? null), title: metadata.title ?? existing?.title ?? null, businessDate: metadata.businessDate ?? existing?.businessDate ?? null, questionKey: metadata.questionKey ?? existing?.questionKey ?? null, @@ -302,6 +333,67 @@ export function noteMonitorTaskLeased( }); } +export function enqueueMonitorLeaseTask(input: { + taskId: string; + jobId?: string; + accountId?: string; + clientId: string; + platform: string; + routing: MonitorSchedulerRouting; + priority?: number; + lane?: string | null; + title?: string | null; + businessDate?: string | null; + questionKey?: string | null; + questionText?: string | null; +}): void { + const now = Date.now(); + + mutatePersistedState((draft) => { + const existing = draft.tasks[input.taskId]; + draft.tasks[input.taskId] = { + taskId: input.taskId, + jobId: input.jobId ?? input.taskId, + accountId: input.accountId ?? existing?.accountId ?? "", + clientId: input.clientId, + platform: input.platform, + routing: input.routing, + source: "monitoring_lease", + state: existing?.state === "active" ? "active" : "queued", + priority: normalizePriority(input.priority, existing?.priority), + lane: normalizeLane(input.lane, existing?.lane), + laneWeight: laneWeightFromLane(input.lane ?? existing?.lane ?? null), + title: normalizeOptionalString(input.title) ?? existing?.title ?? null, + businessDate: normalizeBusinessDate(input.businessDate) ?? existing?.businessDate ?? null, + questionKey: normalizeOptionalString(input.questionKey) ?? existing?.questionKey ?? null, + questionText: normalizeOptionalString(input.questionText) ?? existing?.questionText ?? null, + updatedAt: now, + enqueuedAt: existing?.enqueuedAt ?? now, + availableAt: existing?.availableAt ?? now, + lastSeenAt: now, + lastLeaseAttemptAt: existing?.lastLeaseAttemptAt ?? null, + lastStartedAt: existing?.lastStartedAt ?? null, + leaseMisses: existing?.leaseMisses ?? 0, + }; + }); +} + +export function noteMonitorTaskActivated(taskId: string): void { + const now = Date.now(); + + mutatePersistedState((draft) => { + const existing = draft.tasks[taskId]; + if (!existing) { + return; + } + existing.state = "active"; + existing.lastLeaseAttemptAt = now; + existing.lastStartedAt = now; + existing.lastSeenAt = now; + existing.availableAt = now; + }); +} + export function noteMonitorTaskLeaseMiss(taskId: string): void { const now = Date.now(); @@ -363,6 +455,12 @@ export function selectNextMonitorTask( const candidates = Object.values(draft.tasks) .filter((task) => task.state === "queued" && task.availableAt <= now) .sort((left, right) => { + if (left.laneWeight !== right.laneWeight) { + return right.laneWeight - left.laneWeight; + } + if (left.priority !== right.priority) { + return right.priority - left.priority; + } if (left.availableAt !== right.availableAt) { return left.availableAt - right.availableAt; } @@ -377,8 +475,10 @@ export function selectNextMonitorTask( continue; } + const bypassCooldowns = candidate.lane === "high"; + const platformCooldown = draft.platformCooldowns[candidate.platform] ?? 0; - if (platformCooldown > now) { + if (!bypassCooldowns && platformCooldown > now) { continue; } @@ -391,7 +491,7 @@ export function selectNextMonitorTask( continue; } const questionCooldown = draft.questionCooldowns[candidate.questionKey] ?? 0; - if (questionCooldown > now) { + if (!bypassCooldowns && questionCooldown > now) { continue; } } @@ -403,6 +503,7 @@ export function selectNextMonitorTask( return { taskId: candidate.taskId, routing: candidate.routing, + source: candidate.source, }; } @@ -412,7 +513,15 @@ export function selectNextMonitorTask( export function getMonitorSchedulerSnapshot(): MonitorSchedulerSnapshot { const state = readPersistedState(); - const tasks = Object.values(state.tasks).sort((left, right) => left.enqueuedAt - right.enqueuedAt); + const tasks = Object.values(state.tasks).sort((left, right) => { + if (left.laneWeight !== right.laneWeight) { + return right.laneWeight - left.laneWeight; + } + if (left.priority !== right.priority) { + return right.priority - left.priority; + } + return left.enqueuedAt - right.enqueuedAt; + }); return { hydratedAt: state.lastHydratedAt || null, queueDepth: tasks.length, @@ -473,9 +582,12 @@ function metadataFromPayload(payload: Record | null): Monitor } function firstString( - payload: Record, + payload: Record | null, keys: string[], ): string | null { + if (!payload) { + return null; + } for (const key of keys) { const value = payload[key]; if (typeof value === "string") { @@ -492,6 +604,26 @@ function firstString( return null; } +function payloadNumber( + payload: Record | null, + key: string, +): number | null { + if (!payload) { + return null; + } + const value = payload[key]; + if (typeof value === "number" && Number.isFinite(value)) { + return value; + } + if (typeof value === "string" && value.trim()) { + const parsed = Number.parseInt(value.trim(), 10); + if (Number.isFinite(parsed)) { + return parsed; + } + } + return null; +} + function currentMonitoringBusinessDate(now: number): string { const parts = new Intl.DateTimeFormat("en-US", { timeZone: "Asia/Shanghai", @@ -522,6 +654,32 @@ function normalizeOptionalString(value: unknown): string | null { return normalized || null; } +function normalizePriority(value: unknown, fallback = 100): number { + if (typeof value === "number" && Number.isFinite(value)) { + return value; + } + return fallback; +} + +function normalizeLane(value: unknown, fallback: string | null = null): string | null { + return normalizeOptionalString(value) ?? fallback; +} + +function laneWeightFromLane(lane: unknown): number { + switch (normalizeOptionalString(lane)) { + case "high": + return 70; + case "normal_boosted": + return 50; + case "normal": + return 40; + case "retry": + return 20; + default: + return 0; + } +} + function digestText(value: string): string { return createHash("sha1").update(value).digest("hex"); } diff --git a/apps/desktop-client/src/main/network-observer.ts b/apps/desktop-client/src/main/network-observer.ts new file mode 100644 index 0000000..f93a3f5 --- /dev/null +++ b/apps/desktop-client/src/main/network-observer.ts @@ -0,0 +1,402 @@ +import { randomUUID } from "node:crypto"; + +import type { Session, WebContents } from "electron/main"; + +import type { + DesktopObservedRequest, + DesktopObservedRequestKind, + DesktopObservedRequestSnapshot, + DesktopObservedRequestSource, +} from "../shared/network-debug"; + +interface ActiveObservedRequest { + id: string; + startedAt: number; + kind: DesktopObservedRequestKind; + source: DesktopObservedRequestSource; + label: string; + method: string; + url: string; + partition: string | null; + resourceType: string | null; +} + +interface StartObservedRequestOptions { + kind?: DesktopObservedRequestKind; + source?: DesktopObservedRequestSource; + label: string; + method?: string; + url: string; + partition?: string | null; + resourceType?: string | null; +} + +const observedSessions = new WeakSet(); +const activeRequests = new Map(); +const observedRecords: DesktopObservedRequest[] = []; +const maxObservedRecords = 240; +const requestFilter = { + urls: ["http://*/*", "https://*/*", "ws://*/*", "wss://*/*"], +}; +const redactedQueryKeys = /token|signature|sig|auth|key|password|session/i; + +let rendererTarget: WebContents | null = null; +let originalFetch: typeof globalThis.fetch | null = null; + +export function installObservedGlobalFetch(): void { + if (originalFetch || typeof globalThis.fetch !== "function") { + return; + } + + originalFetch = globalThis.fetch.bind(globalThis); + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = requestURL(input); + if (!url || !isInspectableURL(url) || !originalFetch) { + return originalFetch!(input, init); + } + + const method = requestMethod(input, init); + const requestId = startObservedRequest({ + kind: inferKindFromHeaders(input, init), + source: "main", + label: "main.fetch", + method, + url, + }); + + try { + const response = await originalFetch(input, init); + resolveObservedRequest(requestId, { + phase: "response", + status: response.status, + }); + return response; + } catch (error) { + failObservedRequest(requestId, error); + throw error; + } + }) as typeof globalThis.fetch; +} + +export function registerObservedRequestRendererTarget(webContents: WebContents): void { + rendererTarget = webContents; + webContents.once("destroyed", () => { + if (rendererTarget === webContents) { + rendererTarget = null; + } + }); +} + +export function observeSessionRequests( + target: Session, + options: { label: string; partition?: string | null }, +): void { + if (observedSessions.has(target)) { + return; + } + observedSessions.add(target); + + target.webRequest.onBeforeRequest(requestFilter, (details, callback) => { + if (!isInspectableURL(details.url)) { + callback({}); + return; + } + + const requestId = sessionRequestId(details.id, options.partition ?? null); + activeRequests.set(requestId, { + id: requestId, + startedAt: Date.now(), + kind: details.resourceType === "webSocket" ? "ws" : "http", + source: "session", + label: options.label, + method: normalizeMethod(details.method), + url: sanitizeURL(details.url), + partition: options.partition ?? null, + resourceType: normalizeResourceType(details.resourceType), + }); + + pushRecord(buildRecord(activeRequests.get(requestId)!, { + at: Date.now(), + phase: "request", + })); + + callback({}); + }); + + target.webRequest.onCompleted(requestFilter, (details) => { + if (!isInspectableURL(details.url)) { + return; + } + + const requestId = sessionRequestId(details.id, options.partition ?? null); + const active = activeRequests.get(requestId) + ?? createFallbackActiveRequest(requestId, { + source: "session", + label: options.label, + method: normalizeMethod(details.method), + url: details.url, + partition: options.partition ?? null, + resourceType: normalizeResourceType(details.resourceType), + }); + + pushRecord(buildRecord(active, { + at: Date.now(), + phase: "response", + status: details.statusCode, + durationMs: Date.now() - active.startedAt, + })); + activeRequests.delete(requestId); + }); + + target.webRequest.onErrorOccurred(requestFilter, (details) => { + if (!isInspectableURL(details.url)) { + return; + } + + const requestId = sessionRequestId(details.id, options.partition ?? null); + const active = activeRequests.get(requestId) + ?? createFallbackActiveRequest(requestId, { + source: "session", + label: options.label, + method: normalizeMethod(details.method), + url: details.url, + partition: options.partition ?? null, + resourceType: normalizeResourceType(details.resourceType), + }); + + pushRecord(buildRecord(active, { + at: Date.now(), + phase: "error", + durationMs: Date.now() - active.startedAt, + error: normalizeErrorMessage(details.error), + })); + activeRequests.delete(requestId); + }); +} + +export function startObservedRequest(options: StartObservedRequestOptions): string { + const id = `main:${randomUUID()}`; + const active: ActiveObservedRequest = { + id, + startedAt: Date.now(), + kind: options.kind ?? "http", + source: options.source ?? "main", + label: options.label, + method: normalizeMethod(options.method), + url: sanitizeURL(options.url), + partition: options.partition ?? null, + resourceType: options.resourceType ?? null, + }; + + activeRequests.set(id, active); + pushRecord(buildRecord(active, { + at: active.startedAt, + phase: "request", + })); + return id; +} + +export function resolveObservedRequest( + id: string, + options: { + phase?: "response" | "close"; + status?: number | null; + error?: string | null; + keepActive?: boolean; + } = {}, +): void { + const active = activeRequests.get(id); + if (!active) { + return; + } + + pushRecord(buildRecord(active, { + at: Date.now(), + phase: options.phase ?? "response", + status: options.status ?? null, + durationMs: Date.now() - active.startedAt, + error: options.error ?? null, + })); + + if (!options.keepActive) { + activeRequests.delete(id); + } +} + +export function failObservedRequest(id: string, error: unknown): void { + const active = activeRequests.get(id); + if (!active) { + return; + } + + pushRecord(buildRecord(active, { + at: Date.now(), + phase: "error", + durationMs: Date.now() - active.startedAt, + error: normalizeErrorMessage(error), + })); + activeRequests.delete(id); +} + +export function getObservedRequestSnapshot(): DesktopObservedRequestSnapshot { + return { + rendererAttached: Boolean(rendererTarget && !rendererTarget.isDestroyed()), + activeCount: activeRequests.size, + recent: observedRecords.slice(-80), + }; +} + +function requestURL(input: RequestInfo | URL): string | null { + if (typeof input === "string") { + return input; + } + if (input instanceof URL) { + return input.toString(); + } + return typeof input.url === "string" ? input.url : null; +} + +function requestMethod(input: RequestInfo | URL, init?: RequestInit): string { + if (init?.method) { + return normalizeMethod(init.method); + } + if (typeof Request !== "undefined" && input instanceof Request) { + return normalizeMethod(input.method); + } + return "GET"; +} + +function inferKindFromHeaders(input: RequestInfo | URL, init?: RequestInit): DesktopObservedRequestKind { + const headers = new Headers(init?.headers); + if (!headers.has("accept") && typeof Request !== "undefined" && input instanceof Request) { + const requestAccept = input.headers.get("accept"); + if (requestAccept) { + headers.set("accept", requestAccept); + } + } + + const accept = headers.get("accept")?.toLowerCase() ?? ""; + if (accept.includes("text/event-stream")) { + return "sse"; + } + return "http"; +} + +function isInspectableURL(url: string): boolean { + try { + const protocol = new URL(url).protocol; + return protocol === "http:" || protocol === "https:" || protocol === "ws:" || protocol === "wss:"; + } catch { + return false; + } +} + +function sanitizeURL(url: string): string { + try { + const parsed = new URL(url); + for (const key of parsed.searchParams.keys()) { + if (redactedQueryKeys.test(key)) { + parsed.searchParams.set(key, "[REDACTED]"); + } + } + return parsed.toString(); + } catch { + return url; + } +} + +function normalizeMethod(value: string | undefined | null): string { + return typeof value === "string" && value.trim() ? value.trim().toUpperCase() : "GET"; +} + +function normalizeResourceType(value: unknown): string | null { + return typeof value === "string" && value.trim() ? value : null; +} + +function normalizeErrorMessage(error: unknown): string { + if (typeof error === "string" && error.trim()) { + return error; + } + if (error instanceof Error) { + return error.message || error.name || "request_failed"; + } + if (error && typeof error === "object" && "message" in error) { + const message = (error as { message?: unknown }).message; + if (typeof message === "string" && message.trim()) { + return message; + } + } + return "request_failed"; +} + +function sessionRequestId(id: number, partition: string | null): string { + return `session:${partition ?? "default"}:${id}`; +} + +function createFallbackActiveRequest( + id: string, + options: { + source: DesktopObservedRequestSource; + label: string; + method?: string; + url: string; + partition?: string | null; + resourceType?: string | null; + }, +): ActiveObservedRequest { + return { + id, + startedAt: Date.now(), + kind: options.resourceType === "webSocket" ? "ws" : "http", + source: options.source, + label: options.label, + method: normalizeMethod(options.method), + url: sanitizeURL(options.url), + partition: options.partition ?? null, + resourceType: options.resourceType ?? null, + }; +} + +function buildRecord( + active: ActiveObservedRequest, + input: { + at: number; + phase: DesktopObservedRequest["phase"]; + status?: number | null; + durationMs?: number | null; + error?: string | null; + }, +): DesktopObservedRequest { + return { + id: active.id, + at: input.at, + phase: input.phase, + kind: active.kind, + source: active.source, + label: active.label, + method: active.method, + url: active.url, + partition: active.partition, + resourceType: active.resourceType, + status: input.status ?? null, + durationMs: input.durationMs ?? null, + error: input.error ?? null, + }; +} + +function pushRecord(record: DesktopObservedRequest): void { + observedRecords.push(record); + if (observedRecords.length > maxObservedRecords) { + observedRecords.splice(0, observedRecords.length - maxObservedRecords); + } + + if (!rendererTarget || rendererTarget.isDestroyed()) { + return; + } + + try { + rendererTarget.send("desktop:network-observed", record); + } catch { + // Best-effort diagnostics only. + } +} diff --git a/apps/desktop-client/src/main/playwright-cdp.ts b/apps/desktop-client/src/main/playwright-cdp.ts index 9abadab..ba119d5 100644 --- a/apps/desktop-client/src/main/playwright-cdp.ts +++ b/apps/desktop-client/src/main/playwright-cdp.ts @@ -10,6 +10,14 @@ const defaultConnectTimeoutMs = 10_000; const defaultPageTimeoutMs = 45_000; 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`; + +interface CDPTargetDescriptor { + id?: string; + type?: string; + title?: string; + url?: string; +} interface HiddenPlaywrightRecord { key: string; @@ -220,17 +228,22 @@ async function connectBrowserOverCDP(): Promise { connectPromise = (async () => { await waitForCDPEndpoint(); const { chromium } = await import("playwright-core"); - const browser = await chromium.connectOverCDP(cdpEndpointURL, { - timeout: defaultConnectTimeoutMs, - }); + await pruneStaleCDPTargets(); - browser.on("disconnected", () => { - connectedBrowser = null; - connectPromise = null; - }); + try { + return await openBrowserOverCDP(chromium); + } catch (error) { + if (!isConnectOverCDPTimeout(error)) { + throw error; + } - connectedBrowser = browser; - return browser; + const prunedTargetCount = await pruneStaleCDPTargets(); + if (prunedTargetCount <= 0) { + throw error; + } + + return openBrowserOverCDP(chromium); + } })(); try { @@ -260,6 +273,76 @@ async function waitForCDPEndpoint(timeoutMs = defaultConnectTimeoutMs): Promise< throw new Error("playwright_cdp_endpoint_unavailable"); } +async function openBrowserOverCDP( + chromium: typeof import("playwright-core").chromium, +): Promise { + const browser = await chromium.connectOverCDP(cdpEndpointURL, { + timeout: defaultConnectTimeoutMs, + }); + + browser.on("disconnected", () => { + connectedBrowser = null; + connectPromise = null; + }); + + connectedBrowser = browser; + return browser; +} + +async function pruneStaleCDPTargets(): Promise { + let targets: CDPTargetDescriptor[] = []; + try { + const response = await fetch(cdpTargetsListURL); + if (!response.ok) { + return 0; + } + const payload = await response.json(); + if (Array.isArray(payload)) { + targets = payload as CDPTargetDescriptor[]; + } + } catch { + return 0; + } + + let closedCount = 0; + for (const target of targets) { + if (!isStaleCDPTarget(target)) { + continue; + } + + try { + const closeURL = `${cdpEndpointURL}/json/close/${encodeURIComponent(target.id ?? "")}`; + const response = await fetch(closeURL); + if (response.ok) { + closedCount += 1; + } + } catch { + // Ignore best-effort stale target cleanup failures. + } + } + + if (closedCount > 0) { + await wait(150); + } + return closedCount; +} + +function isStaleCDPTarget(target: CDPTargetDescriptor): boolean { + if ((target.type ?? "").trim() !== "page") { + return false; + } + if (!(target.id ?? "").trim()) { + return false; + } + return !(target.title ?? "").trim() && !(target.url ?? "").trim(); +} + +function isConnectOverCDPTimeout(error: unknown): boolean { + return error instanceof Error + && error.name === "TimeoutError" + && error.message.includes("connectOverCDP"); +} + async function waitForNewPage( browser: PlaywrightBrowser, knownPages: Set, @@ -270,6 +353,10 @@ async function waitForNewPage( while (Date.now() - startedAt < defaultPageTimeoutMs) { const nextPage = context.pages().find((page) => !knownPages.has(page)); if (nextPage) { + if (isDevtoolsPage(nextPage)) { + knownPages.add(nextPage); + continue; + } return nextPage; } await wait(100); @@ -298,13 +385,12 @@ function safePageURL(page: PlaywrightPage): string { } } -function closeRecord(record: HiddenPlaywrightRecord): void { - try { - record.page.close().catch(() => undefined); - } catch { - // Ignore and fall through to window close. - } +function isDevtoolsPage(page: PlaywrightPage): boolean { + const url = safePageURL(page); + return url.startsWith("devtools://"); +} +function closeRecord(record: HiddenPlaywrightRecord): void { if (!record.window.isDestroyed()) { record.window.close(); } diff --git a/apps/desktop-client/src/main/renderer-devtools-proxy.ts b/apps/desktop-client/src/main/renderer-devtools-proxy.ts index 94700af..762b1bf 100644 --- a/apps/desktop-client/src/main/renderer-devtools-proxy.ts +++ b/apps/desktop-client/src/main/renderer-devtools-proxy.ts @@ -1,4 +1,5 @@ -import type { WebContents } from "electron/main"; +import { ipcMain } from "electron/main"; +import type { IpcMainEvent, WebContents } from "electron/main"; interface RendererProxyRequest { url: string; @@ -17,6 +18,16 @@ interface RendererProxyResponse { } let rendererWebContents: WebContents | null = null; +let requestSequence = 0; +const pendingRequests = new Map void; + reject: (error: Error) => void; + timeout: ReturnType; +}>(); +const rendererProxyRequestChannel = "desktop:renderer-devtools-proxy:request"; +const rendererProxyResponseChannel = "desktop:renderer-devtools-proxy:response"; + +ipcMain.on(rendererProxyResponseChannel, handleRendererProxyResponse); export function registerRendererDevtoolsProxyTarget(webContents: WebContents): void { rendererWebContents = webContents; @@ -28,7 +39,7 @@ export function registerRendererDevtoolsProxyTarget(webContents: WebContents): v } export function canUseRendererDevtoolsProxy(): boolean { - return Boolean(process.env.ELECTRON_RENDERER_URL && rendererWebContents && !rendererWebContents.isDestroyed()); + return Boolean(rendererWebContents && !rendererWebContents.isDestroyed()); } export async function rendererDevtoolsFetch( @@ -38,37 +49,75 @@ export async function rendererDevtoolsFetch( throw new Error("renderer_devtools_proxy_unavailable"); } - const script = `(async () => { + const requestId = `renderer-proxy-${Date.now()}-${requestSequence += 1}`; + + return await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + pendingRequests.delete(requestId); + reject(new Error("renderer_devtools_proxy_timeout")); + }, 30_000); + + pendingRequests.set(requestId, { + resolve, + reject, + timeout, + }); + try { - const response = await fetch(${JSON.stringify(request.url)}, { - method: ${JSON.stringify(request.method)}, - headers: ${JSON.stringify(request.headers ?? {})}, - body: ${JSON.stringify(request.body ?? null)}, - }); - const bodyText = await response.text(); - return JSON.stringify({ - ok: response.ok, - status: response.status, - statusText: response.statusText, - headers: Object.fromEntries(response.headers.entries()), - bodyText, + rendererWebContents?.send(rendererProxyRequestChannel, { + requestId, + request, }); } catch (error) { - return JSON.stringify({ - ok: false, - status: 0, - statusText: "", - headers: {}, - bodyText: "", - error: String(error && (error.message || error)), - }); + clearTimeout(timeout); + pendingRequests.delete(requestId); + reject(error instanceof Error ? error : new Error("renderer_devtools_proxy_send_failed")); } - })()`; + }); +} - const serialized = await rendererWebContents.executeJavaScript(script, true); - if (typeof serialized !== "string" || !serialized) { - throw new Error("renderer_devtools_proxy_empty_response"); +function handleRendererProxyResponse( + event: IpcMainEvent, + payload: { + requestId?: unknown; + response?: unknown; + }, +): void { + if (rendererWebContents && event.sender !== rendererWebContents) { + return; } - return JSON.parse(serialized) as RendererProxyResponse; + const requestId = typeof payload?.requestId === "string" ? payload.requestId : ""; + if (!requestId) { + return; + } + + const pending = pendingRequests.get(requestId); + if (!pending) { + return; + } + + clearTimeout(pending.timeout); + pendingRequests.delete(requestId); + + const response = payload.response; + if (!isRendererProxyResponse(response)) { + pending.reject(new Error("renderer_devtools_proxy_invalid_response")); + return; + } + + pending.resolve(response); +} + +function isRendererProxyResponse(value: unknown): value is RendererProxyResponse { + if (!value || typeof value !== "object") { + return false; + } + + const candidate = value as Partial; + return typeof candidate.ok === "boolean" + && typeof candidate.status === "number" + && typeof candidate.statusText === "string" + && typeof candidate.bodyText === "string" + && Boolean(candidate.headers && typeof candidate.headers === "object"); } diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index 9fac1a0..f0e69a7 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -5,6 +5,9 @@ import type { DesktopArticleContent, DesktopAccountInfo, DesktopClientInfo, + MonitoringLeaseTask, + MonitoringSourceItem, + MonitoringTaskResultPayload, DesktopRuntimeSessionSyncRequest, DesktopTaskEventMessage, DesktopTaskInfo, @@ -43,8 +46,10 @@ import { releaseHotView, retainHotView } from "./view-pool"; import { getLeaseManagerSnapshot, noteLeaseExtended, noteLeaseReleased, setActiveLease } from "./lease-manager"; import { enqueueMonitorTaskFromEvent, + enqueueMonitorLeaseTask, getMonitorSchedulerSnapshot, initMonitorScheduler, + noteMonitorTaskActivated, noteMonitorTaskCanceled, noteMonitorTaskCompleted, noteMonitorTaskLeaseMiss, @@ -72,23 +77,28 @@ import { } from "./scheduler"; import { clearSessionHandle, createSessionHandle } from "./session-registry"; import { + cancelDesktopTask, completeDesktopTask, + cancelMonitoringTask, configureTransport, deleteDesktopAccount, extendDesktopTask, getDesktopArticleContent, heartbeatDesktopClient, leaseDesktopTask, + leaseMonitoringTasks, listDesktopAccounts, noteTransportHeartbeat, noteTransportPull, offlineDesktopClient, + resumeMonitoringTasks, revokeDesktopClient, setTransportAuthState, - setTransportSseState, + setTransportDispatchState, + skipMonitoringTask, + submitMonitoringTaskResult, upsertDesktopAccount, } from "./transport/api-client"; -import { SseClient, SseClientError } from "./transport/sse-client"; import { DispatchWsClient } from "./transport/dispatch-ws-client"; type RuntimeTone = "info" | "success" | "warn" | "danger"; @@ -100,13 +110,22 @@ type RuntimeTaskStatus = | "unknown" | "aborted"; type RuntimeTaskRouting = "rabbitmq-primary" | "db-recovery"; +type MonitorExecutionPhase = + | "PREPARING" + | "NAVIGATING" + | "AUTHENTICATING" + | "INPUTTING" + | "SUBMITTING" + | "WAITING" + | "PARSING" + | "POSTING"; const heartbeatIntervalMs = 25_000; const pullIntervalMs = 60_000; const leaseExtendIntervalMs = 60_000; const maxActivityItems = 60; const monitorBusinessTimeZone = "Asia/Shanghai"; -const authRequiredMonitorPlatforms = new Set(["yuanbao", "kimi", "deepseek"]); +const authRequiredMonitorPlatforms = new Set(["yuanbao", "kimi", "deepseek", "doubao"]); interface RuntimeTaskRecord { id: string; @@ -141,7 +160,7 @@ export interface RuntimeControllerSnapshot { session: DesktopRuntimeSessionSyncRequest | null; client: DesktopClientInfo | null; running: boolean; - sseConnected: boolean; + dispatchConnected: boolean; queueDepth: number; accounts: DesktopAccountInfo[]; accountProfiles: Record; @@ -159,6 +178,7 @@ export interface RuntimeControllerSnapshot { interface PendingTaskRequest { taskId: string; routing: RuntimeTaskRouting; + source?: "desktop_task" | "monitoring_lease"; } interface ActiveRuntimeExecution { @@ -167,13 +187,17 @@ interface ActiveRuntimeExecution { platform: string; accountId: string; abortController: AbortController; + executionPhase: MonitorExecutionPhase | null; + interruptRequested: boolean; + interruptGeneration: number; + interruptReason: string | null; + lastSafePointAt: number | null; } interface RuntimeState { session: DesktopRuntimeSessionSyncRequest | null; client: DesktopClientInfo | null; running: boolean; - sseConnected: boolean; publishQueue: PendingTaskRequest[]; queuedPublishTaskIds: Set; tasks: Map; @@ -184,7 +208,6 @@ interface RuntimeState { leaseInFlight: boolean; heartbeatTimer: ReturnType | null; pullTimer: ReturnType | null; - sseClient: SseClient | null; dispatchWsClient: DispatchWsClient | null; dispatchWsConnected: boolean; lastHeartbeatAt: number; @@ -201,7 +224,6 @@ const state: RuntimeState = { session: null, client: null, running: false, - sseConnected: false, publishQueue: [], queuedPublishTaskIds: new Set(), tasks: new Map(), @@ -212,7 +234,6 @@ const state: RuntimeState = { leaseInFlight: false, heartbeatTimer: null, pullTimer: null, - sseClient: null, dispatchWsClient: null, dispatchWsConnected: false, lastHeartbeatAt: 0, @@ -325,7 +346,7 @@ export function getRuntimeControllerSnapshot(): RuntimeControllerSnapshot { session: state.session ? { ...state.session } : null, client: state.client ? { ...state.client } : null, running: state.running, - sseConnected: state.sseConnected, + dispatchConnected: state.dispatchWsConnected, queueDepth: localQueueDepth(), accounts: state.accounts.map((item) => ({ ...item })), accountProfiles: Object.fromEntries( @@ -356,21 +377,20 @@ function startRuntime(): void { state.running = true; setSchedulerPhase("running"); setSchedulerError(null); - recordActivity("success", "调度节点已启动", "已启动 SSE、心跳上报和兜底拉取。"); + recordActivity("success", "调度节点已启动", "已启动 WebSocket 分发、心跳上报和兜底拉取。"); - connectEventStream(); connectDispatchWs(); scheduleHeartbeatLoop(); schedulePullLoop(); void sendHeartbeat("startup"); void syncAccounts("startup"); + void resumeLeasedMonitoringTasks("startup"); pumpExecutionLoop(); } function stopRuntime(): void { state.running = false; - state.sseConnected = false; state.leaseInFlight = false; for (const execution of state.activeExecutions.values()) { execution.abortController.abort(); @@ -385,17 +405,13 @@ function stopRuntime(): void { clearInterval(state.pullTimer); state.pullTimer = null; } - if (state.sseClient) { - state.sseClient.stop(); - state.sseClient = null; - } if (state.dispatchWsClient) { state.dispatchWsClient.stop(); state.dispatchWsClient = null; } state.dispatchWsConnected = false; + setTransportDispatchState("idle"); - setTransportSseState("idle"); setSchedulerPhase("idle"); setSchedulerCurrentTask(null); setSchedulerQueueDepth(localQueueDepth()); @@ -448,96 +464,13 @@ function schedulePullLoop(): void { }, pullIntervalMs); } -function connectEventStream(): void { - if (!canRunLive(state.session)) { - return; - } - - state.sseClient?.stop(); - - const sseClient = new SseClient({ - url: buildApiUrl(state.session.api_base_url, "/api/desktop/events"), - headers: { - Authorization: `Bearer ${state.session.client_token}`, - }, - }); - - sseClient.on("state", (payload) => { - if (payload === "streaming") { - setTransportSseState("streaming"); - return; - } - if (payload === "connecting") { - setTransportSseState("connecting"); - return; - } - setTransportSseState("idle"); - }); - - sseClient.on("connected", (payload) => { - state.sseConnected = true; - noteSchedulerSseEvent(); - setSchedulerError(null); - - if (isConnectedEvent(payload)) { - state.lastServerTime = payload.server_time; - } - - recordActivity("success", "单点事件流已连接", "服务端已建立定向任务事件流。"); - }); - - sseClient.on("ping", () => { - state.sseConnected = true; - noteSchedulerSseEvent(); - }); - - const taskEventHandler = (payload: unknown) => { - if (!isDesktopTaskEvent(payload)) { - return; - } - state.sseConnected = true; - noteSchedulerSseEvent(); - handleTaskEvent(payload); - }; - - sseClient.on("task_available", taskEventHandler); - sseClient.on("task_leased", taskEventHandler); - sseClient.on("task_extended", taskEventHandler); - sseClient.on("task_completed", taskEventHandler); - sseClient.on("task_canceled", taskEventHandler); - sseClient.on("task_reconciled", taskEventHandler); - - sseClient.on("error", (payload) => { - state.sseConnected = false; - - if (payload instanceof SseClientError && payload.status === 401) { - handleAuthExpired(payload); - return; - } - - const message = errorMessage(payload); - state.lastError = message; - setSchedulerError(message); - recordActivity("warn", "单点事件流重连中", message); - }); - - sseClient.on("reconnect", (payload) => { - if (!isReconnectPayload(payload)) { - return; - } - recordActivity("info", "单点事件流退避等待", `将在 ${payload.delayMs}ms 后重连 SSE。`); - }); - - state.sseClient = sseClient; - sseClient.start(); -} - function connectDispatchWs(): void { if (!canRunLive(state.session)) { return; } state.dispatchWsClient?.stop(); + setTransportDispatchState("connecting"); const wsUrl = buildApiUrl(state.session.api_base_url, "/api/desktop/dispatch") .replace(/^http:\/\//, "ws://") @@ -549,39 +482,73 @@ function connectDispatchWs(): void { Authorization: `Bearer ${state.session.client_token}`, }, }); + let hasConnectedOnce = false; client.on("open", () => { state.dispatchWsConnected = true; + setTransportDispatchState("streaming"); setSchedulerError(null); - recordActivity("success", "任务分发通道已连接", "AMQP 直推 WebSocket 已就绪,发布任务将以低延迟到达。"); + recordActivity("success", "任务分发通道已连接", "统一 WebSocket 分发通道已就绪。"); + void pullNextTask("monitor"); + if (hasConnectedOnce) { + void resumeLeasedMonitoringTasks("reconnect"); + } + hasConnectedOnce = true; }); - client.on("connected", () => { + client.on("connected", (payload) => { state.dispatchWsConnected = true; + setTransportDispatchState("streaming"); + if (isConnectedEvent(payload)) { + state.lastServerTime = payload.server_time; + } }); - client.on("task_available", (payload) => { + const dispatchTaskHandler = (payload: unknown) => { + state.dispatchWsConnected = true; + noteSchedulerSseEvent(); + + if (isMonitoringTaskControlEvent(payload)) { + void handleMonitoringTaskControl(payload); + return; + } + if (isMonitoringDispatchSignal(payload)) { + void handleMonitoringDispatchSignal(payload); + return; + } if (!isDesktopTaskEvent(payload)) { return; } - state.dispatchWsConnected = true; handleTaskEvent(payload); - }); + }; + + client.on("task_available", dispatchTaskHandler); + client.on("task_leased", dispatchTaskHandler); + client.on("task_extended", dispatchTaskHandler); + client.on("task_completed", dispatchTaskHandler); + client.on("task_canceled", dispatchTaskHandler); + client.on("task_reconciled", dispatchTaskHandler); + client.on("task_control", dispatchTaskHandler); client.on("error", (payload) => { state.dispatchWsConnected = false; + setTransportDispatchState("connecting"); const message = errorMessage(payload); + state.lastError = message; + setSchedulerError(message); recordActivity("warn", "任务分发通道异常", message); }); client.on("close", () => { state.dispatchWsConnected = false; + setTransportDispatchState(state.running ? "connecting" : "idle"); }); client.on("reconnect", (payload) => { if (!isReconnectPayload(payload)) { return; } + setTransportDispatchState("connecting"); recordActivity("info", "任务分发通道退避等待", `将在 ${payload.delayMs}ms 后重连 WebSocket。`); }); @@ -590,7 +557,16 @@ function connectDispatchWs(): void { } function handleTaskEvent(event: DesktopTaskEventMessage): void { + if (event.type === "task_control" || event.signal_only || !event.task_id.trim()) { + return; + } + const existing = state.tasks.get(event.task_id); + const eventUpdatedAt = parseTimestamp(event.updated_at) ?? Date.now(); + if (existing && eventUpdatedAt < existing.updatedAt) { + return; + } + const next: RuntimeTaskRecord = { id: event.task_id, jobId: event.job_id, @@ -603,7 +579,7 @@ function handleTaskEvent(event: DesktopTaskEventMessage): void { status: event.status, routing: existing?.routing ?? "rabbitmq-primary", leaseExpiresAt: event.status === "in_progress" ? existing?.leaseExpiresAt ?? null : null, - updatedAt: parseTimestamp(event.updated_at) ?? Date.now(), + updatedAt: eventUpdatedAt, summary: summaryFromEventType(event.type, event.status), payload: existing?.payload ?? {}, error: existing?.error ?? null, @@ -631,6 +607,132 @@ function handleTaskEvent(event: DesktopTaskEventMessage): void { } } +async function handleMonitoringDispatchSignal(event: DesktopTaskEventMessage): Promise { + if (event.target_client_id !== currentClientID()) { + return; + } + + if ((event.lane ?? "").trim() === "high") { + recordActivity("info", "收到监控高优先级信号", "客户端将按平台隔离和并发上限，优先领取 collect-now 桌面监控任务。"); + } + + if (event.task_id.trim()) { + enqueueMonitorTaskFromEvent(event, "rabbitmq-primary"); + pumpExecutionLoop(); + return; + } + + await pullNextTask("monitor"); +} + +async function handleMonitoringTaskControl(event: DesktopTaskEventMessage): Promise { + if ( + event.target_client_id !== currentClientID() + || event.control !== "interrupt_requested" + || !event.task_id.trim() + ) { + return; + } + + const taskId = event.task_id.trim(); + const generation = toInterruptGeneration(event.interrupt_generation); + const existing = state.tasks.get(taskId) ?? null; + if (existing && generation < toInterruptGeneration(existing.payload.interrupt_generation)) { + return; + } + + const active = state.activeExecutions.get(taskId); + if (active?.kind === "monitor") { + if (isDesktopTaskID(taskId)) { + if (existing) { + existing.summary = "已收到立即收集请求；当前执行继续保留，high lane 任务会按平台隔离和并发上限调度。"; + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + } + recordActivity( + "info", + "监控任务保留当前执行", + `${existing?.title ?? taskId} 当前执行继续保留；若 high lane 属于其他平台可使用第二个窗口并行，否则等待当前平台释放。`, + ); + return; + } + + if (existing) { + existing.summary = "已收到抢占请求；当前监控执行将自然完成，随后优先处理 high lane 任务。"; + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + } + recordActivity("warn", "监控任务收到抢占请求", `${existing?.title ?? taskId} 当前已在执行，Phase 1 将等待自然完成。`); + return; + } + + if (existing) { + existing.payload = { + ...existing.payload, + interrupt_requested: true, + interrupt_generation: generation, + interrupt_reason: event.reason?.trim() || "collect_now_preempt", + replacement_task_id: event.replacement_task_id?.trim() || null, + }; + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + } + + if (!existing?.leaseToken) { + noteMonitorTaskCanceled(taskId); + if (existing) { + existing.status = "aborted"; + existing.summary = "已收到抢占请求，本地排队中的旧监控任务已让路。"; + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + } + await pullMonitoringTasks("rabbitmq-primary"); + return; + } + + const monitoringTaskID = Number.parseInt(taskId, 10); + if (!Number.isFinite(monitoringTaskID)) { + noteMonitorTaskCanceled(taskId); + if (existing) { + existing.status = "aborted"; + existing.leaseToken = null; + existing.summary = "已收到抢占请求，但任务不是 legacy monitoring lease，已从本地调度队列移除。"; + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + } + await pullMonitoringTasks("rabbitmq-primary"); + return; + } + + try { + await cancelMonitoringTask(monitoringTaskID, { + lease_token: existing.leaseToken, + reason: event.reason?.trim() || "collect_now_preempt", + }); + noteMonitorTaskCanceled(taskId); + existing.status = "aborted"; + existing.leaseToken = null; + existing.summary = "已响应抢占请求，释放旧监控租约并为高优任务让路。"; + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + recordActivity("info", "已取消旧监控租约", `${existing.title} 已从本地排队队列中移除。`); + await pullMonitoringTasks("rabbitmq-primary"); + } catch (error) { + if (isApiClientError(error, 401)) { + handleAuthExpired(error); + return; + } + + const message = errorMessage(error); + if (existing) { + existing.summary = `收到抢占请求，但取消租约失败：${message}`; + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + } + recordActivity("warn", "监控租约抢占失败", `${existing?.title ?? taskId} 取消租约失败：${message}`); + } +} + async function sendHeartbeat(source: "startup" | "timer"): Promise { if (!canRunLive(state.session)) { return; @@ -829,6 +931,7 @@ async function syncAccounts(source: "startup" | "manual"): Promise { } refreshAccountNames(); + pumpExecutionLoop(); void sendHeartbeat("timer"); } catch (error) { const message = errorMessage(error); @@ -947,12 +1050,6 @@ function pumpExecutionLoop(): void { } return; } - - if (canStartAnotherMonitorExecution()) { - // Monitor has not been migrated to the dispatch channel yet, so it still - // relies on the periodic pull as its primary signal. - void pullNextTask("monitor"); - } } async function leaseSpecificTask( @@ -966,6 +1063,12 @@ async function leaseSpecificTask( state.leaseInFlight = true; try { + if (expectedKind === "monitor" && request.source === "monitoring_lease") { + state.leaseInFlight = false; + await executeLeasedMonitoringTask(request.taskId, request.routing); + return; + } + const leased = await leaseDesktopTask({ task_id: request.taskId }); if (!leased.task) { if (expectedKind === "monitor") { @@ -977,7 +1080,19 @@ async function leaseSpecificTask( await executeLeasedTask(leased, request.routing); } catch (error) { if (expectedKind === "monitor") { - noteMonitorTaskLeaseMiss(request.taskId); + if (error instanceof ApiClientError && error.code === 40991) { + noteMonitorTaskCanceled(request.taskId); + const existing = state.tasks.get(request.taskId); + if (existing) { + existing.status = "aborted"; + existing.leaseToken = null; + existing.summary = "本地缓存的监控任务已不在服务端 queued，已从调度队列移除。"; + existing.updatedAt = Date.now(); + state.tasks.set(request.taskId, existing); + } + } else { + noteMonitorTaskLeaseMiss(request.taskId); + } } if (isApiClientError(error, 401)) { @@ -1000,12 +1115,49 @@ async function leaseSpecificTask( } async function pullNextTask(kind: "publish" | "monitor"): Promise { + if (kind === "monitor") { + if (!canRunLive(state.session) || state.leaseInFlight) { + return; + } + + state.leaseInFlight = true; + try { + const leased = await leaseDesktopTask({ kind: "monitor" }); + state.lastPullAt = Date.now(); + state.lastPullStatus = "success"; + noteTransportPull(true); + noteSchedulerPull(); + + if (leased.task) { + state.leaseInFlight = false; + await executeLeasedTask(leased, "rabbitmq-primary"); + return; + } + } catch (error) { + state.lastPullAt = Date.now(); + state.lastPullStatus = "failed"; + state.lastError = errorMessage(error); + noteTransportPull(false); + setSchedulerError(state.lastError); + + if (isApiClientError(error, 401)) { + handleAuthExpired(error); + return; + } + } finally { + state.leaseInFlight = false; + syncSchedulerSurface(); + } + + await pullMonitoringTasks("db-recovery"); + return; + } + if (!canRunLive(state.session) || state.leaseInFlight) { return; } state.leaseInFlight = true; - let leasedTask = false; try { const leased = await leaseDesktopTask({ kind }); @@ -1018,7 +1170,6 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise { return; } - leasedTask = true; state.leaseInFlight = false; await executeLeasedTask(leased, "db-recovery"); } catch (error) { @@ -1038,11 +1189,709 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise { } finally { state.leaseInFlight = false; syncSchedulerSurface(); + } +} - if (!leasedTask && kind === "publish" && canStartAnotherMonitorExecution()) { - void pullNextTask("monitor"); +async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise { + const leaseLimit = resolveLegacyMonitoringLeaseLimit(); + if (!canRunLive(state.session) || state.leaseInFlight || leaseLimit <= 0) { + return; + } + + state.leaseInFlight = true; + + try { + const leased = await leaseMonitoringTasks({ + limit: leaseLimit, + }); + state.lastPullAt = Date.now(); + state.lastPullStatus = "success"; + noteTransportPull(true); + noteSchedulerPull(); + + enqueueLeasedMonitoringTasks(leased.tasks, routing); + } catch (error) { + state.lastPullAt = Date.now(); + state.lastPullStatus = "failed"; + state.lastError = errorMessage(error); + noteTransportPull(false); + setSchedulerError(state.lastError); + + if (isApiClientError(error, 401)) { + handleAuthExpired(error); + return; + } + + recordActivity("warn", "监控任务拉取失败", state.lastError); + } finally { + state.leaseInFlight = false; + syncSchedulerSurface(); + } +} + +async function resumeLeasedMonitoringTasks(source: "startup" | "reconnect"): Promise { + if (!canRunLive(state.session)) { + return; + } + + try { + const resumed = await resumeMonitoringTasks({}); + enqueueLeasedMonitoringTasks(resumed.tasks, "db-recovery"); + if (resumed.tasks.length > 0) { + recordActivity("info", "已恢复监控租约", `从服务端恢复了 ${resumed.tasks.length} 条未完成监控任务（${source}）。`); + } + void pullNextTask("monitor"); + } catch (error) { + if (isApiClientError(error, 401)) { + handleAuthExpired(error); + return; + } + recordActivity("warn", "恢复监控租约失败", errorMessage(error)); + } +} + +function enqueueLeasedMonitoringTasks( + tasks: MonitoringLeaseTask[], + routing: RuntimeTaskRouting, +): void { + if (tasks.length === 0) { + return; + } + + for (const task of tasks) { + const taskId = String(task.task_id); + const accountId = resolveMonitoringAccountID(task.ai_platform_id); + const now = Date.now(); + state.tasks.set(taskId, { + id: taskId, + jobId: taskId, + title: buildMonitoringTaskTitle(task), + kind: "monitor", + platform: task.ai_platform_id, + accountId, + accountName: accountId ? resolveAccountName(accountId, task.platform_name) : task.platform_name, + clientId: state.client?.id ?? taskId, + status: "queued", + routing, + leaseExpiresAt: parseTimestamp(task.lease_expires_at), + updatedAt: now, + summary: `已领取监控租约，等待本地调度执行（${describeRouting(routing)}）。`, + payload: { + title: buildMonitoringTaskTitle(task), + question_id: task.question_id, + question_hash: task.question_hash, + question_text: task.question_text, + ai_platform_id: task.ai_platform_id, + platform_name: task.platform_name, + collector_type: task.collector_type, + run_mode: task.run_mode, + trigger_source: task.trigger_source, + business_date: task.business_date, + priority: typeof task.priority === "number" ? task.priority : null, + lane: typeof task.lane === "string" ? task.lane : null, + monitoring_task_id: task.task_id, + lease_token: task.lease_token, + lease_expires_at: task.lease_expires_at, + scheduler_group_key: task.question_hash || `qid:${task.question_id}`, + legacy_monitoring_task: true, + }, + error: null, + result: null, + attemptId: null, + leaseToken: task.lease_token, + }); + enqueueMonitorLeaseTask({ + taskId, + jobId: taskId, + accountId, + clientId: state.client?.id ?? taskId, + platform: task.ai_platform_id, + routing, + priority: typeof task.priority === "number" + ? task.priority + : task.trigger_source === "manual" + ? 5000 + : 100, + lane: typeof task.lane === "string" + ? task.lane + : task.trigger_source === "manual" + ? "high" + : "normal", + title: buildMonitoringTaskTitle(task), + businessDate: task.business_date, + questionKey: task.question_hash || `qid:${task.question_id}`, + questionText: task.question_text, + }); + } + + syncSchedulerSurface(); + pumpExecutionLoop(); +} + +async function executeLeasedMonitoringTask( + taskId: string, + routing: RuntimeTaskRouting, +): Promise { + const taskRecord = state.tasks.get(taskId); + if (!taskRecord || taskRecord.kind !== "monitor" || !taskRecord.leaseToken) { + noteMonitorTaskLeaseMiss(taskId); + return; + } + + const monitoringTaskID = Number.parseInt(taskId, 10); + if (!Number.isFinite(monitoringTaskID)) { + noteMonitorTaskCanceled(taskId); + state.tasks.delete(taskId); + return; + } + + setSchedulerPhase("running"); + setActiveLease({ + taskId, + attemptId: null, + leaseExpiresAt: isoFromTimestamp(taskRecord.leaseExpiresAt), + }); + + recordActivity("info", "监控任务开始执行", `${taskRecord.title} 已进入本地执行队列。`); + + const abortController = new AbortController(); + state.activeExecutions.set(taskRecord.id, { + taskId: taskRecord.id, + kind: "monitor", + platform: taskRecord.platform, + accountId: taskRecord.accountId, + abortController, + executionPhase: "PREPARING", + interruptRequested: Boolean(taskRecord.payload.interrupt_requested), + interruptGeneration: toInterruptGeneration(taskRecord.payload.interrupt_generation), + interruptReason: interruptReasonFromTask(taskRecord), + lastSafePointAt: Date.now(), + }); + noteMonitorTaskActivated(taskId); + syncSchedulerSurface(); + + try { + const execution = await executeTaskAdapter(taskRecord, abortController.signal); + if (execution.status === "unknown") { + const skipReason = monitoringSkipReasonFromExecution(execution); + await skipMonitoringTask(monitoringTaskID, { + lease_token: taskRecord.leaseToken, + skip_reason: skipReason, + error_message: execution.summary, + completed_at: new Date().toISOString(), + }); + const existing = state.tasks.get(taskId); + if (existing) { + existing.status = "aborted"; + existing.summary = execution.summary; + existing.error = execution.error ?? null; + existing.updatedAt = Date.now(); + existing.leaseToken = null; + state.tasks.set(taskId, existing); + } + noteLeaseReleased("completed", taskId); + noteMonitorTaskCompleted(taskId); + recordActivity( + skipReason === "stale_business_date" ? "info" : "warn", + skipReason === "stale_business_date" ? "监控任务已按跨日策略丢弃" : "监控任务结果待确认", + `${taskRecord.title} ${execution.summary}`, + ); + return; + } + + await submitMonitoringTaskResult(monitoringTaskID, buildMonitoringResultPayload(taskRecord.leaseToken, execution)); + const existing = state.tasks.get(taskId); + if (existing) { + existing.status = execution.status; + existing.summary = execution.summary; + existing.result = execution.payload ?? null; + existing.error = execution.error ?? null; + existing.updatedAt = Date.now(); + existing.leaseToken = null; + state.tasks.set(taskId, existing); + } + noteLeaseReleased(execution.status === "failed" ? "failed" : "completed", taskId); + noteMonitorTaskCompleted(taskId); + recordActivity( + execution.status === "failed" ? "danger" : "success", + execution.status === "failed" ? "监控任务执行失败" : "监控任务执行成功", + `${taskRecord.title} ${execution.summary}`, + ); + } catch (error) { + if (isMonitorAbortReason(taskRecord, error, "stale_business_date")) { + let dropSummary = staleMonitoringDropSummary(taskRecord); + try { + await skipMonitoringTask(monitoringTaskID, { + lease_token: taskRecord.leaseToken, + skip_reason: "stale_business_date", + error_message: dropSummary, + completed_at: new Date().toISOString(), + }); + } catch (skipError) { + dropSummary = `${dropSummary}（skip 回写失败：${errorMessage(skipError)}）`; + } + + const existing = state.tasks.get(taskId); + if (existing) { + existing.status = "aborted"; + existing.summary = dropSummary; + existing.error = { + ...toStructuredError(error), + dropped_reason: "stale_business_date", + }; + existing.updatedAt = Date.now(); + existing.leaseToken = null; + state.tasks.set(taskId, existing); + } + noteLeaseReleased("completed", taskId); + noteMonitorTaskCompleted(taskId); + recordActivity("info", "监控任务已按跨日策略丢弃", dropSummary); + return; + } + + const message = errorMessage(error); + const existing = state.tasks.get(taskId); + if (existing) { + existing.status = "failed"; + existing.summary = message; + existing.error = toStructuredError(error); + existing.updatedAt = Date.now(); + existing.leaseToken = null; + state.tasks.set(taskId, existing); + } + noteLeaseReleased("failed", taskId); + noteMonitorTaskCompleted(taskId); + recordActivity("danger", "监控任务执行失败", `${taskRecord.title} 执行失败：${message}`); + } finally { + state.activeExecutions.delete(taskId); + syncSchedulerSurface(); + noteSchedulerTaskFinished(); + pumpExecutionLoop(); + } +} + +function buildMonitoringTaskTitle(task: MonitoringLeaseTask): string { + const base = task.question_text.trim(); + if (!base) { + return `监控任务 · ${task.platform_name}`; + } + return `${task.platform_name} · ${base}`; +} + +function resolveMonitoringAccountID(platform: string): string { + const live = state.accounts.find((account) => + account.platform === platform + && getProjectedAccountHealth({ + accountId: account.id, + platform: account.platform, + health: account.health, + verifiedAt: account.verified_at, + }).health === "live", + ); + if (live) { + return live.id; + } + return state.accounts.find((account) => account.platform === platform)?.id ?? ""; +} + +function buildMonitoringResultPayload( + leaseToken: string, + execution: AdapterExecutionResult, +): MonitoringTaskResultPayload { + const payload = execution.payload ?? {}; + return { + lease_token: leaseToken, + status: execution.status === "failed" ? "failed" : "succeeded", + provider_model: asOptionalString(payload.provider_model), + provider_request_id: asOptionalString(payload.provider_request_id), + request_id: asOptionalString(payload.request_id), + answer: asOptionalString(payload.answer), + raw_response_json: asRecord(payload.raw_response_json), + citations: asSourceItems(payload.citations), + search_results: asSourceItems(payload.search_results), + brand_mentioned: asOptionalBoolean(payload.brand_mentioned), + brand_mention_position: asOptionalString(payload.brand_mention_position), + first_recommended: asOptionalBoolean(payload.first_recommended), + sentiment_label: asOptionalString(payload.sentiment_label), + matched_brand_terms: asStringArray(payload.matched_brand_terms), + error_message: execution.status === "failed" ? execution.summary : undefined, + completed_at: new Date().toISOString(), + }; +} + +function buildMonitoringFailurePayload( + leaseToken: string, + failureMessage: string, + structuredError: Record, +): MonitoringTaskResultPayload { + return { + lease_token: leaseToken, + status: "failed", + raw_response_json: { + runtime_error: structuredError, + }, + error_message: failureMessage, + completed_at: new Date().toISOString(), + }; +} + +async function executeLeasedDesktopMonitorTask( + taskRecord: RuntimeTaskRecord, + leased: LeaseDesktopTaskResponse, + routing: RuntimeTaskRouting, + signal: AbortSignal, +): Promise { + const monitoringTaskID = extractMonitoringTaskID(taskRecord.payload); + if (!monitoringTaskID) { + const missingMessage = "monitor desktop task is missing monitoring_task_id"; + const completed = await completeDesktopTask(taskRecord.id, { + lease_token: leased.lease_token as string, + status: "failed", + error: { + code: "monitoring_task_id_missing", + message: missingMessage, + }, + }); + upsertTaskFromInfo(completed, { + routing, + summary: missingMessage, + attemptId: null, + leaseToken: null, + }); + noteLeaseReleased("failed", taskRecord.id); + noteMonitorTaskCompleted(taskRecord.id); + recordActivity("danger", "监控任务执行失败", `${taskRecord.title} 缺少 monitoring_task_id，无法回写结果。`); + return; + } + + try { + const execution = await executeTaskAdapter(taskRecord, signal); + if (execution.status === "unknown") { + const skipReason = monitoringSkipReasonFromExecution(execution); + await skipMonitoringTask(monitoringTaskID, { + lease_token: leased.lease_token as string, + skip_reason: skipReason, + error_message: execution.summary, + completed_at: new Date().toISOString(), + }); + if (skipReason === "stale_business_date") { + const canceled = await cancelDesktopTask(taskRecord.id, { + lease_token: leased.lease_token as string, + reason: skipReason, + }).catch((cancelError) => { + const existing = state.tasks.get(taskRecord.id); + if (existing) { + existing.status = "aborted"; + existing.error = { + ...(execution.error ?? {}), + cancel_error: errorMessage(cancelError), + dropped_reason: "stale_business_date", + }; + existing.summary = `${execution.summary}（desktop task 取消失败：${errorMessage(cancelError)}）`; + existing.updatedAt = Date.now(); + existing.leaseToken = null; + state.tasks.set(taskRecord.id, existing); + } + return null; + }); + if (canceled) { + upsertTaskFromInfo(canceled, { + routing, + summary: execution.summary, + attemptId: null, + leaseToken: null, + }); + } + noteLeaseReleased("completed", taskRecord.id); + noteMonitorTaskCompleted(taskRecord.id); + recordActivity("info", "监控任务已按跨日策略丢弃", `${taskRecord.title} ${execution.summary}`); + return; + } + const completed = await completeDesktopTask(taskRecord.id, { + lease_token: leased.lease_token as string, + status: "unknown", + error: execution.error ?? { + code: "runtime_unknown", + message: execution.summary, + }, + }); + upsertTaskFromInfo(completed, { + routing, + summary: execution.summary, + attemptId: null, + leaseToken: null, + }); + noteLeaseReleased("completed", taskRecord.id); + noteMonitorTaskCompleted(taskRecord.id); + recordActivity("warn", "监控任务结果待确认", `${taskRecord.title} ${execution.summary}`); + return; + } + + await submitMonitoringTaskResult( + monitoringTaskID, + buildMonitoringResultPayload(leased.lease_token as string, execution), + ); + + const completed = await completeDesktopTask(taskRecord.id, { + lease_token: leased.lease_token as string, + status: execution.status, + payload: execution.payload, + error: execution.error, + }); + upsertTaskFromInfo(completed, { + routing, + summary: execution.summary, + attemptId: null, + leaseToken: null, + }); + noteLeaseReleased(execution.status === "failed" ? "failed" : "completed", taskRecord.id); + noteMonitorTaskCompleted(taskRecord.id); + recordActivity( + execution.status === "failed" ? "danger" : "success", + execution.status === "failed" ? "监控任务执行失败" : "监控任务执行成功", + `${taskRecord.title} ${execution.summary}`, + ); + } catch (error) { + if (isMonitorAbortReason(taskRecord, error, "stale_business_date")) { + let dropSummary = staleMonitoringDropSummary(taskRecord); + try { + await skipMonitoringTask(monitoringTaskID, { + lease_token: leased.lease_token as string, + skip_reason: "stale_business_date", + error_message: dropSummary, + completed_at: new Date().toISOString(), + }); + } catch (skipError) { + dropSummary = `${dropSummary}（monitoring skip 回写失败：${errorMessage(skipError)}）`; + } + + const canceled = await cancelDesktopTask(taskRecord.id, { + lease_token: leased.lease_token as string, + reason: "stale_business_date", + }).catch((cancelError) => { + const existing = state.tasks.get(taskRecord.id); + if (existing) { + existing.status = "aborted"; + existing.error = { + ...toStructuredError(error), + cancel_error: errorMessage(cancelError), + dropped_reason: "stale_business_date", + }; + existing.summary = `${dropSummary}（desktop task 取消失败：${errorMessage(cancelError)}）`; + existing.updatedAt = Date.now(); + existing.leaseToken = null; + state.tasks.set(taskRecord.id, existing); + } + return null; + }); + + if (canceled) { + upsertTaskFromInfo(canceled, { + routing, + summary: dropSummary, + attemptId: null, + leaseToken: null, + }); + } + noteLeaseReleased("completed", taskRecord.id); + noteMonitorTaskCompleted(taskRecord.id); + recordActivity("info", "监控任务已按跨日策略丢弃", dropSummary); + return; + } + + if (isDesktopTaskMonitorInterrupt(taskRecord, error)) { + const canceled = await cancelDesktopTask(taskRecord.id, { + lease_token: leased.lease_token as string, + reason: interruptReasonFromTask(taskRecord), + }); + upsertTaskFromInfo(canceled, { + routing, + summary: "已响应 Phase 2 抢占请求，当前监控任务已回队等待 retry。", + attemptId: null, + leaseToken: null, + }); + noteLeaseReleased("completed", taskRecord.id); + noteMonitorTaskCompleted(taskRecord.id); + recordActivity("warn", "监控任务已让路", `${taskRecord.title} 已安全中止，并回队为 retry。`); + return; + } + + const failureMessage = errorMessage(error); + const structuredError = toStructuredError(error); + let monitoringResultWriteError: unknown = null; + + try { + await submitMonitoringTaskResult( + monitoringTaskID, + buildMonitoringFailurePayload( + leased.lease_token as string, + failureMessage || "监控任务执行失败。", + structuredError, + ), + ); + } catch (submitError) { + monitoringResultWriteError = submitError; + } + + const failureSummary = monitoringResultWriteError + ? `${failureMessage || "监控任务执行失败。"}（monitoring 失败回传失败：${errorMessage(monitoringResultWriteError)}）` + : (failureMessage || "监控任务执行失败。"); + const completed = await completeDesktopTask(taskRecord.id, { + lease_token: leased.lease_token as string, + status: "failed", + error: structuredError, + }).catch(async (completeError) => { + const existing = state.tasks.get(taskRecord.id); + if (existing) { + existing.status = "failed"; + existing.error = structuredError; + existing.summary = `${failureSummary}（desktop task 回写失败：${errorMessage(completeError)}）`; + existing.updatedAt = Date.now(); + state.tasks.set(taskRecord.id, existing); + } + return null; + }); + + if (completed) { + upsertTaskFromInfo(completed, { + routing, + summary: failureSummary, + attemptId: null, + leaseToken: null, + }); + } + noteLeaseReleased("failed", taskRecord.id); + noteMonitorTaskCompleted(taskRecord.id); + recordActivity("danger", "监控任务执行失败", `${taskRecord.title} 执行失败：${failureSummary || "未知错误"}`); + } +} + +function extractMonitoringTaskID(payload: Record): number | null { + const raw = payload.monitoring_task_id; + if (typeof raw === "number" && Number.isFinite(raw)) { + return raw; + } + if (typeof raw === "string" && raw.trim()) { + const parsed = Number.parseInt(raw.trim(), 10); + if (Number.isFinite(parsed)) { + return parsed; } } + return null; +} + +function noteMonitorExecutionSafePoint( + taskId: string, + phase: MonitorExecutionPhase | null, +): void { + const active = state.activeExecutions.get(taskId); + if (!active || active.kind !== "monitor") { + return; + } + + if (phase) { + active.executionPhase = phase; + } + active.lastSafePointAt = Date.now(); + state.activeExecutions.set(taskId, active); + + const taskRecord = state.tasks.get(taskId); + if (taskRecord?.kind === "monitor") { + const staleBusinessDate = resolveStaleMonitoringBusinessDate(taskRecord.payload); + if (staleBusinessDate) { + requestMonitorStaleBusinessDateDrop(taskId, staleBusinessDate); + return; + } + } + + if (active.interruptRequested && !active.abortController.signal.aborted) { + active.abortController.abort(); + } +} + +function requestMonitorStaleBusinessDateDrop(taskId: string, staleBusinessDate: string): void { + const active = state.activeExecutions.get(taskId); + if (!active || active.kind !== "monitor") { + return; + } + + active.interruptReason = "stale_business_date"; + active.lastSafePointAt = Date.now(); + state.activeExecutions.set(taskId, active); + + const existing = state.tasks.get(taskId); + if (existing) { + existing.payload = { + ...existing.payload, + interrupt_reason: "stale_business_date", + dropped_reason: "stale_business_date", + }; + existing.summary = staleMonitoringDropSummary(existing, staleBusinessDate); + existing.updatedAt = Date.now(); + state.tasks.set(taskId, existing); + } + + if (!active.abortController.signal.aborted) { + active.abortController.abort(); + } +} + +function monitorExecutionPhaseFromStage(stage: string): MonitorExecutionPhase { + const normalized = stage.trim().toLowerCase(); + if (!normalized) { + return "WAITING"; + } + if (normalized.includes("bootstrap") || normalized.includes("page_ready")) { + return "NAVIGATING"; + } + if (normalized.includes("login") || normalized.includes("runtime_state")) { + return "AUTHENTICATING"; + } + if (normalized.includes("input")) { + return "INPUTTING"; + } + if (normalized.includes("query") || normalized.includes("submit")) { + return "SUBMITTING"; + } + if (normalized.includes("parse")) { + return "PARSING"; + } + return "WAITING"; +} + +function isDesktopTaskMonitorInterrupt(taskRecord: RuntimeTaskRecord, error: unknown): boolean { + return isDesktopTaskID(taskRecord.id) + && isMonitorAbortReason(taskRecord, error, "collect_now_preempt"); +} + +function isMonitorAbortReason(taskRecord: RuntimeTaskRecord, error: unknown, reason: string): boolean { + return errorMessage(error).includes("adapter_aborted") + && monitorAbortReason(taskRecord) === reason; +} + +function monitorAbortReason(taskRecord: RuntimeTaskRecord): string | null { + const active = state.activeExecutions.get(taskRecord.id); + if (active?.interruptReason?.trim()) { + return active.interruptReason.trim(); + } + + const raw = taskRecord.payload.interrupt_reason; + if (typeof raw === "string" && raw.trim()) { + return raw.trim(); + } + + if (Boolean(taskRecord.payload.interrupt_requested) || Boolean(active?.interruptRequested)) { + return "collect_now_preempt"; + } + return null; +} + +function interruptReasonFromTask(taskRecord: RuntimeTaskRecord): string { + return monitorAbortReason(taskRecord) ?? "collect_now_preempt"; +} + +function isDesktopTaskID(taskId: string): boolean { + return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(taskId.trim()); } async function executeLeasedTask( @@ -1083,6 +1932,11 @@ async function executeLeasedTask( platform: taskRecord.platform, accountId: taskRecord.accountId, abortController, + executionPhase: taskRecord.kind === "monitor" ? "PREPARING" : null, + interruptRequested: taskRecord.kind === "monitor" && Boolean(taskRecord.payload.interrupt_requested), + interruptGeneration: taskRecord.kind === "monitor" ? toInterruptGeneration(taskRecord.payload.interrupt_generation) : 0, + interruptReason: taskRecord.kind === "monitor" ? interruptReasonFromTask(taskRecord) : null, + lastSafePointAt: taskRecord.kind === "monitor" ? Date.now() : null, }); syncSchedulerSurface(); @@ -1095,6 +1949,11 @@ async function executeLeasedTask( }, leaseExtendIntervalMs); try { + if (task.kind === "monitor") { + await executeLeasedDesktopMonitorTask(taskRecord, leased, routing, abortController.signal); + return; + } + const execution = await executeTaskAdapter(taskRecord, abortController.signal); const completed = await completeDesktopTask(taskRecord.id, { lease_token: leased.lease_token, @@ -1121,9 +1980,6 @@ async function executeLeasedTask( `${taskRecord.title} ${execution.summary}`, ); - if (task.kind === "monitor") { - noteMonitorTaskCompleted(taskRecord.id); - } } catch (error) { const failureMessage = errorMessage(error); const structuredError = toStructuredError(error); @@ -1170,9 +2026,6 @@ async function executeLeasedTask( noteLeaseReleased("failed", taskRecord.id); recordActivity("danger", "任务执行失败", `${taskRecord.title} 执行失败：${failureMessage || "未知错误"}`); - if (task.kind === "monitor") { - noteMonitorTaskCompleted(taskRecord.id); - } } finally { clearInterval(extendHandle); state.activeExecutions.delete(taskRecord.id); @@ -1183,6 +2036,15 @@ async function executeLeasedTask( } async function extendActiveLease(taskId: string, leaseToken: string): Promise { + const existing = state.tasks.get(taskId); + if (existing?.kind === "monitor") { + const staleBusinessDate = resolveStaleMonitoringBusinessDate(existing.payload); + if (staleBusinessDate) { + requestMonitorStaleBusinessDateDrop(taskId, staleBusinessDate); + return; + } + } + try { const task = await extendDesktopTask(taskId, { lease_token: leaseToken }); @@ -1255,6 +2117,7 @@ async function executeTaskAdapter( phase: "initial", article: await loadPublishArticle(task), reportProgress(stage: string) { + noteMonitorExecutionSafePoint(task.id, null); updateTaskProgress(task.id, `publish adapter progress: ${stage}`); }, }, @@ -1323,6 +2186,7 @@ async function executeTaskAdapter( signal, phase: "initial", reportProgress(stage: string) { + noteMonitorExecutionSafePoint(task.id, monitorExecutionPhaseFromStage(stage)); updateTaskProgress(task.id, `monitor adapter progress: ${stage}`); }, }, @@ -1514,6 +2378,22 @@ function resolveStaleMonitoringBusinessDate(payload: Record): return businessDate < currentMonitoringBusinessDate() ? businessDate : null; } +function monitoringSkipReasonFromExecution(execution: AdapterExecutionResult): string { + return isStaleMonitoringExecutionResult(execution) ? "stale_business_date" : "runtime_unknown"; +} + +function isStaleMonitoringExecutionResult(execution: AdapterExecutionResult): boolean { + return execution.payload?.dropped_reason === "stale_business_date"; +} + +function staleMonitoringDropSummary(task: RuntimeTaskRecord, staleBusinessDate?: string | null): string { + const businessDate = staleBusinessDate ?? resolveMonitoringBusinessDate(task.payload); + if (businessDate) { + return `${task.title} 已过业务日 ${businessDate}，按漏采策略直接丢弃。`; + } + return `${task.title} 已跨业务日，按漏采策略直接丢弃。`; +} + function resolveMonitoringBusinessDate(payload: Record): string | null { const candidates = [ payload.business_date, @@ -1653,17 +2533,23 @@ function activeMonitorCount(): number { } function activeMonitorPlatforms(): Set { - return new Set( + const platforms = new Set( [...state.activeExecutions.values()] .filter((item) => item.kind === "monitor") .map((item) => item.platform), ); + for (const task of getMonitorSchedulerSnapshot().tasks) { + if (task.state === "active" || task.state === "leasing") { + platforms.add(task.platform); + } + } + return platforms; } function activeMonitorQuestionKeys(): Set { return new Set( getMonitorSchedulerSnapshot().tasks - .filter((task) => task.state === "active" && Boolean(task.questionKey)) + .filter((task) => (task.state === "active" || task.state === "leasing") && Boolean(task.questionKey)) .map((task) => task.questionKey as string), ); } @@ -1673,7 +2559,30 @@ function canStartAnotherMonitorExecution(): boolean { return false; } const limit = resolveAdaptiveMonitorConcurrency(); - return limit > 0 && activeMonitorCount() < limit; + if (limit <= 0) { + return false; + } + const scheduler = getMonitorSchedulerSnapshot(); + return activeMonitorCount() + scheduler.leasingCount < limit; +} + +function resolveLegacyMonitoringLeaseLimit(): number { + if (!canStartAnotherMonitorExecution()) { + return 0; + } + + const snapshot = getMonitorSchedulerSnapshot(); + if (snapshot.queuedCount > 0 || snapshot.leasingCount > 0) { + return 0; + } + + // Legacy monitoring leases cannot be extended while they sit in the local + // scheduler, so only pull one when it can start immediately. + if (activeMonitorCount() > 0) { + return 0; + } + + return 1; } function resolveAdaptiveMonitorConcurrency(): number { @@ -1788,6 +2697,10 @@ function resolveAccountName(accountId: string, fallback = "待同步账号"): st return account?.display_name ?? fallback; } +function currentClientID(): string { + return state.client?.id ?? state.session?.desktop_client?.id ?? ""; +} + function resolveTaskTitle( task: DesktopTaskInfo, payload: Record, @@ -1870,6 +2783,114 @@ function normalizeJsonObjectOrNull( return { ...value }; } +function asOptionalString(value: unknown): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const normalized = value.trim(); + return normalized || undefined; +} + +function asOptionalBoolean(value: unknown): boolean | undefined { + return typeof value === "boolean" ? value : undefined; +} + +function asOptionalNumber(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined; +} + +function asStringArray(value: unknown): string[] | undefined { + if (!Array.isArray(value)) { + return undefined; + } + const normalized = value + .filter((item): item is string => typeof item === "string") + .map((item) => item.trim()) + .filter((item) => item.length > 0); + return normalized.length > 0 ? normalized : undefined; +} + +function asRecord(value: unknown): Record | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return undefined; + } + return value as Record; +} + +function asSourceItems(value: unknown): MonitoringSourceItem[] | undefined { + if (!Array.isArray(value)) { + return undefined; + } + + const items: MonitoringSourceItem[] = []; + for (const entry of value) { + const record = asRecord(entry); + const url = asOptionalString(record?.url); + if (!record || !url) { + continue; + } + + const item: MonitoringSourceItem = { url }; + const title = asOptionalString(record.title); + const siteName = asOptionalString(record.site_name); + const siteKey = asOptionalString(record.site_key); + const normalizedURL = asOptionalString(record.normalized_url); + const host = asOptionalString(record.host); + const registrableDomain = asOptionalString(record.registrable_domain); + const subdomain = asOptionalString(record.subdomain); + const suffix = asOptionalString(record.suffix); + const articleID = asOptionalNumber(record.article_id); + const publishRecordID = asOptionalNumber(record.publish_record_id); + const resolutionStatus = asOptionalString(record.resolution_status); + const resolutionConfidence = asOptionalString(record.resolution_confidence); + + if (title !== undefined) { + item.title = title; + } + if (siteName !== undefined) { + item.site_name = siteName; + } + if (siteKey !== undefined) { + item.site_key = siteKey; + } + if (normalizedURL !== undefined) { + item.normalized_url = normalizedURL; + } + if (host !== undefined) { + item.host = host; + } + if (registrableDomain !== undefined) { + item.registrable_domain = registrableDomain; + } + if (subdomain !== undefined) { + item.subdomain = subdomain; + } + if (suffix !== undefined) { + item.suffix = suffix; + } + if (articleID !== undefined) { + item.article_id = articleID; + } + if (publishRecordID !== undefined) { + item.publish_record_id = publishRecordID; + } + if (resolutionStatus !== undefined) { + item.resolution_status = resolutionStatus; + } + if (resolutionConfidence !== undefined) { + item.resolution_confidence = resolutionConfidence; + } + + items.push(item); + } + + return items.length > 0 ? items : undefined; +} + +function toInterruptGeneration(value: unknown): number { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.floor(value) : 0; +} + function toStructuredError(error: unknown): Record { if (error instanceof ApiClientError) { return { @@ -1926,17 +2947,49 @@ function isDesktopTaskEvent(value: unknown): value is DesktopTaskEventMessage { const event = value as Partial; return Boolean( typeof event.type === "string" + && event.type !== "task_control" && typeof event.task_id === "string" - && typeof event.job_id === "string" - && typeof event.target_account_id === "string" + && event.task_id.trim() !== "" && typeof event.target_client_id === "string" - && typeof event.platform === "string" - && typeof event.kind === "string" + && event.target_client_id.trim() !== "" + && (event.kind === "publish" || event.kind === "monitor") && typeof event.status === "string" && typeof event.updated_at === "string", ); } +function isMonitoringDispatchSignal(value: unknown): value is DesktopTaskEventMessage { + if (typeof value !== "object" || value === null) { + return false; + } + + const event = value as Partial; + return ( + event.kind === "monitor" + && event.type === "task_available" + && event.signal_only === true + && typeof event.target_client_id === "string" + && event.target_client_id.trim() !== "" + ); +} + +function isMonitoringTaskControlEvent(value: unknown): value is DesktopTaskEventMessage { + if (typeof value !== "object" || value === null) { + return false; + } + + const event = value as Partial; + return ( + event.kind === "monitor" + && event.type === "task_control" + && event.control === "interrupt_requested" + && typeof event.task_id === "string" + && event.task_id.trim() !== "" + && typeof event.target_client_id === "string" + && event.target_client_id.trim() !== "" + ); +} + function isConnectedEvent(value: unknown): value is { server_time: string } { if (typeof value !== "object" || value === null) { return false; diff --git a/apps/desktop-client/src/main/runtime-snapshot.ts b/apps/desktop-client/src/main/runtime-snapshot.ts index 669d553..9b52330 100644 --- a/apps/desktop-client/src/main/runtime-snapshot.ts +++ b/apps/desktop-client/src/main/runtime-snapshot.ts @@ -7,6 +7,7 @@ import { collectRecoverySnapshot } from "./crash-recovery"; import { getKeepAlivePlan } from "./keep-alive"; import { getLeaseManagerSnapshot } from "./lease-manager"; import { getMonitorSchedulerSnapshot } from "./monitor-scheduler"; +import { getObservedRequestSnapshot } from "./network-observer"; import { getHiddenPlaywrightSnapshot } from "./playwright-cdp"; import { getRateLimiterSnapshot } from "./rate-limiter"; import { getRuntimeControllerSnapshot } from "./runtime-controller"; @@ -131,7 +132,7 @@ function createLiveRuntimeSnapshot(controller: ReturnType { console.warn("[desktop-session] clearHostResolverCache failed", error); }); @@ -121,7 +129,10 @@ export function createSessionHandle(accountId?: string): SessionHandle { const handle: SessionHandle = { accountId: resolvedAccountID, partition, - session: prepareSession(session.fromPartition(partition)), + session: prepareSession(session.fromPartition(partition), { + label: resolvedAccountID, + partition, + }), }; registry.set(resolvedAccountID, handle); return handle; @@ -136,7 +147,10 @@ export function createSessionHandleForPartition(accountId: string, partition: st const handle: SessionHandle = { accountId, partition, - session: prepareSession(session.fromPartition(partition)), + session: prepareSession(session.fromPartition(partition), { + label: accountId, + partition, + }), }; registry.set(accountId, handle); rememberPersistedPartition(accountId, partition); @@ -149,7 +163,10 @@ export function createPendingSessionHandle(seed = "bind"): SessionHandle { const handle: SessionHandle = { accountId, partition, - session: prepareSession(session.fromPartition(partition)), + session: prepareSession(session.fromPartition(partition), { + label: accountId, + partition, + }), }; registry.set(accountId, handle); return handle; @@ -189,7 +206,10 @@ export async function clearSessionHandle(accountId: string): Promise { return; } - const target = prepareSession(session.fromPartition(partition)); + const target = prepareSession(session.fromPartition(partition), { + label: accountId, + partition, + }); try { await target.clearStorageData(); diff --git a/apps/desktop-client/src/main/transport/api-client.ts b/apps/desktop-client/src/main/transport/api-client.ts index c9b8cb2..4fd87f3 100644 --- a/apps/desktop-client/src/main/transport/api-client.ts +++ b/apps/desktop-client/src/main/transport/api-client.ts @@ -2,6 +2,7 @@ import { ApiClientError, createApiClient, type ApiClient } from "@geo/http-clien import type { ApiEnvelope, ApiErrorEnvelope, + CancelDesktopTaskRequest, CompleteDesktopTaskRequest, CreatePublishJobResponse, DesktopArticleContent, @@ -15,19 +16,50 @@ import type { LeaseDesktopTaskRequest, LeaseDesktopTaskResponse, ListDesktopPublishTasksParams, + MonitoringLeaseTasksPayload, + MonitoringLeaseTasksResponse, + MonitoringResumeTasksPayload, + MonitoringResumeTasksResponse, + MonitoringSkipTaskPayload, + MonitoringSkipTaskResponse, + MonitoringTaskResultPayload, + MonitoringTaskResultResponse, UpsertDesktopAccountRequest, } from "@geo/shared-types"; import { canUseRendererDevtoolsProxy, rendererDevtoolsFetch } from "../renderer-devtools-proxy"; +import { + failObservedRequest, + resolveObservedRequest, + startObservedRequest, +} from "../network-observer"; type TransportAuthState = "pending" | "registered" | "expired"; -type TransportSseState = "idle" | "connecting" | "streaming"; +type TransportDispatchState = "idle" | "connecting" | "streaming"; + +type ObservedAxiosConfig = { + method?: string; + url?: string; + baseURL?: string; + headers?: Record; + __observedRequestId?: string; +}; interface DesktopTransportSession { baseURL: string; clientToken: string | null; } +interface MonitoringCancelTaskPayload { + lease_token: string; + reason?: string | null; +} + +interface MonitoringCancelTaskResponse { + task_id: number; + task_status: string; +} + let desktopApiClient: ApiClient | null = null; let transportSession: DesktopTransportSession = { baseURL: process.env.DESKTOP_API_BASE_URL ?? "http://localhost:8080", @@ -40,7 +72,7 @@ const transportState = { mode: "rabbitmq-first" as const, requestDebugMode: "main-process" as "main-process" | "renderer-devtools", auth: "pending" as TransportAuthState, - sseState: "idle" as TransportSseState, + dispatchState: "idle" as TransportDispatchState, lastHeartbeatAt: 0, lastHeartbeatStatus: "idle" as "idle" | "success" | "failed", lastPullAt: 0, @@ -58,9 +90,48 @@ function createDesktopClient(baseURL: string): ApiClient { config.headers = config.headers ?? {}; (config.headers as Record).Authorization = `Bearer ${transportSession.clientToken}`; } + + const observedConfig = config as ObservedAxiosConfig; + observedConfig.__observedRequestId = startObservedRequest({ + source: "transport", + label: "desktop.api", + method: config.method, + url: resolveObservedRequestURL(config.url, config.baseURL ?? baseURL), + }); return config; }); + client.raw.interceptors.response.use( + (response) => { + const observedConfig = response.config as ObservedAxiosConfig; + if (observedConfig.__observedRequestId) { + resolveObservedRequest(observedConfig.__observedRequestId, { + phase: "response", + status: response.status, + }); + } + return response; + }, + (error: unknown) => { + const responseError = error as { + config?: ObservedAxiosConfig; + response?: { status?: number }; + }; + const requestId = responseError.config?.__observedRequestId; + if (requestId) { + if (typeof responseError.response?.status === "number") { + resolveObservedRequest(requestId, { + phase: "response", + status: responseError.response.status, + }); + } else { + failObservedRequest(requestId, error); + } + } + return Promise.reject(error); + }, + ); + return client; } @@ -72,6 +143,33 @@ function resolveDesktopURL(path: string): string { return new URL(path, transportSession.baseURL).toString(); } +function resolveObservedRequestURL(path: string | undefined, baseURL: string): string { + if (!path) { + return baseURL; + } + + try { + return new URL(path, baseURL).toString(); + } catch { + return path; + } +} + +function shouldFallbackToMainProcess(error: unknown): boolean { + if (error instanceof ApiClientError) { + if (typeof error.status === "number" && error.status > 0) { + return false; + } + return /Failed to fetch|renderer_devtools_proxy_/i.test(error.message); + } + + if (error instanceof Error) { + return /Failed to fetch|renderer_devtools_proxy_/i.test(error.message); + } + + return false; +} + async function proxyDesktopRequest( method: "GET" | "POST" | "DELETE", path: string, @@ -134,7 +232,14 @@ async function proxyDesktopRequest( async function desktopGet(path: string): Promise { transportState.requestDebugMode = currentRequestDebugMode(); if (transportState.requestDebugMode === "renderer-devtools") { - return proxyDesktopRequest("GET", path); + try { + return await proxyDesktopRequest("GET", path); + } catch (error) { + if (!shouldFallbackToMainProcess(error)) { + throw error; + } + transportState.requestDebugMode = "main-process"; + } } return getDesktopApiClient().get(path); } @@ -142,7 +247,14 @@ async function desktopGet(path: string): Promise { async function desktopPost(path: string, body?: B): Promise { transportState.requestDebugMode = currentRequestDebugMode(); if (transportState.requestDebugMode === "renderer-devtools") { - return proxyDesktopRequest("POST", path, body); + try { + return await proxyDesktopRequest("POST", path, body); + } catch (error) { + if (!shouldFallbackToMainProcess(error)) { + throw error; + } + transportState.requestDebugMode = "main-process"; + } } return getDesktopApiClient().post(path, body); } @@ -150,7 +262,14 @@ async function desktopPost(path: string, body?: B): Promise { async function desktopDelete(path: string): Promise { transportState.requestDebugMode = currentRequestDebugMode(); if (transportState.requestDebugMode === "renderer-devtools") { - return proxyDesktopRequest("DELETE", path); + try { + return await proxyDesktopRequest("DELETE", path); + } catch (error) { + if (!shouldFallbackToMainProcess(error)) { + throw error; + } + transportState.requestDebugMode = "main-process"; + } } return getDesktopApiClient().remove(path); } @@ -176,6 +295,7 @@ export function configureTransport(session: DesktopRuntimeSessionSyncRequest | n transportState.baseURL = transportSession.baseURL; transportState.requestDebugMode = currentRequestDebugMode(); transportState.auth = "pending"; + transportState.dispatchState = "idle"; desktopApiClient = null; return null; } @@ -187,6 +307,7 @@ export function configureTransport(session: DesktopRuntimeSessionSyncRequest | n transportState.baseURL = transportSession.baseURL; transportState.requestDebugMode = currentRequestDebugMode(); transportState.auth = session.client_token ? "registered" : "pending"; + transportState.dispatchState = session.client_token ? "connecting" : "idle"; transportState.initializedAt = Date.now(); desktopApiClient = createDesktopClient(transportSession.baseURL); return desktopApiClient; @@ -196,8 +317,8 @@ export function setTransportAuthState(next: TransportAuthState): void { transportState.auth = next; } -export function setTransportSseState(next: TransportSseState): void { - transportState.sseState = next; +export function setTransportDispatchState(next: TransportDispatchState): void { + transportState.dispatchState = next; } export function noteTransportHeartbeat(success: boolean): void { @@ -328,3 +449,61 @@ export async function completeDesktopTask( payload, ); } + +export async function cancelDesktopTask( + taskId: string, + payload: CancelDesktopTaskRequest, +): Promise { + return desktopPost( + `/api/desktop/tasks/${taskId}/cancel`, + payload, + ); +} + +export async function leaseMonitoringTasks( + payload: MonitoringLeaseTasksPayload = {}, +): Promise { + return desktopPost( + "/api/desktop/monitoring/tasks/lease", + payload, + ); +} + +export async function resumeMonitoringTasks( + payload: MonitoringResumeTasksPayload = {}, +): Promise { + return desktopPost( + "/api/desktop/monitoring/tasks/resume", + payload, + ); +} + +export async function submitMonitoringTaskResult( + taskId: number, + payload: MonitoringTaskResultPayload, +): Promise { + return desktopPost( + `/api/desktop/monitoring/tasks/${taskId}/result`, + payload, + ); +} + +export async function skipMonitoringTask( + taskId: number, + payload: MonitoringSkipTaskPayload, +): Promise { + return desktopPost( + `/api/desktop/monitoring/tasks/${taskId}/skip`, + payload, + ); +} + +export async function cancelMonitoringTask( + taskId: number, + payload: MonitoringCancelTaskPayload, +): Promise { + return desktopPost( + `/api/desktop/monitoring/tasks/${taskId}/cancel`, + payload, + ); +} diff --git a/apps/desktop-client/src/main/transport/dispatch-ws-client.ts b/apps/desktop-client/src/main/transport/dispatch-ws-client.ts index 8567bbd..d500a2d 100644 --- a/apps/desktop-client/src/main/transport/dispatch-ws-client.ts +++ b/apps/desktop-client/src/main/transport/dispatch-ws-client.ts @@ -1,5 +1,11 @@ import WebSocket, { type RawData } from "ws"; +import { + failObservedRequest, + resolveObservedRequest, + startObservedRequest, +} from "../network-observer"; + export type DispatchEventHandler = (payload: unknown) => void; export interface DispatchWsClientOptions { @@ -34,6 +40,7 @@ export class DispatchWsClient { #socket: WebSocket | null = null; #retryDelayMs: number; #reconnectTimer: ReturnType | null = null; + #observedRequestId: string | null = null; constructor(options: DispatchWsClientOptions) { this.#options = options; @@ -65,6 +72,15 @@ export class DispatchWsClient { this.#clearReconnect(); this.#setState("idle"); + if (this.#observedRequestId) { + resolveObservedRequest(this.#observedRequestId, { + phase: "close", + status: 1000, + error: "client_stop", + }); + this.#observedRequestId = null; + } + const socket = this.#socket; this.#socket = null; if (!socket) { @@ -112,6 +128,13 @@ export class DispatchWsClient { return; } this.#setState("connecting"); + this.#observedRequestId = startObservedRequest({ + kind: "ws", + source: "transport", + label: "desktop.dispatch-ws", + method: "GET", + url: this.#options.url, + }); let socket: WebSocket; try { @@ -120,6 +143,10 @@ export class DispatchWsClient { handshakeTimeout: 15_000, }); } catch (error) { + if (this.#observedRequestId) { + failObservedRequest(this.#observedRequestId, error); + this.#observedRequestId = null; + } this.emit("error", error); this.#scheduleReconnect(); return; @@ -130,6 +157,13 @@ export class DispatchWsClient { socket.on("open", () => { this.#retryDelayMs = this.#options.initialRetryDelayMs ?? 1_000; this.#setState("open"); + if (this.#observedRequestId) { + resolveObservedRequest(this.#observedRequestId, { + phase: "response", + status: 101, + keepActive: true, + }); + } this.emit("open", null); }); @@ -154,11 +188,23 @@ export class DispatchWsClient { }); socket.on("error", (error: Error) => { + if (this.#observedRequestId) { + failObservedRequest(this.#observedRequestId, error); + this.#observedRequestId = null; + } this.emit("error", error); }); socket.on("close", (code: number, reason: Buffer) => { this.#socket = null; + if (this.#observedRequestId) { + resolveObservedRequest(this.#observedRequestId, { + phase: "close", + status: code, + error: reason.toString() || null, + }); + this.#observedRequestId = null; + } this.emit("close", { code, reason: reason.toString() }); if (!this.#running) { return; diff --git a/apps/desktop-client/src/main/view-pool.ts b/apps/desktop-client/src/main/view-pool.ts index aab67a4..f20542d 100644 --- a/apps/desktop-client/src/main/view-pool.ts +++ b/apps/desktop-client/src/main/view-pool.ts @@ -1,6 +1,8 @@ import { WebContentsView } from "electron/main"; import type { WebContentsView as ElectronWebContentsView } from "electron/main"; +import { maybeOpenExecutionDevtools } from "./execution-devtools"; + export interface HotViewHandle { accountId: string; view: ElectronWebContentsView; @@ -68,6 +70,7 @@ export function acquireHotView(accountId: string): HotViewHandle { lastReleasedAt: Date.now(), retainCount: 0, }; + maybeOpenExecutionDevtools(handle.view.webContents, `hot-view:${accountId}`, { background: true }); hotViews.set(accountId, handle); return handle; } diff --git a/apps/desktop-client/src/preload/bridge.ts b/apps/desktop-client/src/preload/bridge.ts index 950f2c7..433d50b 100644 --- a/apps/desktop-client/src/preload/bridge.ts +++ b/apps/desktop-client/src/preload/bridge.ts @@ -8,6 +8,7 @@ import type { ListDesktopPublishTasksParams, } from "@geo/shared-types"; import type { DesktopRuntimeSnapshot } from "../renderer/types"; +import type { DesktopObservedRequest } from "../shared/network-debug"; interface DesktopDeviceInfo { device_name: string; @@ -15,6 +16,68 @@ interface DesktopDeviceInfo { cpu_arch: string; } +const rendererProxyRequestChannel = "desktop:renderer-devtools-proxy:request"; +const rendererProxyResponseChannel = "desktop:renderer-devtools-proxy:response"; + +ipcRenderer.on( + rendererProxyRequestChannel, + async ( + _event, + payload: { + requestId?: unknown; + request?: { + url?: unknown; + method?: unknown; + headers?: unknown; + body?: unknown; + }; + }, + ) => { + const requestId = typeof payload?.requestId === "string" ? payload.requestId : ""; + const request = payload?.request; + if (!requestId || !request || typeof request !== "object") { + return; + } + + const url = typeof request.url === "string" ? request.url : ""; + const method = typeof request.method === "string" ? request.method : "GET"; + const headers = isStringRecord(request.headers) ? request.headers : {}; + const body = typeof request.body === "string" ? request.body : undefined; + + const response = await (async () => { + try { + const fetchResponse = await fetch(url, { + method, + headers, + body, + }); + const bodyText = await fetchResponse.text(); + return { + ok: fetchResponse.ok, + status: fetchResponse.status, + statusText: fetchResponse.statusText, + headers: Object.fromEntries(fetchResponse.headers.entries()), + bodyText, + }; + } catch (error) { + return { + ok: false, + status: 0, + statusText: "", + headers: {}, + bodyText: "", + error: String(error && ((error as Error).message || error)), + }; + } + })(); + + ipcRenderer.send(rendererProxyResponseChannel, { + requestId, + response, + }); + }, +); + const desktopBridge = { app: { ping: () => ipcRenderer.invoke("desktop:ping") as Promise, @@ -42,6 +105,8 @@ const desktopBridge = { ipcRenderer.invoke("desktop:runtime-session-sync", session) as Promise, releaseRuntimeSession: (revoke?: boolean) => ipcRenderer.invoke("desktop:runtime-session-release", revoke) as Promise, + setWindowMode: (mode: "login" | "main") => + ipcRenderer.invoke("desktop:set-window-mode", mode) as Promise, onRuntimeInvalidated: ( listener: (event: { reason: "account-health"; at: number }) => void, ) => { @@ -53,7 +118,24 @@ const desktopBridge = { ipcRenderer.off("desktop:runtime-invalidated", wrapped); }; }, + onNetworkObserved: (listener: (event: DesktopObservedRequest) => void) => { + const wrapped = (_event: IpcRendererEvent, payload: DesktopObservedRequest) => { + listener(payload); + }; + ipcRenderer.on("desktop:network-observed", wrapped); + return () => { + ipcRenderer.off("desktop:network-observed", wrapped); + }; + }, }, }; contextBridge.exposeInMainWorld("desktopBridge", desktopBridge); + +function isStringRecord(value: unknown): value is Record { + if (!value || typeof value !== "object") { + return false; + } + + return Object.values(value).every((item) => typeof item === "string"); +} diff --git a/apps/desktop-client/src/renderer/App.vue b/apps/desktop-client/src/renderer/App.vue index 2c5665e..b930a7b 100644 --- a/apps/desktop-client/src/renderer/App.vue +++ b/apps/desktop-client/src/renderer/App.vue @@ -1,5 +1,5 @@ diff --git a/apps/desktop-client/src/renderer/components/DesktopShell.vue b/apps/desktop-client/src/renderer/components/DesktopShell.vue index b1e43d8..afd5cad 100644 --- a/apps/desktop-client/src/renderer/components/DesktopShell.vue +++ b/apps/desktop-client/src/renderer/components/DesktopShell.vue @@ -82,6 +82,7 @@ const clientLabel = computed(