From 55e1c2e74b01e66f838fbb01adcab6333bf6cb7a Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 20 Apr 2026 17:16:15 +0800 Subject: [PATCH] feat(desktop): add monitor scheduler, Playwright CDP layer, and account health Move monitor-task scheduling authority onto the client with a durable file-backed queue that survives restart, drops stale cross-day tasks, enforces per-platform serialism, and adapts global concurrency from Electron process metrics. Publish tasks keep their existing FIFO. Add a hidden Playwright CDP manager that attaches to Electron Chromium on account session partitions, lets adapters opt into `executionMode: "playwright"`, and leaves the existing hidden WebContentsView path in place for current adapters. Introduce an account-health subsystem with silent probes, projected health/auth states, and IPC invalidation events so the renderer can show accurate auth/probe status and verification timestamps. Server-side, derive and forward title/business_date/scheduler_group_key/ question_text metadata on desktop task events so the local scheduler can defer same-question fan-out before leasing. --- .../desktop-client/src/main/account-binder.ts | 376 ++++++++++- .../desktop-client/src/main/account-health.ts | 638 ++++++++++++++++++ apps/desktop-client/src/main/adapters/base.ts | 5 + apps/desktop-client/src/main/auth-types.ts | 61 ++ apps/desktop-client/src/main/bootstrap.ts | 20 + apps/desktop-client/src/main/lease-manager.ts | 71 +- .../src/main/monitor-scheduler.ts | 535 +++++++++++++++ .../src/main/platform-auth-adapters.ts | 102 +++ .../desktop-client/src/main/playwright-cdp.ts | 317 +++++++++ .../src/main/runtime-controller.ts | 566 +++++++++++++--- .../desktop-client/src/main/runtime-events.ts | 34 + .../src/main/runtime-snapshot.ts | 25 +- apps/desktop-client/src/preload/bridge.ts | 12 + .../renderer/composables/useDesktopRuntime.ts | 20 + apps/desktop-client/src/renderer/env.d.ts | 3 + apps/desktop-client/src/renderer/types.ts | 17 + .../src/renderer/views/AccountsView.vue | 29 +- .../src/renderer/views/AiPlatformsView.vue | 50 +- .../src/renderer/views/HomeView.vue | 2 +- findings.md | 21 + packages/shared-types/src/index.ts | 4 + progress.md | 49 ++ .../tenant/app/desktop_task_events.go | 111 ++- .../tenant/app/desktop_task_service.go | 25 +- .../tenant/app/publish_job_service.go | 25 +- task_plan.md | 36 +- 26 files changed, 3001 insertions(+), 153 deletions(-) create mode 100644 apps/desktop-client/src/main/account-health.ts create mode 100644 apps/desktop-client/src/main/auth-types.ts create mode 100644 apps/desktop-client/src/main/monitor-scheduler.ts create mode 100644 apps/desktop-client/src/main/platform-auth-adapters.ts create mode 100644 apps/desktop-client/src/main/playwright-cdp.ts create mode 100644 apps/desktop-client/src/main/runtime-events.ts diff --git a/apps/desktop-client/src/main/account-binder.ts b/apps/desktop-client/src/main/account-binder.ts index b388d90..8484465 100644 --- a/apps/desktop-client/src/main/account-binder.ts +++ b/apps/desktop-client/src/main/account-binder.ts @@ -37,6 +37,7 @@ import { normalizeAccountPlatformUid, sameAccountPlatformUid, } from "../shared/account-identity"; +import type { AccountHealthProfile, AuthProbeResult } from "./auth-types"; interface DetectedAccount { platformUid: string; @@ -57,7 +58,7 @@ interface PublishPlatformBindingDefinition { detect(context: DetectContext): Promise; } -interface PublishAccountIdentity { +export interface PublishAccountIdentity { id: string; platform: string; platformUid: string; @@ -689,6 +690,7 @@ interface GenericAIPageState { strongAuthSignalCount: number; loginSignalCount: number; loggedOutSignalCount: number; + challengeSignalCount: number; } async function readGenericAIPageState( @@ -721,6 +723,7 @@ async function readGenericAIPageState( const stopWords = /^(登录|注册|设置|帮助|历史|聊天|新建|菜单|账户|账号|个人中心|联网搜索|深度思考|工具|更多|安装电脑版|下载电脑版|DeepSeek|元宝|混元|豆包|Kimi|Qwen|通义千问)$/i; const loginPattern = /(登录|注册|立即登录|扫码登录|微信登录|QQ登录|手机号登录|请先登录)/i; const loggedOutPattern = /(未登录|扫码登录|快捷登录|使用其他头像、昵称或账号|登录后继续|请先登录|立即登录|微信快捷登录)/i; + const challengePattern = /(验证码|滑块|人机验证|安全验证|请完成验证|短信验证|手机验证码|扫码验证|verify|verification|captcha|security check)/i; const credentialKeyPattern = /(token|session|auth|login|passport|ticket|jwt|refresh|access)/i; const isInsideAuthSurface = (node) => { if (!(node instanceof Element)) { @@ -850,6 +853,26 @@ async function readGenericAIPageState( } return matches; }; + const collectChallengeSignals = () => { + const matches = []; + if (bodyText && challengePattern.test(bodyText)) { + matches.push('body-text'); + } + const selectors = ['button', 'a', '[role="button"]', '[role="dialog"]', 'input', 'label']; + for (const selector of selectors) { + const nodes = document.querySelectorAll(selector); + for (const node of nodes) { + if (!isVisible(node)) continue; + const text = normalize(node.textContent || node.getAttribute?.('aria-label') || ""); + if (!text || !challengePattern.test(text)) continue; + matches.push(text); + if (matches.length >= 4) { + return matches; + } + } + } + return matches; + }; const collectCredentialStorage = (storage, prefix) => { const items = []; try { @@ -902,6 +925,7 @@ async function readGenericAIPageState( } const loginSignals = collectLoginSignals(); const loggedOutSignals = collectLoggedOutSignals(); + const challengeSignals = collectChallengeSignals(); const credentialFingerprintParts = [ ...collectCredentialStorage(window.localStorage, 'local'), ...collectCredentialStorage(window.sessionStorage, 'session'), @@ -924,6 +948,7 @@ async function readGenericAIPageState( strongAuthSignalCount: strongSignals.length, loginSignalCount: loginSignals.length, loggedOutSignalCount: loggedOutSignals.length, + challengeSignalCount: challengeSignals.length, }; } catch { return null; @@ -955,6 +980,10 @@ async function readGenericAIPageState( typeof typed.loggedOutSignalCount === "number" && Number.isFinite(typed.loggedOutSignalCount) ? typed.loggedOutSignalCount : 0, + challengeSignalCount: + typeof typed.challengeSignalCount === "number" && Number.isFinite(typed.challengeSignalCount) + ? typed.challengeSignalCount + : 0, }; } @@ -1285,6 +1314,29 @@ function toPublishAccountProfile(detected: DetectedAccount | null): PublishAccou }; } +function toAccountHealthProfile(profile: PublishAccountProfile | null): AccountHealthProfile | null { + if (!profile) { + return null; + } + + return { + subjectUid: profile.platformUid, + displayName: profile.displayName, + avatarUrl: profile.avatarUrl, + }; +} + +function fallbackAccountHealthProfile(account: PublishAccountIdentity, overrides: { + displayName?: string | null; + avatarUrl?: string | null; +} = {}): AccountHealthProfile { + return { + subjectUid: normalizeAccountPlatformUid(account.platformUid) || account.platformUid, + displayName: overrides.displayName ?? account.displayName, + avatarUrl: overrides.avatarUrl ?? null, + }; +} + async function resolvePublishAccountProfileFromHandle( account: PublishAccountIdentity, handle: SessionHandle, @@ -1332,6 +1384,328 @@ async function resolvePublishAccountProfileFromHandle( return toPublishAccountProfile(detected); } +export async function probePublishAccountSession( + account: PublishAccountIdentity, + partition?: string, +): Promise { + const handle = partition + ? createSessionHandleForPartition(account.id, partition) + : await findLocalPublishAccountSessionHandle(account); + + if (!handle) { + return { + verdict: "expired", + reason: "missing_partition", + profile: null, + sessionFingerprint: null, + }; + } + + const definition = platformBindingDefinitions[account.platform]; + if (!definition) { + return { + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: null, + evidence: { + platform: account.platform, + code: "desktop_platform_not_supported", + }, + }; + } + + const window = createBoundWindow( + `${definition.label} 静默探针`, + definition.consoleUrl, + handle.session, + { show: false }, + ); + + let mainFrameLoadError: + | { + code: number; + description: string; + url: string; + } + | undefined; + window.webContents.on( + "did-fail-load", + (_event, errorCode, errorDescription, validatedURL, isMainFrame) => { + if (isMainFrame && errorCode !== -3) { + mainFrameLoadError = { + code: errorCode, + description: errorDescription, + url: validatedURL, + }; + } + }, + ); + + try { + await waitForWindowNavigationSettled(window); + if (window.isDestroyed()) { + return { + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: null, + }; + } + + const loadError = mainFrameLoadError; + if (loadError) { + return { + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: null, + evidence: { + errorCode: loadError.code, + errorDescription: loadError.description, + url: loadError.url, + }, + }; + } + + const currentURL = window.webContents.getURL(); + if (!currentURL || currentURL.startsWith("data:text/html")) { + return { + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: null, + evidence: { + url: currentURL, + }, + }; + } + + if (looksLikeLoginRedirect(currentURL, definition.loginUrl)) { + return { + verdict: "expired", + reason: "login_redirect", + profile: null, + sessionFingerprint: null, + evidence: { + url: currentURL, + }, + }; + } + + if (account.platform === "toutiaohao") { + const detected = await detectToutiaoFromSession(handle.session); + if (detected) { + return { + verdict: "active", + reason: "probe_success", + profile: toAccountHealthProfile(toPublishAccountProfile(detected)), + sessionFingerprint: detected.platformUid, + }; + } + + const consoleReady = await verifyToutiaoConsoleAccess(window, handle.session); + if (consoleReady) { + return { + verdict: "active", + reason: "probe_success", + profile: fallbackAccountHealthProfile(account), + sessionFingerprint: normalizeAccountPlatformUid(account.platformUid) || account.platformUid, + }; + } + + return { + verdict: "expired", + reason: "login_redirect", + profile: null, + sessionFingerprint: null, + }; + } + + if (isAIPlatformBinding(account.platform)) { + const pageState = await readGenericAIPageState(window.webContents); + if ((pageState?.challengeSignalCount ?? 0) > 0) { + return { + verdict: "challenge_required", + reason: "captcha_gate", + profile: null, + sessionFingerprint: null, + evidence: { + url: currentURL, + challengeSignalCount: pageState?.challengeSignalCount ?? 0, + }, + }; + } + + const detected = await detectGenericAIPlatform(account.platform, definition.label, { + session: handle.session, + webContents: window.webContents, + }); + if (detected) { + return { + verdict: "active", + reason: "probe_success", + profile: toAccountHealthProfile(toPublishAccountProfile(detected)), + sessionFingerprint: detected.platformUid, + }; + } + + if (!pageState) { + return { + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: null, + evidence: { + url: currentURL, + }, + }; + } + + if (pageState.loggedOutSignalCount > 0 || looksLikeLoginRedirect(currentURL, definition.loginUrl)) { + return { + verdict: "expired", + reason: "login_redirect", + profile: null, + sessionFingerprint: null, + evidence: { + url: currentURL, + loggedOutSignalCount: pageState.loggedOutSignalCount, + }, + }; + } + + if ((pageState.loginSignalCount ?? 0) > 0 && (pageState.strongAuthSignalCount ?? 0) === 0) { + return { + verdict: "expired", + reason: "login_redirect", + profile: null, + sessionFingerprint: null, + evidence: { + url: currentURL, + loginSignalCount: pageState.loginSignalCount, + }, + }; + } + + const fingerprint = await buildGenericAISessionFingerprint(account.platform, handle.session, pageState); + if (genericAIPageLooksAuthenticated(currentURL, pageState)) { + return { + verdict: "active", + reason: "probe_success", + profile: fallbackAccountHealthProfile(account, { + displayName: pageState.displayName ?? account.displayName, + avatarUrl: pageState.avatarUrl ?? null, + }), + sessionFingerprint: fingerprint, + }; + } + + return { + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: fingerprint, + evidence: { + url: currentURL, + strongAuthSignalCount: pageState.strongAuthSignalCount, + loginSignalCount: pageState.loginSignalCount, + loggedOutSignalCount: pageState.loggedOutSignalCount, + }, + }; + } + + const detected = await definition.detect({ + session: handle.session, + webContents: window.webContents, + }); + if (detected) { + return { + verdict: "active", + reason: "probe_success", + profile: toAccountHealthProfile(toPublishAccountProfile(detected)), + sessionFingerprint: sanitizeDetectedAccount(detected).platformUid, + }; + } + + const consoleReady = await verifyPublishAccountConsoleAccess(account, handle); + if (consoleReady) { + return { + verdict: "active", + reason: "probe_success", + profile: fallbackAccountHealthProfile(account), + sessionFingerprint: normalizeAccountPlatformUid(account.platformUid) || account.platformUid, + }; + } + + return { + verdict: "expired", + reason: "login_redirect", + profile: null, + sessionFingerprint: null, + evidence: { + url: currentURL, + }, + }; + } catch (error) { + return { + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: null, + evidence: { + message: error instanceof Error ? error.message : String(error), + }, + }; + } finally { + if (!window.isDestroyed()) { + window.destroy(); + } + } +} + +export async function silentRefreshPublishAccountSession( + account: PublishAccountIdentity, + partition?: string, +): Promise { + const handle = partition + ? createSessionHandleForPartition(account.id, partition) + : await findLocalPublishAccountSessionHandle(account); + + if (!handle) { + return false; + } + + const definition = platformBindingDefinitions[account.platform]; + if (!definition) { + return false; + } + + const window = createBoundWindow( + `${definition.label} 静默续期`, + definition.consoleUrl, + handle.session, + { show: false }, + ); + + try { + await waitForWindowNavigationSettled(window, 10_000); + if (window.isDestroyed()) { + return false; + } + + await sleep(1_200); + await flushSessionPersistence(handle.session); + return !looksLikeLoginRedirect(window.webContents.getURL(), definition.loginUrl); + } catch { + return false; + } finally { + if (!window.isDestroyed()) { + window.destroy(); + } + } +} + export async function inspectPublishAccountLocalSession( account: PublishAccountIdentity, ): Promise { diff --git a/apps/desktop-client/src/main/account-health.ts b/apps/desktop-client/src/main/account-health.ts new file mode 100644 index 0000000..fd98551 --- /dev/null +++ b/apps/desktop-client/src/main/account-health.ts @@ -0,0 +1,638 @@ +import { mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { dirname, join } from "node:path"; + +import { app } from "electron/main"; +import type { DesktopAccountInfo } from "@geo/shared-types"; + +import type { + AccountAuthReason, + AccountAuthState, + AccountHealthProfile, + AccountHealthSnapshot, + AccountProbeState, + AuthProbeResult, + PlatformFailureClassification, +} from "./auth-types"; +import { emitRuntimeInvalidated } from "./runtime-events"; +import { getPersistedPartition, getSessionHandle } from "./session-registry"; +import { + getPlatformAdapter, + type PlatformFailureInput, +} from "./platform-auth-adapters"; +import type { PublishAccountIdentity } from "./account-binder"; + +const PROBE_TICK_MS = 10_000; +const FIRST_PROBE_MIN_MS = 5_000; +const FIRST_PROBE_MAX_MS = 15_000; +const REGULAR_PROBE_MIN_MS = 30 * 60_000; +const REGULAR_PROBE_MAX_MS = 60 * 60_000; +const CONFIRM_BACKOFF_STEPS_MS = [30_000, 120_000] as const; +const NETWORK_RETRY_MS = 2 * 60_000; +const STALE_AFTER_MS = 45 * 60_000; +const PREFLIGHT_MAX_AGE_MS = 15 * 60_000; + +interface PersistedAccountHealthRecord { + accountId: string; + platform: string; + lastVerifiedAt: number | null; + lastAuthFailureAt: number | null; + profile: AccountHealthProfile | null; + sessionFingerprint: string | null; +} + +interface AccountHealthRecord { + accountId: string; + platform: string; + authState: AccountAuthState; + probeState: AccountProbeState; + authReason: AccountAuthReason; + lastVerifiedAt: number | null; + lastProbeAt: number | null; + nextProbeAt: number | null; + lastAuthFailureAt: number | null; + profile: AccountHealthProfile | null; + sessionFingerprint: string | null; + suspectedExpiredAt: number | null; + confirmFailureCount: number; +} + +const trackedAccounts = new Map(); +const records = new Map(); +const accountLocks = new Map>(); +let initialized = false; +let probeTimer: ReturnType | null = null; + +function randomInt(min: number, max: number): number { + if (max <= min) { + return min; + } + return min + Math.floor(Math.random() * (max - min)); +} + +function persistedHealthPath(): string { + return join(app.getPath("userData"), "desktop-account-health.json"); +} + +function resolvePartition(accountId: string): string { + return getSessionHandle(accountId)?.partition ?? getPersistedPartition(accountId) ?? `persist:acc-${accountId}`; +} + +function nextInitialProbeAt(now = Date.now()): number { + return now + randomInt(FIRST_PROBE_MIN_MS, FIRST_PROBE_MAX_MS); +} + +function nextRegularProbeAt(now = Date.now()): number { + return now + randomInt(REGULAR_PROBE_MIN_MS, REGULAR_PROBE_MAX_MS); +} + +function normalizePersisted( + record: PersistedAccountHealthRecord, +): AccountHealthRecord { + return { + accountId: record.accountId, + platform: record.platform, + authState: "unknown", + probeState: "idle", + authReason: null, + lastVerifiedAt: record.lastVerifiedAt ?? null, + lastProbeAt: null, + nextProbeAt: nextInitialProbeAt(), + lastAuthFailureAt: record.lastAuthFailureAt ?? null, + profile: record.profile ?? null, + sessionFingerprint: record.sessionFingerprint ?? null, + suspectedExpiredAt: null, + confirmFailureCount: 0, + }; +} + +function loadPersistedRecords(): void { + try { + const raw = JSON.parse(readFileSync(persistedHealthPath(), "utf8")) as PersistedAccountHealthRecord[]; + for (const item of raw) { + if (!item?.accountId || !item?.platform) { + continue; + } + records.set(item.accountId, normalizePersisted(item)); + } + } catch { + records.clear(); + } +} + +function persistRecords(): void { + const payload: PersistedAccountHealthRecord[] = [...records.values()].map((record) => ({ + accountId: record.accountId, + platform: record.platform, + lastVerifiedAt: record.lastVerifiedAt, + lastAuthFailureAt: record.lastAuthFailureAt, + profile: record.profile, + sessionFingerprint: record.sessionFingerprint, + })); + + const target = persistedHealthPath(); + mkdirSync(dirname(target), { recursive: true }); + writeFileSync(target, JSON.stringify(payload, null, 2), "utf8"); +} + +function ensureInitialized(): void { + if (initialized) { + return; + } + + initialized = true; + loadPersistedRecords(); +} + +function ensureProbeTimer(): void { + ensureInitialized(); + if (probeTimer) { + return; + } + + probeTimer = setInterval(() => { + void runDueProbes(); + }, PROBE_TICK_MS); +} + +function coerceDerivedAuthState(record: AccountHealthRecord, now = Date.now()): AccountAuthState { + if ( + record.authState === "active" + && record.lastVerifiedAt + && now - record.lastVerifiedAt >= STALE_AFTER_MS + ) { + return "expiring_soon"; + } + + return record.authState; +} + +function toPublicSnapshot(record: AccountHealthRecord): AccountHealthSnapshot { + return { + accountId: record.accountId, + platform: record.platform, + partition: resolvePartition(record.accountId), + authState: coerceDerivedAuthState(record), + probeState: record.probeState, + authReason: record.authReason, + lastVerifiedAt: record.lastVerifiedAt, + lastProbeAt: record.lastProbeAt, + nextProbeAt: record.nextProbeAt, + lastAuthFailureAt: record.lastAuthFailureAt, + profile: record.profile, + sessionFingerprint: record.sessionFingerprint, + }; +} + +function snapshotSignature(record: AccountHealthRecord): string { + return JSON.stringify(toPublicSnapshot(record)); +} + +function commitRecord(record: AccountHealthRecord, previousSignature: string | null): AccountHealthSnapshot { + records.set(record.accountId, record); + persistRecords(); + + const nextSignature = snapshotSignature(record); + if (previousSignature !== nextSignature) { + emitRuntimeInvalidated("account-health"); + } + + return toPublicSnapshot(record); +} + +function ensureRecord(account: PublishAccountIdentity): AccountHealthRecord { + ensureInitialized(); + + const existing = records.get(account.id); + if (existing) { + existing.platform = account.platform; + return existing; + } + + const created: AccountHealthRecord = { + accountId: account.id, + platform: account.platform, + authState: "unknown", + probeState: "idle", + authReason: null, + lastVerifiedAt: null, + lastProbeAt: null, + nextProbeAt: nextInitialProbeAt(), + lastAuthFailureAt: null, + profile: null, + sessionFingerprint: null, + suspectedExpiredAt: null, + confirmFailureCount: 0, + }; + + records.set(account.id, created); + persistRecords(); + return created; +} + +async function withAccountLock(accountId: string, work: () => Promise): Promise { + const previous = accountLocks.get(accountId) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise((resolve) => { + release = resolve; + }); + + accountLocks.set(accountId, current); + await previous.catch(() => undefined); + + try { + return await work(); + } finally { + release(); + if (accountLocks.get(accountId) === current) { + accountLocks.delete(accountId); + } + } +} + +function confirmBackoffMs(failureCount: number): number { + const index = Math.min(Math.max(failureCount - 1, 0), CONFIRM_BACKOFF_STEPS_MS.length - 1); + return CONFIRM_BACKOFF_STEPS_MS[index] ?? CONFIRM_BACKOFF_STEPS_MS[CONFIRM_BACKOFF_STEPS_MS.length - 1]; +} + +function applyActiveResult( + record: AccountHealthRecord, + result: AuthProbeResult, + now: number, +): AccountHealthSnapshot { + const previousSignature = snapshotSignature(record); + + record.authState = "active"; + record.probeState = "idle"; + record.authReason = result.reason ?? "probe_success"; + record.lastVerifiedAt = now; + record.lastProbeAt = now; + record.nextProbeAt = nextRegularProbeAt(now); + record.profile = result.profile ?? record.profile; + record.sessionFingerprint = result.sessionFingerprint ?? record.sessionFingerprint; + record.suspectedExpiredAt = null; + record.confirmFailureCount = 0; + + return commitRecord(record, previousSignature); +} + +function applyExpiredResult( + record: AccountHealthRecord, + result: AuthProbeResult, + now: number, +): AccountHealthSnapshot { + const previousSignature = snapshotSignature(record); + + record.authState = "expired"; + record.probeState = "idle"; + record.authReason = result.reason ?? "login_redirect"; + record.lastProbeAt = now; + record.nextProbeAt = nextRegularProbeAt(now); + record.lastAuthFailureAt = now; + record.suspectedExpiredAt = null; + record.confirmFailureCount = 0; + + return commitRecord(record, previousSignature); +} + +function applyChallengeResult( + record: AccountHealthRecord, + result: AuthProbeResult, + now: number, +): AccountHealthSnapshot { + const previousSignature = snapshotSignature(record); + + record.authState = "challenge_required"; + record.probeState = "idle"; + record.authReason = result.reason ?? "captcha_gate"; + record.lastProbeAt = now; + record.nextProbeAt = nextRegularProbeAt(now); + record.lastAuthFailureAt = now; + record.suspectedExpiredAt = null; + record.confirmFailureCount = 0; + + return commitRecord(record, previousSignature); +} + +function applyNetworkResult( + record: AccountHealthRecord, + result: AuthProbeResult, + now: number, + trigger: "scheduled" | "manual" | "preflight" | "confirm", +): AccountHealthSnapshot { + const previousSignature = snapshotSignature(record); + const hadSuspicion = Boolean(record.suspectedExpiredAt); + + record.probeState = "network_error"; + record.authReason = result.reason ?? "network_error"; + record.lastProbeAt = now; + if (hadSuspicion && trigger === "confirm") { + record.confirmFailureCount += 1; + record.authState = record.authState === "unknown" ? "unknown" : "expiring_soon"; + record.nextProbeAt = now + confirmBackoffMs(record.confirmFailureCount); + } else { + record.nextProbeAt = now + NETWORK_RETRY_MS; + if (record.authState === "active") { + record.authState = "expiring_soon"; + } + } + + return commitRecord(record, previousSignature); +} + +async function performProbeLocked( + account: PublishAccountIdentity, + options: { + trigger: "scheduled" | "manual" | "preflight" | "confirm"; + allowSilentRefresh: boolean; + }, +): Promise { + const record = ensureRecord(account); + const adapter = getPlatformAdapter(account.platform); + const previousSignature = snapshotSignature(record); + + record.probeState = "probing"; + commitRecord(record, previousSignature); + + const partition = resolvePartition(account.id); + + if (options.allowSilentRefresh) { + const refreshed = await adapter.silentRefresh(account, partition).catch(() => false); + if (!refreshed) { + const currentSignature = snapshotSignature(record); + record.authReason = "silent_refresh_failed"; + commitRecord(record, currentSignature); + } + } + + const result = await adapter.probe(account, partition).catch(() => ({ + verdict: "network_error", + reason: "network_error", + profile: null, + sessionFingerprint: null, + })); + + const now = Date.now(); + switch (result.verdict) { + case "active": + return applyActiveResult(record, result, now); + case "expired": + return applyExpiredResult(record, result, now); + case "challenge_required": + return applyChallengeResult(record, result, now); + default: + return applyNetworkResult(record, result, now, options.trigger); + } +} + +async function runDueProbes(): Promise { + const now = Date.now(); + const dueAccounts = [...trackedAccounts.values()].filter((account) => { + const record = records.get(account.id); + return Boolean(record?.nextProbeAt && record.nextProbeAt <= now); + }); + + for (const account of dueAccounts) { + const record = records.get(account.id); + if (!record) { + continue; + } + + const trigger = record.suspectedExpiredAt ? "confirm" : "scheduled"; + void withAccountLock(account.id, async () => { + await performProbeLocked(account, { + trigger, + allowSilentRefresh: false, + }); + }); + } +} + +function parseVerifiedAt(value: string | null | undefined): number | null { + if (!value) { + return null; + } + + const timestamp = Date.parse(value); + return Number.isNaN(timestamp) ? null : timestamp; +} + +function fallbackAuthState( + health: DesktopAccountInfo["health"] | null | undefined, +): AccountAuthState { + switch (health) { + case "live": + return "unknown"; + case "captcha": + return "challenge_required"; + case "expired": + return "expired"; + default: + return "unknown"; + } +} + +export function initAccountHealth(): void { + ensureInitialized(); + ensureProbeTimer(); +} + +export function syncTrackedAccounts(accounts: PublishAccountIdentity[]): void { + syncTrackedAccountsInternal(accounts, true); +} + +function syncTrackedAccountsInternal( + accounts: PublishAccountIdentity[], + pruneMissing: boolean, +): void { + initAccountHealth(); + let removed = false; + + if (pruneMissing) { + const nextIds = new Set(accounts.map((account) => account.id)); + for (const [accountId] of trackedAccounts.entries()) { + if (!nextIds.has(accountId)) { + trackedAccounts.delete(accountId); + records.delete(accountId); + removed = true; + } + } + } + + for (const account of accounts) { + trackedAccounts.set(account.id, account); + ensureRecord(account); + } + + persistRecords(); + if (removed) { + emitRuntimeInvalidated("account-health"); + } +} + +export function forgetTrackedAccountHealth( + accountId: string, +): void { + trackedAccounts.delete(accountId); + if (records.delete(accountId)) { + persistRecords(); + emitRuntimeInvalidated("account-health"); + } +} + +export function getAccountHealthSnapshot(accountId: string): AccountHealthSnapshot | null { + ensureInitialized(); + const record = records.get(accountId); + return record ? toPublicSnapshot(record) : null; +} + +export function getProjectedAccountHealth(input: { + accountId: string; + platform: string; + health?: DesktopAccountInfo["health"]; + verifiedAt?: string | null; +}) { + const snapshot = getAccountHealthSnapshot(input.accountId); + if (!snapshot) { + return { + health: input.health ?? "risk", + authState: fallbackAuthState(input.health), + probeState: "idle" as AccountProbeState, + authReason: null as AccountAuthReason, + lastVerifiedAt: parseVerifiedAt(input.verifiedAt), + lastProbeAt: null, + nextProbeAt: null, + lastAuthFailureAt: null, + profile: null, + sessionFingerprint: null, + partition: resolvePartition(input.accountId), + }; + } + + const health = + snapshot.authState === "active" + ? ("live" as const) + : snapshot.authState === "challenge_required" + ? ("captcha" as const) + : snapshot.authState === "expired" || snapshot.authState === "revoked" + ? ("expired" as const) + : ("risk" as const); + + return { + ...snapshot, + health, + }; +} + +export async function probeTrackedAccount( + account: PublishAccountIdentity, + options: { + trigger?: "scheduled" | "manual" | "preflight" | "confirm"; + allowSilentRefresh?: boolean; + } = {}, +): Promise { + syncTrackedAccountsInternal([account], false); + return await withAccountLock(account.id, async () => { + return await performProbeLocked(account, { + trigger: options.trigger ?? "manual", + allowSilentRefresh: Boolean(options.allowSilentRefresh), + }); + }); +} + +export async function ensureAccountReady( + account: PublishAccountIdentity, +): Promise { + syncTrackedAccountsInternal([account], false); + const snapshot = getAccountHealthSnapshot(account.id); + const isFresh = + snapshot?.lastVerifiedAt + ? Date.now() - snapshot.lastVerifiedAt < PREFLIGHT_MAX_AGE_MS + : false; + + if ( + snapshot + && snapshot.authState === "active" + && snapshot.probeState !== "network_error" + && isFresh + ) { + return snapshot; + } + + if (snapshot && ["expired", "challenge_required", "revoked"].includes(snapshot.authState)) { + return snapshot; + } + + return await withAccountLock(account.id, async () => { + const lockedSnapshot = getAccountHealthSnapshot(account.id); + const lockedFresh = + lockedSnapshot?.lastVerifiedAt + ? Date.now() - lockedSnapshot.lastVerifiedAt < PREFLIGHT_MAX_AGE_MS + : false; + + if ( + lockedSnapshot + && lockedSnapshot.authState === "active" + && lockedSnapshot.probeState !== "network_error" + && lockedFresh + ) { + return lockedSnapshot; + } + + return await performProbeLocked(account, { + trigger: "preflight", + allowSilentRefresh: true, + }); + }); +} + +export async function reportAccountFailure( + account: PublishAccountIdentity, + input: PlatformFailureInput, +): Promise { + syncTrackedAccountsInternal([account], false); + const adapter = getPlatformAdapter(account.platform); + const classification = adapter.classifyFailure(input); + + if (classification === "ok") { + return getAccountHealthSnapshot(account.id); + } + + return await withAccountLock(account.id, async () => { + const record = ensureRecord(account); + const previousSignature = snapshotSignature(record); + const now = Date.now(); + + if (classification === "challenge") { + record.authState = "challenge_required"; + record.probeState = "idle"; + record.authReason = "captcha_gate"; + record.lastAuthFailureAt = now; + record.lastProbeAt = now; + record.nextProbeAt = nextRegularProbeAt(now); + record.suspectedExpiredAt = null; + record.confirmFailureCount = 0; + return commitRecord(record, previousSignature); + } + + if (classification === "network") { + record.probeState = "network_error"; + record.authReason = "network_error"; + record.lastProbeAt = now; + record.nextProbeAt = now + NETWORK_RETRY_MS; + return commitRecord(record, previousSignature); + } + + record.lastAuthFailureAt = now; + record.suspectedExpiredAt = record.suspectedExpiredAt ?? now; + if (record.authState === "active") { + record.authState = "expiring_soon"; + } + record.nextProbeAt = now; + commitRecord(record, previousSignature); + + return await performProbeLocked(account, { + trigger: "confirm", + allowSilentRefresh: false, + }); + }); +} diff --git a/apps/desktop-client/src/main/adapters/base.ts b/apps/desktop-client/src/main/adapters/base.ts index f541311..908526c 100644 --- a/apps/desktop-client/src/main/adapters/base.ts +++ b/apps/desktop-client/src/main/adapters/base.ts @@ -1,5 +1,6 @@ import type { DesktopArticleContent, JsonValue } from "@geo/shared-types"; import type { Session, WebContentsView } from "electron"; +import type { Browser as PlaywrightBrowser, Page as PlaywrightPage } from "playwright-core"; export type AdapterTaskPhase = "initial" | "resume"; export type AdapterCompletionStatus = "succeeded" | "failed" | "unknown"; @@ -10,6 +11,10 @@ export interface AdapterContext { accountId: string; session: Session; view: WebContentsView | null; + playwright: { + browser: PlaywrightBrowser; + page: PlaywrightPage; + } | null; signal: AbortSignal; phase: AdapterTaskPhase; reportProgress(stage: string): void; diff --git a/apps/desktop-client/src/main/auth-types.ts b/apps/desktop-client/src/main/auth-types.ts new file mode 100644 index 0000000..43c55b5 --- /dev/null +++ b/apps/desktop-client/src/main/auth-types.ts @@ -0,0 +1,61 @@ +export type AccountAuthState = + | "unknown" + | "active" + | "expiring_soon" + | "challenge_required" + | "expired" + | "revoked"; + +export type AccountProbeState = + | "idle" + | "probing" + | "network_error"; + +export type AccountAuthReason = + | "bind_success" + | "probe_success" + | "login_redirect" + | "http_401" + | "captcha_gate" + | "missing_partition" + | "silent_refresh_failed" + | "manual_unbind" + | "too_many_failures" + | "network_error" + | null; + +export interface AccountHealthProfile { + displayName: string | null; + avatarUrl: string | null; + subjectUid: string | null; +} + +export interface AccountHealthSnapshot { + accountId: string; + platform: string; + partition: string; + authState: AccountAuthState; + probeState: AccountProbeState; + authReason: AccountAuthReason; + lastVerifiedAt: number | null; + lastProbeAt: number | null; + nextProbeAt: number | null; + lastAuthFailureAt: number | null; + profile: AccountHealthProfile | null; + sessionFingerprint: string | null; +} + +export interface AuthProbeResult { + verdict: "active" | "expired" | "challenge_required" | "network_error"; + reason?: AccountAuthReason; + profile?: AccountHealthProfile | null; + sessionFingerprint?: string | null; + evidence?: Record; +} + +export type PlatformFailureClassification = + | "auth_failure" + | "challenge" + | "network" + | "ok"; + diff --git a/apps/desktop-client/src/main/bootstrap.ts b/apps/desktop-client/src/main/bootstrap.ts index 9a4c85f..76c3617 100644 --- a/apps/desktop-client/src/main/bootstrap.ts +++ b/apps/desktop-client/src/main/bootstrap.ts @@ -4,13 +4,17 @@ import { join } from "node:path"; import { BrowserWindow, WebContentsView, app, ipcMain, nativeTheme } from "electron/main"; import type { BrowserWindow as ElectronBrowserWindow, + WebContents as ElectronWebContents, WebContentsView as ElectronWebContentsView, } from "electron/main"; import type { DesktopAccountInfo, DesktopRuntimeSessionSyncRequest } from "@geo/shared-types"; +import { initAccountHealth } from "./account-health"; import { bindPublishAccount, openPublishAccountConsole } from "./account-binder"; import { initProcessMetricsSampler } from "./process-metrics"; +import { getPlaywrightCDPPort, startHiddenPlaywrightReaper } from "./playwright-cdp"; import { registerRendererDevtoolsProxyTarget } from "./renderer-devtools-proxy"; +import { onRuntimeInvalidated } from "./runtime-events"; import { refreshRuntimeAccounts, releaseRuntimeSession, @@ -32,6 +36,7 @@ app.commandLine.appendSwitch( ); app.commandLine.appendSwitch("disable-quic"); app.commandLine.appendSwitch("disable-background-networking"); +app.commandLine.appendSwitch("remote-debugging-port", String(getPlaywrightCDPPort())); app.userAgentFallback = STANDARD_USER_AGENT; console.info("[desktop-main] boot", { @@ -50,6 +55,7 @@ process.on("uncaughtException", (error) => { }); let mainWindow: ElectronBrowserWindow | null = null; +let mainRendererContents: ElectronWebContents | null = null; let quitReleaseInFlight = false; function rendererURL(): string | null { @@ -77,9 +83,15 @@ async function mountRendererView(window: ElectronBrowserWindow): Promise { view.webContents.setWindowOpenHandler(() => ({ action: "deny" })); registerRendererDevtoolsProxyTarget(view.webContents); + mainRendererContents = view.webContents; window.contentView.addChildView(view); syncViewBounds(window, view); window.on("resize", () => syncViewBounds(window, view)); + window.on("closed", () => { + if (mainRendererContents === view.webContents) { + mainRendererContents = null; + } + }); const devServerURL = rendererURL(); if (devServerURL) { @@ -189,10 +201,18 @@ if (!initSingleInstance(() => { app.whenReady().then(async () => { registerBridgeHandlers(); await initSessionRegistry(); + initAccountHealth(); initTransport(); initScheduler(); initProcessMetricsSampler(); startHotViewReaper(); + startHiddenPlaywrightReaper(); + onRuntimeInvalidated((event) => { + if (!mainRendererContents || mainRendererContents.isDestroyed()) { + return; + } + mainRendererContents.send("desktop:runtime-invalidated", event); + }); mainWindow = await createMainWindow(); initTray(() => { if (!mainWindow) { diff --git a/apps/desktop-client/src/main/lease-manager.ts b/apps/desktop-client/src/main/lease-manager.ts index 60bb581..c371fb5 100644 --- a/apps/desktop-client/src/main/lease-manager.ts +++ b/apps/desktop-client/src/main/lease-manager.ts @@ -1,6 +1,8 @@ interface LeaseSnapshot { activeLeaseId: string | null; activeTaskId: string | null; + activeTaskIds: string[]; + activeCount: number; attemptId: string | null; startedAt: number | null; leaseExpiresAt: number | null; @@ -13,11 +15,16 @@ interface ActiveLeaseState { taskId: string; attemptId: string | null; leaseExpiresAt: number | null; + startedAt: number; + lastExtendedAt: number | null; } +const activeLeases = new Map(); const leaseState: LeaseSnapshot = { activeLeaseId: null, activeTaskId: null, + activeTaskIds: [], + activeCount: 0, attemptId: null, startedAt: null, leaseExpiresAt: null, @@ -45,8 +52,11 @@ export function setActiveLease(input: { leaseExpiresAt?: number | string | null; } | null): void { if (!input) { + activeLeases.clear(); leaseState.activeLeaseId = null; leaseState.activeTaskId = null; + leaseState.activeTaskIds = []; + leaseState.activeCount = 0; leaseState.attemptId = null; leaseState.startedAt = null; leaseState.leaseExpiresAt = null; @@ -59,28 +69,51 @@ export function setActiveLease(input: { taskId: input.taskId, attemptId: input.attemptId ?? null, leaseExpiresAt: normalizeLeaseExpiresAt(input.leaseExpiresAt), + startedAt: Date.now(), + lastExtendedAt: null, }; - leaseState.activeLeaseId = normalized.taskId; - leaseState.activeTaskId = normalized.taskId; - leaseState.attemptId = normalized.attemptId; - leaseState.startedAt = Date.now(); - leaseState.leaseExpiresAt = normalized.leaseExpiresAt; + activeLeases.set(normalized.taskId, normalized); + syncLeaseSnapshot(); leaseState.lastOutcome = "leased"; } -export function noteLeaseExtended(leaseExpiresAt?: number | string | null): void { +export function noteLeaseExtended( + leaseExpiresAt?: number | string | null, + taskId?: string, +): void { + if (taskId) { + const active = activeLeases.get(taskId); + if (active) { + active.leaseExpiresAt = normalizeLeaseExpiresAt(leaseExpiresAt) ?? active.leaseExpiresAt; + active.lastExtendedAt = Date.now(); + activeLeases.set(taskId, active); + } + } else { + const first = activeLeases.values().next().value as ActiveLeaseState | undefined; + if (first) { + first.leaseExpiresAt = normalizeLeaseExpiresAt(leaseExpiresAt) ?? first.leaseExpiresAt; + first.lastExtendedAt = Date.now(); + activeLeases.set(first.taskId, first); + } + } + + syncLeaseSnapshot(); leaseState.lastExtendedAt = Date.now(); - leaseState.leaseExpiresAt = normalizeLeaseExpiresAt(leaseExpiresAt) ?? leaseState.leaseExpiresAt; leaseState.lastOutcome = "extended"; } -export function noteLeaseReleased(reason: Exclude): void { - leaseState.activeLeaseId = null; - leaseState.activeTaskId = null; - leaseState.attemptId = null; - leaseState.startedAt = null; - leaseState.leaseExpiresAt = null; +export function noteLeaseReleased( + reason: Exclude, + taskId?: string, +): void { + if (taskId) { + activeLeases.delete(taskId); + } else { + activeLeases.clear(); + } + + syncLeaseSnapshot(); leaseState.lastReleasedAt = Date.now(); leaseState.lastOutcome = reason; } @@ -88,3 +121,15 @@ export function noteLeaseReleased(reason: Exclude; + questionCooldowns: Record; + platformCooldowns: Record; + lastHydratedAt: number; +} + +export interface MonitorSchedulerSnapshot { + hydratedAt: number | null; + queueDepth: number; + activeCount: number; + leasingCount: number; + queuedCount: number; + questionCooldownCount: number; + platformCooldownCount: number; + tasks: MonitorSchedulerTaskRecord[]; +} + +export interface MonitorSchedulerSelection { + taskId: string; + routing: MonitorSchedulerRouting; +} + +interface MonitorSchedulerSelectionOptions { + now?: number; + maxConcurrency: number; + activePlatforms: ReadonlySet; + activeQuestionKeys: ReadonlySet; + activeCount: number; +} + +interface MonitorTaskMetadata { + title: string | null; + businessDate: string | null; + questionKey: string | null; + questionText: string | null; +} + +const schedulerStateVersion = 1; +const schedulerQuestionCooldownMs = 45_000; +const schedulerPlatformCooldownMs = 5_000; +const schedulerRestartRecoveryDelayMs = 90_000; +const schedulerUnknownQuestionParallelismFloor = 1; +const schedulerMaxLeaseMisses = 4; + +let persistedStateCache: PersistedMonitorSchedulerState | null = null; + +function persistedStatePath(): string { + return join(app.getPath("userData"), "desktop-monitor-scheduler.json"); +} + +function defaultPersistedState(): PersistedMonitorSchedulerState { + return { + version: schedulerStateVersion, + tasks: {}, + questionCooldowns: {}, + platformCooldowns: {}, + lastHydratedAt: 0, + }; +} + +function clonePersistedState(state: PersistedMonitorSchedulerState): PersistedMonitorSchedulerState { + return { + version: state.version, + tasks: Object.fromEntries( + Object.entries(state.tasks).map(([taskId, task]) => [taskId, { ...task }]), + ), + questionCooldowns: { ...state.questionCooldowns }, + platformCooldowns: { ...state.platformCooldowns }, + lastHydratedAt: state.lastHydratedAt, + }; +} + +function readPersistedState(): PersistedMonitorSchedulerState { + if (persistedStateCache) { + return persistedStateCache; + } + + try { + const parsed = JSON.parse(readFileSync(persistedStatePath(), "utf8")) as PersistedMonitorSchedulerState; + persistedStateCache = normalizePersistedState(parsed); + } catch { + persistedStateCache = defaultPersistedState(); + } + + return persistedStateCache; +} + +function writePersistedState(next: PersistedMonitorSchedulerState): void { + persistedStateCache = next; + const target = persistedStatePath(); + mkdirSync(dirname(target), { recursive: true }); + writeFileSync(target, JSON.stringify(next, null, 2), "utf8"); +} + +function mutatePersistedState( + mutator: (draft: PersistedMonitorSchedulerState) => void, +): PersistedMonitorSchedulerState { + const draft = clonePersistedState(readPersistedState()); + mutator(draft); + prunePersistedState(draft, Date.now()); + writePersistedState(draft); + return draft; +} + +function normalizePersistedState(input: PersistedMonitorSchedulerState | null | undefined): PersistedMonitorSchedulerState { + const next = defaultPersistedState(); + if (!input || typeof input !== "object") { + return next; + } + + next.version = typeof input.version === "number" ? input.version : schedulerStateVersion; + next.lastHydratedAt = typeof input.lastHydratedAt === "number" ? input.lastHydratedAt : 0; + + if (input.tasks && typeof input.tasks === "object") { + for (const [taskId, value] of Object.entries(input.tasks)) { + if (!value || typeof value !== "object") { + continue; + } + + const task = value as Partial; + if (typeof task.taskId !== "string" || task.taskId !== taskId || typeof task.platform !== "string") { + continue; + } + + next.tasks[taskId] = { + taskId, + jobId: typeof task.jobId === "string" ? task.jobId : "", + accountId: typeof task.accountId === "string" ? task.accountId : "", + clientId: typeof task.clientId === "string" ? task.clientId : "", + platform: task.platform, + routing: task.routing === "db-recovery" ? "db-recovery" : "rabbitmq-primary", + state: task.state === "active" || task.state === "leasing" ? task.state : "queued", + title: typeof task.title === "string" ? task.title : null, + businessDate: normalizeBusinessDate(task.businessDate), + questionKey: normalizeOptionalString(task.questionKey), + questionText: normalizeOptionalString(task.questionText), + updatedAt: typeof task.updatedAt === "number" ? task.updatedAt : Date.now(), + enqueuedAt: typeof task.enqueuedAt === "number" ? task.enqueuedAt : Date.now(), + availableAt: typeof task.availableAt === "number" ? task.availableAt : Date.now(), + lastSeenAt: typeof task.lastSeenAt === "number" ? task.lastSeenAt : Date.now(), + lastLeaseAttemptAt: typeof task.lastLeaseAttemptAt === "number" ? task.lastLeaseAttemptAt : null, + lastStartedAt: typeof task.lastStartedAt === "number" ? task.lastStartedAt : null, + leaseMisses: typeof task.leaseMisses === "number" ? Math.max(0, task.leaseMisses) : 0, + }; + } + } + + if (input.questionCooldowns && typeof input.questionCooldowns === "object") { + for (const [key, value] of Object.entries(input.questionCooldowns)) { + if (typeof value === "number" && value > 0) { + next.questionCooldowns[key] = value; + } + } + } + + if (input.platformCooldowns && typeof input.platformCooldowns === "object") { + for (const [key, value] of Object.entries(input.platformCooldowns)) { + if (typeof value === "number" && value > 0) { + next.platformCooldowns[key] = value; + } + } + } + + prunePersistedState(next, Date.now()); + return next; +} + +function prunePersistedState(state: PersistedMonitorSchedulerState, now: number): void { + const currentBusinessDate = currentMonitoringBusinessDate(now); + + for (const [taskId, task] of Object.entries(state.tasks)) { + if (task.businessDate && task.businessDate < currentBusinessDate) { + delete state.tasks[taskId]; + continue; + } + + if (task.state === "active" || task.state === "leasing") { + task.state = "queued"; + task.availableAt = Math.max(task.availableAt, now + schedulerRestartRecoveryDelayMs); + } + + if (task.leaseMisses >= schedulerMaxLeaseMisses) { + delete state.tasks[taskId]; + } + } + + for (const [questionKey, until] of Object.entries(state.questionCooldowns)) { + if (until <= now) { + delete state.questionCooldowns[questionKey]; + } + } + + for (const [platform, until] of Object.entries(state.platformCooldowns)) { + if (until <= now) { + delete state.platformCooldowns[platform]; + } + } +} + +export function initMonitorScheduler(): void { + const now = Date.now(); + const next = clonePersistedState(readPersistedState()); + prunePersistedState(next, now); + next.lastHydratedAt = now; + writePersistedState(next); +} + +export function enqueueMonitorTaskFromEvent( + event: DesktopTaskEventMessage, + routing: MonitorSchedulerRouting, +): void { + const metadata = metadataFromEvent(event); + const now = Date.now(); + + mutatePersistedState((draft) => { + const existing = draft.tasks[event.task_id]; + draft.tasks[event.task_id] = { + taskId: event.task_id, + jobId: event.job_id, + accountId: event.target_account_id, + clientId: event.target_client_id, + platform: event.platform, + routing, + state: existing?.state === "active" ? "active" : "queued", + title: metadata.title ?? existing?.title ?? null, + businessDate: metadata.businessDate ?? existing?.businessDate ?? null, + questionKey: metadata.questionKey ?? existing?.questionKey ?? null, + questionText: metadata.questionText ?? existing?.questionText ?? null, + updatedAt: parseTimestamp(event.updated_at) ?? 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 noteMonitorTaskLeased( + task: DesktopTaskInfo, + routing: MonitorSchedulerRouting, +): void { + const now = Date.now(); + const metadata = metadataFromPayload(task.payload ?? null); + + mutatePersistedState((draft) => { + const existing = draft.tasks[task.id]; + draft.tasks[task.id] = { + taskId: task.id, + jobId: task.job_id, + accountId: task.target_account_id, + clientId: task.target_client_id, + platform: task.platform, + routing, + state: "active", + title: metadata.title ?? existing?.title ?? null, + businessDate: metadata.businessDate ?? existing?.businessDate ?? null, + questionKey: metadata.questionKey ?? existing?.questionKey ?? null, + questionText: metadata.questionText ?? existing?.questionText ?? null, + updatedAt: parseTimestamp(task.updated_at) ?? now, + enqueuedAt: existing?.enqueuedAt ?? now, + availableAt: now, + lastSeenAt: now, + lastLeaseAttemptAt: now, + lastStartedAt: now, + leaseMisses: existing?.leaseMisses ?? 0, + }; + }); +} + +export function noteMonitorTaskLeaseMiss(taskId: string): void { + const now = Date.now(); + + mutatePersistedState((draft) => { + const existing = draft.tasks[taskId]; + if (!existing) { + return; + } + + const leaseMisses = existing.leaseMisses + 1; + if (leaseMisses >= schedulerMaxLeaseMisses) { + delete draft.tasks[taskId]; + return; + } + + existing.state = "queued"; + existing.leaseMisses = leaseMisses; + existing.lastLeaseAttemptAt = now; + existing.availableAt = now + leaseMissBackoffMs(leaseMisses); + existing.lastSeenAt = now; + }); +} + +export function noteMonitorTaskCompleted(taskId: string): void { + const now = Date.now(); + + mutatePersistedState((draft) => { + const existing = draft.tasks[taskId]; + if (!existing) { + return; + } + + if (existing.questionKey) { + draft.questionCooldowns[existing.questionKey] = now + schedulerQuestionCooldownMs; + } + draft.platformCooldowns[existing.platform] = now + schedulerPlatformCooldownMs; + delete draft.tasks[taskId]; + }); +} + +export function noteMonitorTaskCanceled(taskId: string): void { + mutatePersistedState((draft) => { + delete draft.tasks[taskId]; + }); +} + +export function selectNextMonitorTask( + options: MonitorSchedulerSelectionOptions, +): MonitorSchedulerSelection | null { + const now = options.now ?? Date.now(); + const draft = clonePersistedState(readPersistedState()); + prunePersistedState(draft, now); + + if (options.activeCount >= options.maxConcurrency) { + writePersistedState(draft); + return null; + } + + const candidates = Object.values(draft.tasks) + .filter((task) => task.state === "queued" && task.availableAt <= now) + .sort((left, right) => { + if (left.availableAt !== right.availableAt) { + return left.availableAt - right.availableAt; + } + if (left.enqueuedAt !== right.enqueuedAt) { + return left.enqueuedAt - right.enqueuedAt; + } + return left.updatedAt - right.updatedAt; + }); + + for (const candidate of candidates) { + if (options.activePlatforms.has(candidate.platform)) { + continue; + } + + const platformCooldown = draft.platformCooldowns[candidate.platform] ?? 0; + if (platformCooldown > now) { + continue; + } + + if (!candidate.questionKey && options.activeCount >= schedulerUnknownQuestionParallelismFloor) { + continue; + } + + if (candidate.questionKey) { + if (options.activeQuestionKeys.has(candidate.questionKey)) { + continue; + } + const questionCooldown = draft.questionCooldowns[candidate.questionKey] ?? 0; + if (questionCooldown > now) { + continue; + } + } + + candidate.state = "leasing"; + candidate.lastLeaseAttemptAt = now; + candidate.lastSeenAt = now; + writePersistedState(draft); + return { + taskId: candidate.taskId, + routing: candidate.routing, + }; + } + + writePersistedState(draft); + return null; +} + +export function getMonitorSchedulerSnapshot(): MonitorSchedulerSnapshot { + const state = readPersistedState(); + const tasks = Object.values(state.tasks).sort((left, right) => left.enqueuedAt - right.enqueuedAt); + return { + hydratedAt: state.lastHydratedAt || null, + queueDepth: tasks.length, + activeCount: tasks.filter((task) => task.state === "active").length, + leasingCount: tasks.filter((task) => task.state === "leasing").length, + queuedCount: tasks.filter((task) => task.state === "queued").length, + questionCooldownCount: Object.keys(state.questionCooldowns).length, + platformCooldownCount: Object.keys(state.platformCooldowns).length, + tasks, + }; +} + +function leaseMissBackoffMs(leaseMisses: number): number { + if (leaseMisses <= 1) { + return 20_000; + } + if (leaseMisses === 2) { + return 60_000; + } + return 3 * 60_000; +} + +function metadataFromEvent(event: DesktopTaskEventMessage): MonitorTaskMetadata { + return { + title: normalizeOptionalString(event.title), + businessDate: normalizeBusinessDate(event.business_date), + questionKey: normalizeOptionalString(event.scheduler_group_key), + questionText: normalizeOptionalString(event.question_text), + }; +} + +function metadataFromPayload(payload: Record | null): MonitorTaskMetadata { + if (!payload) { + return { + title: null, + businessDate: null, + questionKey: null, + questionText: null, + }; + } + + const title = firstString(payload, ["title", "question_title", "questionTitle"]); + const businessDate = normalizeBusinessDate( + firstString(payload, ["business_date", "businessDate", "metric_date", "date"]), + ); + const questionText = firstString(payload, ["question_text", "questionText", "query", "question", "prompt", "content"]); + const questionHash = firstString(payload, ["question_hash", "questionHash"]); + const questionID = firstString(payload, ["question_id", "questionId"]); + + return { + title, + businessDate, + questionKey: normalizeOptionalString(firstString(payload, ["scheduler_group_key", "schedulerGroupKey"])) + ?? normalizeOptionalString(questionHash) + ?? (questionID ? `qid:${questionID}` : questionText ? `qtxt:${digestText(questionText)}` : null), + questionText, + }; +} + +function firstString( + payload: Record, + keys: string[], +): string | null { + for (const key of keys) { + const value = payload[key]; + if (typeof value === "string") { + const normalized = value.trim(); + if (normalized) { + return normalized; + } + continue; + } + if (typeof value === "number" && Number.isFinite(value)) { + return String(value); + } + } + return null; +} + +function currentMonitoringBusinessDate(now: number): string { + const parts = new Intl.DateTimeFormat("en-US", { + timeZone: "Asia/Shanghai", + year: "numeric", + month: "2-digit", + day: "2-digit", + }).formatToParts(now); + + const year = parts.find((part) => part.type === "year")?.value ?? ""; + const month = parts.find((part) => part.type === "month")?.value ?? ""; + const day = parts.find((part) => part.type === "day")?.value ?? ""; + return `${year}-${month}-${day}`; +} + +function normalizeBusinessDate(value: unknown): string | null { + if (typeof value !== "string") { + return null; + } + const normalized = value.trim(); + return /^\d{4}-\d{2}-\d{2}$/.test(normalized) ? normalized : null; +} + +function normalizeOptionalString(value: unknown): string | null { + if (typeof value !== "string") { + return null; + } + const normalized = value.trim(); + return normalized || null; +} + +function digestText(value: string): string { + return createHash("sha1").update(value).digest("hex"); +} + +function parseTimestamp(value: string | null | undefined): number | null { + if (!value) { + return null; + } + const timestamp = Date.parse(value); + return Number.isNaN(timestamp) ? null : timestamp; +} diff --git a/apps/desktop-client/src/main/platform-auth-adapters.ts b/apps/desktop-client/src/main/platform-auth-adapters.ts new file mode 100644 index 0000000..3eeba43 --- /dev/null +++ b/apps/desktop-client/src/main/platform-auth-adapters.ts @@ -0,0 +1,102 @@ +import { aiPlatformCatalog } from "@geo/shared-types"; + +import { + probePublishAccountSession, + silentRefreshPublishAccountSession, + type PublishAccountIdentity, +} from "./account-binder"; +import type { + AuthProbeResult, + PlatformFailureClassification, +} from "./auth-types"; + +export interface PlatformFailureInput { + summary?: string | null; + error?: Record | null; + message?: string | null; +} + +export interface PlatformAdapter { + probe(account: PublishAccountIdentity, partition: string): Promise; + silentRefresh(account: PublishAccountIdentity, partition: string): Promise; + classifyFailure(input: PlatformFailureInput): PlatformFailureClassification; +} + +function normalizeFailureText(input: PlatformFailureInput): string { + const values = [ + input.summary, + input.message, + typeof input.error?.code === "string" ? input.error.code : null, + typeof input.error?.message === "string" ? input.error.message : null, + typeof input.error?.detail === "string" ? input.error.detail : null, + typeof input.error?.verify_scene === "string" ? input.error.verify_scene : null, + ]; + + return values + .filter((value): value is string => typeof value === "string" && Boolean(value.trim())) + .join(" | ") + .toLowerCase(); +} + +function classifyFailure(input: PlatformFailureInput): PlatformFailureClassification { + const text = normalizeFailureText(input); + if (!text) { + return "ok"; + } + + if ( + /(verify_scene|verification required|captcha|challenge|验证码|滑块|人机验证|安全验证|请完成验证|短信验证|扫码验证)/i.test(text) + ) { + return "challenge"; + } + + if ( + /(not_logged_in|unauthorized|unauthenticated|login redirect|login required|signin|sign in|expired|401|登录态失效|请先登录|未登录)/i + .test(text) + ) { + return "auth_failure"; + } + + if ( + /(network|timeout|timed out|failed to fetch|econnreset|econnrefused|enotfound|dns|socket hang up|502|503|504)/i + .test(text) + ) { + return "network"; + } + + return "ok"; +} + +const genericAdapter: PlatformAdapter = { + async probe(account, partition) { + return await probePublishAccountSession(account, partition); + }, + async silentRefresh(account, partition) { + return await silentRefreshPublishAccountSession(account, partition); + }, + classifyFailure, +}; + +const adapterRegistry = new Map( + [ + ...aiPlatformCatalog.map((platform) => platform.id), + "toutiaohao", + "baijiahao", + "sohu", + "qiehao", + "zhihu", + "wangyihao", + "jianshu", + "bilibili", + "juejin", + "smzdm", + "weixin_gzh", + "zol", + "dongchedi", + ].map((platform) => [platform, genericAdapter]), +); + +export function getPlatformAdapter(platformId: string): PlatformAdapter { + return adapterRegistry.get(platformId) ?? genericAdapter; +} + diff --git a/apps/desktop-client/src/main/playwright-cdp.ts b/apps/desktop-client/src/main/playwright-cdp.ts new file mode 100644 index 0000000..9abadab --- /dev/null +++ b/apps/desktop-client/src/main/playwright-cdp.ts @@ -0,0 +1,317 @@ +import { BrowserWindow, nativeTheme } from "electron/main"; +import type { BrowserWindow as ElectronBrowserWindow, Session } from "electron/main"; +import type { Browser as PlaywrightBrowser, Page as PlaywrightPage } from "playwright-core"; + +import { STANDARD_USER_AGENT } from "./user-agent"; + +const hiddenPlaywrightIdleTTLms = 5 * 60_000; +const hiddenPlaywrightReaperIntervalMs = 60_000; +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}`; + +interface HiddenPlaywrightRecord { + key: string; + accountId: string; + title: string; + targetURL: string; + window: ElectronBrowserWindow; + browser: PlaywrightBrowser; + page: PlaywrightPage; + retainCount: number; + activatedAt: number; + lastReleasedAt: number | null; +} + +export interface HiddenPlaywrightLease { + accountId: string; + browser: PlaywrightBrowser; + page: PlaywrightPage; + release(): Promise; +} + +const records = new Map(); +let reaperHandle: ReturnType | null = null; +let connectPromise: Promise | null = null; +let connectedBrowser: PlaywrightBrowser | null = null; + +export function getPlaywrightCDPPort(): number { + return cdpPort; +} + +export function startHiddenPlaywrightReaper(): void { + if (reaperHandle) { + return; + } + + reaperHandle = setInterval(() => { + reapIdleHiddenPlaywrightPages(); + }, hiddenPlaywrightReaperIntervalMs); + reaperHandle.unref?.(); +} + +export async function retainHiddenPlaywrightPage(options: { + accountId: string; + session: Session; + targetURL: string; + title: string; +}): Promise { + startHiddenPlaywrightReaper(); + + const key = hiddenPlaywrightKey(options.accountId, options.targetURL); + const existing = records.get(key); + if (existing && !existing.window.isDestroyed() && !existing.page.isClosed()) { + await ensureRecordTarget(existing, options.targetURL); + existing.retainCount += 1; + existing.activatedAt = Date.now(); + existing.lastReleasedAt = null; + return createLease(existing); + } + + const browser = await connectBrowserOverCDP(); + const knownPages = new Set(defaultContext(browser).pages()); + const window = createHiddenWindow(options.title, options.session); + const page = await waitForNewPage(browser, knownPages); + + page.setDefaultTimeout(defaultPageTimeoutMs); + page.setDefaultNavigationTimeout(defaultPageTimeoutMs); + await page.goto(options.targetURL, { + timeout: defaultPageTimeoutMs, + waitUntil: "domcontentloaded", + }); + + const record: HiddenPlaywrightRecord = { + key, + accountId: options.accountId, + title: options.title, + targetURL: options.targetURL, + window, + browser, + page, + retainCount: 1, + activatedAt: Date.now(), + lastReleasedAt: null, + }; + + window.once("closed", () => { + records.delete(key); + }); + page.once("close", () => { + records.delete(key); + }); + + records.set(key, record); + return createLease(record); +} + +export async function releaseHiddenPlaywrightPage(accountId: string, targetURL: string): Promise { + const key = hiddenPlaywrightKey(accountId, targetURL); + const record = records.get(key); + if (!record) { + return; + } + + if (record.retainCount > 0) { + record.retainCount -= 1; + } + if (record.retainCount === 0) { + record.lastReleasedAt = Date.now(); + } +} + +export function reapIdleHiddenPlaywrightPages( + now = Date.now(), + idleTTLms = hiddenPlaywrightIdleTTLms, +): string[] { + const reclaimed: string[] = []; + + for (const [key, record] of records.entries()) { + if (record.retainCount > 0) { + continue; + } + + const idleSince = record.lastReleasedAt ?? record.activatedAt; + if (now - idleSince < idleTTLms) { + continue; + } + + closeRecord(record); + records.delete(key); + reclaimed.push(key); + } + + return reclaimed; +} + +export function getHiddenPlaywrightSnapshot() { + return { + cdpPort, + cdpEndpointURL, + connected: Boolean(connectedBrowser?.isConnected()), + pageCount: records.size, + idleTTLms: hiddenPlaywrightIdleTTLms, + records: [...records.values()].map((record) => ({ + accountId: record.accountId, + title: record.title, + targetURL: record.targetURL, + retainCount: record.retainCount, + activatedAt: record.activatedAt, + lastReleasedAt: record.lastReleasedAt, + windowDestroyed: record.window.isDestroyed(), + pageClosed: record.page.isClosed(), + })), + }; +} + +function createLease(record: HiddenPlaywrightRecord): HiddenPlaywrightLease { + return { + accountId: record.accountId, + browser: record.browser, + page: record.page, + release: () => releaseHiddenPlaywrightPage(record.accountId, record.targetURL), + }; +} + +function createHiddenWindow(title: string, session: Session): ElectronBrowserWindow { + const window = new BrowserWindow({ + show: false, + width: 1320, + height: 900, + autoHideMenuBar: true, + backgroundColor: nativeTheme.shouldUseDarkColors ? "#15191a" : "#f4f1ea", + title, + webPreferences: { + session, + sandbox: true, + contextIsolation: true, + nodeIntegration: false, + backgroundThrottling: false, + }, + }); + + window.webContents.setUserAgent(STANDARD_USER_AGENT); + void window.loadURL("about:blank"); + return window; +} + +async function ensureRecordTarget(record: HiddenPlaywrightRecord, targetURL: string): Promise { + record.targetURL = targetURL; + + const currentURL = safePageURL(record.page); + if (currentURL === targetURL) { + return; + } + + await record.page.goto(targetURL, { + timeout: defaultPageTimeoutMs, + waitUntil: "domcontentloaded", + }); +} + +async function connectBrowserOverCDP(): Promise { + if (connectedBrowser?.isConnected()) { + return connectedBrowser; + } + if (connectPromise) { + return connectPromise; + } + + connectPromise = (async () => { + await waitForCDPEndpoint(); + const { chromium } = await import("playwright-core"); + const browser = await chromium.connectOverCDP(cdpEndpointURL, { + timeout: defaultConnectTimeoutMs, + }); + + browser.on("disconnected", () => { + connectedBrowser = null; + connectPromise = null; + }); + + connectedBrowser = browser; + return browser; + })(); + + try { + return await connectPromise; + } finally { + if (!connectedBrowser?.isConnected()) { + connectPromise = null; + } + } +} + +async function waitForCDPEndpoint(timeoutMs = defaultConnectTimeoutMs): Promise { + const startedAt = Date.now(); + while (Date.now() - startedAt < timeoutMs) { + try { + const response = await fetch(`${cdpEndpointURL}/json/version`); + if (response.ok) { + return; + } + } catch { + // Ignore and continue polling. + } + + await wait(200); + } + + throw new Error("playwright_cdp_endpoint_unavailable"); +} + +async function waitForNewPage( + browser: PlaywrightBrowser, + knownPages: Set, +): Promise { + const context = defaultContext(browser); + const startedAt = Date.now(); + + while (Date.now() - startedAt < defaultPageTimeoutMs) { + const nextPage = context.pages().find((page) => !knownPages.has(page)); + if (nextPage) { + return nextPage; + } + await wait(100); + } + + throw new Error("playwright_hidden_page_unavailable"); +} + +function defaultContext(browser: PlaywrightBrowser) { + const context = browser.contexts()[0]; + if (!context) { + throw new Error("playwright_default_context_unavailable"); + } + return context; +} + +function hiddenPlaywrightKey(accountId: string, targetURL: string): string { + return `${accountId}::${targetURL}`; +} + +function safePageURL(page: PlaywrightPage): string { + try { + return page.url(); + } catch { + return ""; + } +} + +function closeRecord(record: HiddenPlaywrightRecord): void { + try { + record.page.close().catch(() => undefined); + } catch { + // Ignore and fall through to window close. + } + + if (!record.window.isDestroyed()) { + record.window.close(); + } +} + +function wait(timeMs: number): Promise { + return new Promise((resolve) => { + setTimeout(resolve, timeMs); + }); +} diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index 5ce0803..ba8bf29 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -11,7 +11,7 @@ import type { JsonValue, LeaseDesktopTaskResponse, } from "@geo/shared-types"; -import { isAIPlatformId } from "@geo/shared-types"; +import { getAIPlatformCatalogItem, isAIPlatformId } from "@geo/shared-types"; import { doubaoAdapter, @@ -22,10 +22,17 @@ import { type PublishAdapter, } from "./adapters"; import { - ensurePublishAccountSessionHandle, - inspectPublishAccountLocalSession, + type PublishAccountIdentity, type PublishAccountProfile, } from "./account-binder"; +import { + ensureAccountReady, + forgetTrackedAccountHealth, + getProjectedAccountHealth, + probeTrackedAccount, + reportAccountFailure, + syncTrackedAccounts, +} from "./account-health"; import { buildAccountIdentityKey, normalizeAccountPlatformUid, @@ -33,6 +40,21 @@ import { } from "../shared/account-identity"; import { releaseHotView, retainHotView } from "./view-pool"; import { getLeaseManagerSnapshot, noteLeaseExtended, noteLeaseReleased, setActiveLease } from "./lease-manager"; +import { + enqueueMonitorTaskFromEvent, + getMonitorSchedulerSnapshot, + initMonitorScheduler, + noteMonitorTaskCanceled, + noteMonitorTaskCompleted, + noteMonitorTaskLeaseMiss, + noteMonitorTaskLeased, + selectNextMonitorTask, +} from "./monitor-scheduler"; +import { + retainHiddenPlaywrightPage, + startHiddenPlaywrightReaper, +} from "./playwright-cdp"; +import { getProcessMetricsSnapshot } from "./process-metrics"; import { initScheduler, noteSchedulerAccountsSync, @@ -47,7 +69,7 @@ import { setSchedulerPhase, setSchedulerQueueDepth, } from "./scheduler"; -import { clearSessionHandle, createSessionHandle, getSessionHandle } from "./session-registry"; +import { clearSessionHandle, createSessionHandle } from "./session-registry"; import { completeDesktopTask, configureTransport, @@ -136,19 +158,26 @@ interface PendingTaskRequest { routing: RuntimeTaskRouting; } +interface ActiveRuntimeExecution { + taskId: string; + kind: "publish" | "monitor"; + platform: string; + accountId: string; + abortController: AbortController; +} + interface RuntimeState { session: DesktopRuntimeSessionSyncRequest | null; client: DesktopClientInfo | null; running: boolean; sseConnected: boolean; - queue: PendingTaskRequest[]; - queuedTaskIds: Set; + publishQueue: PendingTaskRequest[]; + queuedPublishTaskIds: Set; tasks: Map; accounts: DesktopAccountInfo[]; accountProfiles: Map; activity: RuntimeControllerActivity[]; - currentTaskId: string | null; - currentExecutionAbort: AbortController | null; + activeExecutions: Map; leaseInFlight: boolean; heartbeatTimer: ReturnType | null; pullTimer: ReturnType | null; @@ -168,14 +197,13 @@ const state: RuntimeState = { client: null, running: false, sseConnected: false, - queue: [], - queuedTaskIds: new Set(), + publishQueue: [], + queuedPublishTaskIds: new Set(), tasks: new Map(), accounts: [], accountProfiles: new Map(), activity: [], - currentTaskId: null, - currentExecutionAbort: null, + activeExecutions: new Map(), leaseInFlight: false, heartbeatTimer: null, pullTimer: null, @@ -190,6 +218,41 @@ const state: RuntimeState = { activitySeq: 0, }; +function accountIdentityFromDesktopAccount(account: DesktopAccountInfo): PublishAccountIdentity { + return { + id: account.id, + platform: account.platform, + platformUid: normalizeAccountPlatformUid(account.platform_uid) || account.platform_uid, + displayName: account.display_name, + }; +} + +function accountIdentityFromTask(task: RuntimeTaskRecord): PublishAccountIdentity | null { + if (!task.accountId) { + return null; + } + + const matched = state.accounts.find((account) => account.id === task.accountId) ?? null; + if (matched) { + return accountIdentityFromDesktopAccount(matched); + } + + return { + id: task.accountId, + platform: task.platform, + platformUid: "", + displayName: task.accountName || "待同步账号", + }; +} + +function isDefinitiveAuthState(authState: string): boolean { + return ["active", "expired", "challenge_required", "revoked"].includes(authState); +} + +function isoFromTimestamp(value: number | null): string | null { + return value ? new Date(value).toISOString() : null; +} + export function syncRuntimeSession(next: DesktopRuntimeSessionSyncRequest | null): void { const normalized = normalizeSession(next); const changed = !sameSession(state.session, normalized); @@ -256,7 +319,7 @@ export function getRuntimeControllerSnapshot(): RuntimeControllerSnapshot { client: state.client ? { ...state.client } : null, running: state.running, sseConnected: state.sseConnected, - queueDepth: state.queue.length, + queueDepth: localQueueDepth(), accounts: state.accounts.map((item) => ({ ...item })), accountProfiles: Object.fromEntries( [...state.accountProfiles.entries()].map(([accountId, profile]) => [accountId, { ...profile }]), @@ -281,6 +344,8 @@ function startRuntime(): void { } initScheduler(); + initMonitorScheduler(); + startHiddenPlaywrightReaper(); state.running = true; setSchedulerPhase("running"); setSchedulerError(null); @@ -292,16 +357,17 @@ function startRuntime(): void { void sendHeartbeat("startup"); void syncAccounts("startup"); - processQueue(); + pumpExecutionLoop(); } function stopRuntime(): void { state.running = false; state.sseConnected = false; state.leaseInFlight = false; - state.currentExecutionAbort?.abort(); - state.currentExecutionAbort = null; - state.currentTaskId = null; + for (const execution of state.activeExecutions.values()) { + execution.abortController.abort(); + } + state.activeExecutions.clear(); if (state.heartbeatTimer) { clearInterval(state.heartbeatTimer); @@ -319,7 +385,7 @@ function stopRuntime(): void { setTransportSseState("idle"); setSchedulerPhase("idle"); setSchedulerCurrentTask(null); - setSchedulerQueueDepth(0); + setSchedulerQueueDepth(localQueueDepth()); setSchedulerNextHeartbeat(null); setSchedulerNextPull(null); @@ -329,12 +395,12 @@ function stopRuntime(): void { } function clearRuntimeState(): void { - state.queue = []; - state.queuedTaskIds.clear(); + state.publishQueue = []; + state.queuedPublishTaskIds.clear(); state.tasks.clear(); state.accounts = []; state.accountProfiles.clear(); - state.currentTaskId = null; + state.activeExecutions.clear(); state.lastHeartbeatAt = 0; state.lastHeartbeatStatus = "idle"; state.lastPullAt = 0; @@ -342,7 +408,7 @@ function clearRuntimeState(): void { state.lastAccountsSyncAt = 0; state.lastServerTime = null; state.lastError = null; - setSchedulerQueueDepth(0); + setSchedulerQueueDepth(localQueueDepth()); } function scheduleHeartbeatLoop(): void { @@ -365,7 +431,7 @@ function schedulePullLoop(): void { setSchedulerNextPull(Date.now() + pullIntervalMs); state.pullTimer = setInterval(() => { setSchedulerNextPull(Date.now() + pullIntervalMs); - processQueue(); + pumpExecutionLoop(); }, pullIntervalMs); } @@ -478,9 +544,20 @@ function handleTaskEvent(event: DesktopTaskEventMessage): void { state.tasks.set(event.task_id, next); + if (event.kind === "monitor") { + if (event.type === "task_available") { + enqueueMonitorTaskFromEvent(event, "rabbitmq-primary"); + } + if (event.type === "task_completed" || event.type === "task_reconciled" || event.type === "task_canceled") { + noteMonitorTaskCanceled(event.task_id); + } + } + if (event.type === "task_available") { - enqueueTask(event.task_id, "rabbitmq-primary"); - processQueue(); + if (event.kind === "publish") { + enqueuePublishTask(event.task_id, "rabbitmq-primary"); + } + pumpExecutionLoop(); } } @@ -498,7 +575,16 @@ async function sendHeartbeat(source: "startup" | "timer"): Promise { cpu_arch: process.arch, client_version: state.client?.client_version ?? "0.1.0-dev", channel: state.client?.channel ?? "dev", - account_ids: state.accounts.filter((account) => account.health === "live").map((account) => account.id), + account_ids: state.accounts + .filter((account) => + getProjectedAccountHealth({ + accountId: account.id, + platform: account.platform, + health: account.health, + verifiedAt: account.verified_at, + }).health === "live", + ) + .map((account) => account.id), }); state.client = response.client; @@ -547,28 +633,41 @@ async function syncAccounts(source: "startup" | "manual"): Promise { noteSchedulerAccountsSync(); setSchedulerError(null); + const identities = accounts.map((account) => accountIdentityFromDesktopAccount(account)); + syncTrackedAccounts(identities); + if (source === "manual") { + await Promise.allSettled( + identities.map((account) => + probeTrackedAccount(account, { + trigger: "manual", + allowSilentRefresh: true, + }), + ), + ); + } + state.accountProfiles.clear(); const syncedAccounts: Array = await Promise.all(accounts.map(async (account) => { const normalizedPlatformUid = normalizeAccountPlatformUid(account.platform_uid) || account.platform_uid; const isAIPlatform = isAIPlatformId(account.platform); - const identity = { - id: account.id, + const identity = accountIdentityFromDesktopAccount(account); + const projected = getProjectedAccountHealth({ + accountId: account.id, platform: account.platform, - platformUid: normalizedPlatformUid, - displayName: account.display_name, - } as const; - const local = await inspectPublishAccountLocalSession(identity); + health: account.health, + verifiedAt: account.verified_at, + }); - if (local.availability === "missing") { - return null; + if (projected.profile?.subjectUid && projected.profile.displayName) { + state.accountProfiles.set(account.id, { + platformUid: projected.profile.subjectUid, + displayName: projected.profile.displayName, + avatarUrl: projected.profile.avatarUrl, + }); } - if (local.profile) { - state.accountProfiles.set(account.id, local.profile); - } - - const normalizedLocalPlatformUid = local.profile - ? (normalizeAccountPlatformUid(local.profile.platformUid) || local.profile.platformUid) + const normalizedLocalPlatformUid = projected.profile?.subjectUid + ? (normalizeAccountPlatformUid(projected.profile.subjectUid) || projected.profile.subjectUid) : null; const hasPlatformUidMismatch = Boolean( normalizedLocalPlatformUid && !sameAccountPlatformUid(normalizedLocalPlatformUid, normalizedPlatformUid), @@ -585,22 +684,27 @@ async function syncAccounts(source: "startup" | "manual"): Promise { let resolvedAccount: DesktopAccountInfo = { ...account, platform_uid: normalizedPlatformUid, - display_name: local.profile?.displayName ?? account.display_name, - avatar_url: local.profile?.avatarUrl ?? account.avatar_url, - health: local.availability === "expired" ? ("expired" as const) : ("live" as const), + display_name: projected.profile?.displayName ?? account.display_name, + avatar_url: projected.profile?.avatarUrl ?? account.avatar_url, + health: projected.health, + verified_at: isoFromTimestamp(projected.lastVerifiedAt) ?? account.verified_at, }; const shouldRelinkClient = Boolean(currentClientId && resolvedAccount.client_id !== currentClientId); - if (shouldRelinkClient) { + const shouldMirrorHealth = + resolvedAccount.health !== account.health + && isDefinitiveAuthState(projected.authState); + + if (shouldRelinkClient || shouldMirrorHealth) { try { const updated = await upsertDesktopAccount({ platform: resolvedAccount.platform, platform_uid: normalizedPlatformUid, - display_name: local.profile?.displayName ?? resolvedAccount.display_name, - avatar_url: local.profile?.avatarUrl ?? resolvedAccount.avatar_url, + display_name: projected.profile?.displayName ?? resolvedAccount.display_name, + avatar_url: projected.profile?.avatarUrl ?? resolvedAccount.avatar_url, account_fingerprint: resolvedAccount.account_fingerprint ?? normalizedPlatformUid, health: resolvedAccount.health, - verified_at: new Date().toISOString(), + verified_at: isoFromTimestamp(projected.lastVerifiedAt) ?? resolvedAccount.verified_at, tags: resolvedAccount.tags, }); @@ -694,6 +798,7 @@ export async function unbindRuntimeAccount(accountId: string, syncVersion: numbe await deleteDesktopAccount(accountId, syncVersion); await clearSessionHandle(accountId); + forgetTrackedAccountHealth(accountId); state.accounts = state.accounts.filter((item) => item.id !== accountId); state.accountProfiles.delete(accountId); @@ -710,41 +815,74 @@ export async function unbindRuntimeAccount(accountId: string, syncVersion: numbe await syncAccounts("manual"); } -function enqueueTask(taskId: string, routing: RuntimeTaskRouting): void { - if (!taskId || state.currentTaskId === taskId || state.queuedTaskIds.has(taskId)) { +function enqueuePublishTask(taskId: string, routing: RuntimeTaskRouting): void { + if (!taskId || state.activeExecutions.has(taskId) || state.queuedPublishTaskIds.has(taskId)) { return; } - state.queue.push({ taskId, routing }); - state.queuedTaskIds.add(taskId); - setSchedulerQueueDepth(state.queue.length); + state.publishQueue.push({ taskId, routing }); + state.queuedPublishTaskIds.add(taskId); + syncSchedulerSurface(); } -function dequeueTask(): PendingTaskRequest | null { - const next = state.queue.shift() ?? null; +function dequeuePublishTask(): PendingTaskRequest | null { + const next = state.publishQueue.shift() ?? null; if (next) { - state.queuedTaskIds.delete(next.taskId); + state.queuedPublishTaskIds.delete(next.taskId); } - setSchedulerQueueDepth(state.queue.length); + syncSchedulerSurface(); return next; } -function processQueue(): void { - if (!state.running || state.currentTaskId || state.leaseInFlight) { +function pumpExecutionLoop(): void { + if (!state.running || state.leaseInFlight) { return; } - const queued = dequeueTask(); - if (queued) { - void leaseSpecificTask(queued); + syncSchedulerSurface(); + + if (hasActivePublishExecution()) { return; } - void pullNextTask(); + if (state.publishQueue.length > 0) { + if (state.activeExecutions.size === 0) { + const queuedPublish = dequeuePublishTask(); + if (queuedPublish) { + void leaseSpecificTask(queuedPublish, "publish"); + } + } + return; + } + + if (canStartAnotherMonitorExecution()) { + const selected = selectNextMonitorTask({ + maxConcurrency: resolveAdaptiveMonitorConcurrency(), + activePlatforms: activeMonitorPlatforms(), + activeQuestionKeys: activeMonitorQuestionKeys(), + activeCount: activeMonitorCount(), + }); + if (selected) { + void leaseSpecificTask(selected, "monitor"); + return; + } + } + + if (state.activeExecutions.size === 0) { + void pullNextTask("publish"); + return; + } + + if (canStartAnotherMonitorExecution()) { + void pullNextTask("monitor"); + } } -async function leaseSpecificTask(request: PendingTaskRequest): Promise { - if (!canRunLive(state.session) || state.currentTaskId || state.leaseInFlight) { +async function leaseSpecificTask( + request: PendingTaskRequest, + expectedKind: "publish" | "monitor", +): Promise { + if (!canRunLive(state.session) || state.leaseInFlight) { return; } @@ -753,11 +891,18 @@ async function leaseSpecificTask(request: PendingTaskRequest): Promise { try { const leased = await leaseDesktopTask({ task_id: request.taskId }); if (!leased.task) { + if (expectedKind === "monitor") { + noteMonitorTaskLeaseMiss(request.taskId); + } return; } state.leaseInFlight = false; await executeLeasedTask(leased, request.routing); } catch (error) { + if (expectedKind === "monitor") { + noteMonitorTaskLeaseMiss(request.taskId); + } + if (isApiClientError(error, 401)) { handleAuthExpired(error); return; @@ -769,22 +914,24 @@ async function leaseSpecificTask(request: PendingTaskRequest): Promise { } finally { if (state.leaseInFlight) { state.leaseInFlight = false; - if (state.queue.length > 0) { - processQueue(); + syncSchedulerSurface(); + if (state.publishQueue.length > 0 || localQueueDepth() > 0) { + pumpExecutionLoop(); } } } } -async function pullNextTask(): Promise { - if (!canRunLive(state.session) || state.currentTaskId || state.leaseInFlight) { +async function pullNextTask(kind: "publish" | "monitor"): Promise { + if (!canRunLive(state.session) || state.leaseInFlight) { return; } state.leaseInFlight = true; + let leasedTask = false; try { - const leased = await leaseDesktopTask({}); + const leased = await leaseDesktopTask({ kind }); state.lastPullAt = Date.now(); state.lastPullStatus = "success"; noteTransportPull(true); @@ -794,6 +941,7 @@ async function pullNextTask(): Promise { return; } + leasedTask = true; state.leaseInFlight = false; await executeLeasedTask(leased, "db-recovery"); } catch (error) { @@ -812,6 +960,11 @@ async function pullNextTask(): Promise { recordActivity("warn", "兜底拉取失败", state.lastError); } finally { state.leaseInFlight = false; + syncSchedulerSurface(); + + if (!leasedTask && kind === "publish" && canStartAnotherMonitorExecution()) { + void pullNextTask("monitor"); + } } } @@ -825,8 +978,6 @@ async function executeLeasedTask( const task = leased.task; - state.currentTaskId = task.id; - setSchedulerCurrentTask(task.id); setSchedulerPhase("running"); setActiveLease({ @@ -849,7 +1000,18 @@ async function executeLeasedTask( ); const abortController = new AbortController(); - state.currentExecutionAbort = abortController; + state.activeExecutions.set(taskRecord.id, { + taskId: taskRecord.id, + kind: taskRecord.kind, + platform: taskRecord.platform, + accountId: taskRecord.accountId, + abortController, + }); + syncSchedulerSurface(); + + if (task.kind === "monitor") { + noteMonitorTaskLeased(task, routing); + } const extendHandle = setInterval(() => { void extendActiveLease(taskRecord.id, leased.lease_token as string); @@ -864,7 +1026,7 @@ async function executeLeasedTask( error: execution.error, }); - noteLeaseReleased(execution.status === "failed" ? "failed" : "completed"); + noteLeaseReleased(execution.status === "failed" ? "failed" : "completed", taskRecord.id); upsertTaskFromInfo(completed, { routing, summary: execution.summary, @@ -881,9 +1043,28 @@ async function executeLeasedTask( : "任务执行成功", `${taskRecord.title} ${execution.summary}`, ); + + if (task.kind === "monitor") { + noteMonitorTaskCompleted(taskRecord.id); + } } catch (error) { const failureMessage = errorMessage(error); const structuredError = toStructuredError(error); + const accountIdentity = accountIdentityFromTask(taskRecord); + + if (accountIdentity) { + await reportAccountFailure(accountIdentity, { + summary: failureMessage, + error: structuredError, + }).catch((reportError) => { + console.warn("[desktop-runtime] thrown failure report failed", { + taskId: taskRecord.id, + accountId: accountIdentity.id, + platform: accountIdentity.platform, + message: reportError instanceof Error ? reportError.message : String(reportError), + }); + }); + } try { const completed = await completeDesktopTask(taskRecord.id, { @@ -909,15 +1090,18 @@ async function executeLeasedTask( } } - noteLeaseReleased("failed"); + noteLeaseReleased("failed", taskRecord.id); recordActivity("danger", "任务执行失败", `${taskRecord.title} 执行失败：${failureMessage || "未知错误"}`); + + if (task.kind === "monitor") { + noteMonitorTaskCompleted(taskRecord.id); + } } finally { clearInterval(extendHandle); - state.currentExecutionAbort = null; - state.currentTaskId = null; - setSchedulerCurrentTask(null); + state.activeExecutions.delete(taskRecord.id); + syncSchedulerSurface(); noteSchedulerTaskFinished(); - processQueue(); + pumpExecutionLoop(); } } @@ -925,7 +1109,7 @@ async function extendActiveLease(taskId: string, leaseToken: string): Promise { + const accountIdentity = accountIdentityFromTask(task); + if (accountIdentity) { + const readiness = await ensureAccountReady(accountIdentity); + if (readiness.authState !== "active") { + return buildAccountBlockedResult(task, readiness); + } + } + const sessionHandle = createSessionHandle(task.accountId || task.id); const payload = task.payload; @@ -959,14 +1151,29 @@ async function executeTaskAdapter( adapter.executionMode === "session" || adapter.executionMode === "playwright" ? null : retainHotView(task.accountId || task.id); + const playwrightLease = + adapter.executionMode === "playwright" + ? await retainHiddenPlaywrightPage({ + accountId: task.accountId || task.id, + session: sessionHandle.session, + targetURL: resolvePlaywrightTargetURL(task.platform, payload), + title: `${task.title} · Hidden Playwright`, + }) + : null; try { - return await adapter.publish( + const result = await adapter.publish( { taskId: task.id, accountId: task.accountId || task.id, session: sessionHandle.session, view: viewHandle?.view ?? null, + playwright: playwrightLease + ? { + browser: playwrightLease.browser, + page: playwrightLease.page, + } + : null, signal, phase: "initial", article: await loadPublishArticle(task), @@ -976,7 +1183,10 @@ async function executeTaskAdapter( }, payload, ); + await maybeReportTaskAuthFailure(task, result, accountIdentity); + return result; } finally { + await playwrightLease?.release(); if (viewHandle) { releaseHotView(viewHandle.accountId); } @@ -1010,14 +1220,29 @@ async function executeTaskAdapter( adapter.executionMode === "session" || adapter.executionMode === "playwright" ? null : retainHotView(task.accountId || task.id); + const playwrightLease = + adapter.executionMode === "playwright" + ? await retainHiddenPlaywrightPage({ + accountId: task.accountId || task.id, + session: sessionHandle.session, + targetURL: resolvePlaywrightTargetURL(task.platform, payload), + title: `${task.title} · Hidden Playwright`, + }) + : null; try { - return await adapter.query( + const result = await adapter.query( { taskId: task.id, accountId: task.accountId || task.id, session: sessionHandle.session, view: viewHandle?.view ?? null, + playwright: playwrightLease + ? { + browser: playwrightLease.browser, + page: playwrightLease.page, + } + : null, signal, phase: "initial", reportProgress(stage: string) { @@ -1026,13 +1251,91 @@ async function executeTaskAdapter( }, payload, ); + await maybeReportTaskAuthFailure(task, result, accountIdentity); + return result; } finally { + await playwrightLease?.release(); if (viewHandle) { releaseHotView(viewHandle.accountId); } } } +function buildAccountBlockedResult( + task: RuntimeTaskRecord, + readiness: Awaited>, +): AdapterExecutionResult { + if (readiness.authState === "challenge_required") { + return { + status: "failed", + summary: `${task.accountName} 需要完成人机验证后才能继续执行。`, + error: { + code: "desktop_account_challenge_required", + message: "desktop account challenge required", + }, + }; + } + + if (readiness.authState === "expired" || readiness.authState === "revoked") { + return { + status: "failed", + summary: `${task.accountName} 登录态已失效，请重新授权后再试。`, + error: { + code: "desktop_account_auth_expired", + message: "desktop account auth expired", + }, + }; + } + + return { + status: "unknown", + summary: `${task.accountName} 最近校验失败，已暂缓执行任务。`, + error: { + code: "desktop_account_probe_pending", + message: "desktop account readiness pending", + }, + }; +} + +async function maybeReportTaskAuthFailure( + task: RuntimeTaskRecord, + result: AdapterExecutionResult, + accountIdentity: PublishAccountIdentity | null, +): Promise { + if (!accountIdentity) { + return; + } + + await reportAccountFailure(accountIdentity, { + summary: result.summary, + error: result.error ?? null, + }).catch((error) => { + console.warn("[desktop-runtime] account failure report failed", { + taskId: task.id, + accountId: accountIdentity.id, + platform: accountIdentity.platform, + message: error instanceof Error ? error.message : String(error), + }); + }); +} + +function resolvePlaywrightTargetURL( + platform: string, + payload: Record, +): string { + const payloadURL = resolveStringField(payload, ["target_url", "targetUrl", "console_url", "consoleUrl", "url"]); + if (payloadURL) { + return payloadURL; + } + + const aiPlatform = getAIPlatformCatalogItem(platform); + if (aiPlatform?.consoleUrl) { + return aiPlatform.consoleUrl; + } + + return "about:blank"; +} + function buildScaffoldResult( task: RuntimeTaskRecord, adapterPayload: Record, @@ -1096,6 +1399,23 @@ function toPositiveInt(value: JsonValue | null): number | null { return null; } +function resolveStringField( + payload: Record, + keys: string[], +): string | null { + for (const key of keys) { + const value = payload[key]; + if (typeof value !== "string") { + continue; + } + const normalized = value.trim(); + if (normalized) { + return normalized; + } + } + return null; +} + function resolveStaleMonitoringBusinessDate(payload: Record): string | null { const businessDate = resolveMonitoringBusinessDate(payload); if (!businessDate) { @@ -1219,6 +1539,90 @@ function recordActivity( } } +function syncSchedulerSurface(): void { + setSchedulerQueueDepth(localQueueDepth()); + setSchedulerCurrentTask(primaryActiveTaskId()); +} + +function localQueueDepth(): number { + const snapshot = getMonitorSchedulerSnapshot(); + return state.publishQueue.length + snapshot.queuedCount + snapshot.leasingCount; +} + +function primaryActiveTaskId(): string | null { + const active = [...state.activeExecutions.values()]; + const publish = active.find((item) => item.kind === "publish") ?? null; + return publish?.taskId ?? active[0]?.taskId ?? null; +} + +function hasActivePublishExecution(): boolean { + return [...state.activeExecutions.values()].some((item) => item.kind === "publish"); +} + +function activeMonitorCount(): number { + return [...state.activeExecutions.values()].filter((item) => item.kind === "monitor").length; +} + +function activeMonitorPlatforms(): Set { + return new Set( + [...state.activeExecutions.values()] + .filter((item) => item.kind === "monitor") + .map((item) => item.platform), + ); +} + +function activeMonitorQuestionKeys(): Set { + return new Set( + getMonitorSchedulerSnapshot().tasks + .filter((task) => task.state === "active" && Boolean(task.questionKey)) + .map((task) => task.questionKey as string), + ); +} + +function canStartAnotherMonitorExecution(): boolean { + if (state.publishQueue.length > 0 || hasActivePublishExecution()) { + return false; + } + const limit = resolveAdaptiveMonitorConcurrency(); + return limit > 0 && activeMonitorCount() < limit; +} + +function resolveAdaptiveMonitorConcurrency(): number { + const monitorAccountCount = state.accounts.filter((account) => + isAIPlatformId(account.platform) + && getProjectedAccountHealth({ + accountId: account.id, + platform: account.platform, + health: account.health, + verifiedAt: account.verified_at, + }).health === "live", + ).length; + if (monitorAccountCount <= 1) { + return monitorAccountCount; + } + + const metrics = getProcessMetricsSnapshot().latestSample; + if (!metrics) { + return 1; + } + + const totalCPUUsage = metrics.processes.reduce((sum, item) => sum + item.percentCPUUsage, 0); + const totalPrivateBytesKB = metrics.totals.privateBytes; + const processCount = metrics.totals.processCount; + const mainPrivateBytesKB = metrics.mainProcess.memory?.private ?? 0; + + if ( + totalCPUUsage >= 180 + || totalPrivateBytesKB >= 2_500_000 + || mainPrivateBytesKB >= 1_000_000 + || processCount >= 40 + ) { + return 1; + } + + return 2; +} + function handleAuthExpired(error: unknown): void { const message = errorMessage(error); state.lastError = message; diff --git a/apps/desktop-client/src/main/runtime-events.ts b/apps/desktop-client/src/main/runtime-events.ts new file mode 100644 index 0000000..9b03a9f --- /dev/null +++ b/apps/desktop-client/src/main/runtime-events.ts @@ -0,0 +1,34 @@ +type RuntimeInvalidationReason = "account-health"; + +type RuntimeInvalidationListener = (event: { + reason: RuntimeInvalidationReason; + at: number; +}) => void; + +const listeners = new Set(); + +export function emitRuntimeInvalidated(reason: RuntimeInvalidationReason): void { + const event = { + reason, + at: Date.now(), + } as const; + + for (const listener of listeners) { + try { + listener(event); + } catch (error) { + console.warn("[desktop-runtime] invalidation listener failed", { + reason, + message: error instanceof Error ? error.message : String(error), + }); + } + } +} + +export function onRuntimeInvalidated(listener: RuntimeInvalidationListener): () => void { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; +} + diff --git a/apps/desktop-client/src/main/runtime-snapshot.ts b/apps/desktop-client/src/main/runtime-snapshot.ts index 93835b1..cdd35dc 100644 --- a/apps/desktop-client/src/main/runtime-snapshot.ts +++ b/apps/desktop-client/src/main/runtime-snapshot.ts @@ -1,10 +1,13 @@ import { app } from "electron/main"; +import { getProjectedAccountHealth } from "./account-health"; import { getProcessMetricsSnapshot } from "./process-metrics"; import { getCircuitBreakerState } from "./circuit-breaker"; import { collectRecoverySnapshot } from "./crash-recovery"; import { getKeepAlivePlan } from "./keep-alive"; import { getLeaseManagerSnapshot } from "./lease-manager"; +import { getMonitorSchedulerSnapshot } from "./monitor-scheduler"; +import { getHiddenPlaywrightSnapshot } from "./playwright-cdp"; import { getRateLimiterSnapshot } from "./rate-limiter"; import { getRuntimeControllerSnapshot } from "./runtime-controller"; import { getSchedulerState } from "./scheduler"; @@ -71,23 +74,35 @@ function createLiveRuntimeSnapshot(controller: ReturnType task.accountId === account.id && ["queued", "in_progress"].includes(task.status), ).length; + const projected = getProjectedAccountHealth({ + accountId: account.id, + platform: account.platform, + health: account.health, + verifiedAt: account.verified_at, + }); return { id: account.id, platform: account.platform, - displayName: account.display_name, + displayName: projected.profile?.displayName ?? account.display_name, platformUid: normalizeAccountPlatformUid(account.platform_uid) || account.platform_uid, - avatarUrl: account.avatar_url ?? null, - health: account.health, + avatarUrl: projected.profile?.avatarUrl ?? account.avatar_url ?? null, + health: projected.health, + authState: projected.authState, + probeState: projected.probeState, + authReason: projected.authReason, tags: account.tags, syncVersion: account.sync_version, clientId: account.client_id ?? primaryClientId, - partition: sessionHandle?.partition ?? `persist:acc-${account.id}`, + partition: projected.partition || sessionHandle?.partition || `persist:acc-${account.id}`, sessionState: isHot ? "hot" : sessionHandle ? "warm" : "cold", lastSyncAt: controller.lastAccountsSyncAt || parseTimestamp(account.verified_at) || now, + lastVerifiedAt: projected.lastVerifiedAt, + lastProbeAt: projected.lastProbeAt, + nextProbeAt: projected.nextProbeAt, queueDepth: accountQueueDepth, online: clientOnline, }; @@ -150,9 +165,11 @@ function createLiveRuntimeSnapshot(controller: ReturnType, releaseRuntimeSession: (revoke?: boolean) => ipcRenderer.invoke("desktop:runtime-session-release", revoke) as Promise, + onRuntimeInvalidated: ( + listener: (event: { reason: "account-health"; at: number }) => void, + ) => { + const wrapped = (_event: IpcRendererEvent, payload: { reason: "account-health"; at: number }) => { + listener(payload); + }; + ipcRenderer.on("desktop:runtime-invalidated", wrapped); + return () => { + ipcRenderer.off("desktop:runtime-invalidated", wrapped); + }; + }, }, }; diff --git a/apps/desktop-client/src/renderer/composables/useDesktopRuntime.ts b/apps/desktop-client/src/renderer/composables/useDesktopRuntime.ts index cd49337..e89e4f1 100644 --- a/apps/desktop-client/src/renderer/composables/useDesktopRuntime.ts +++ b/apps/desktop-client/src/renderer/composables/useDesktopRuntime.ts @@ -11,6 +11,7 @@ const error = shallowRef(null); let intervalHandle: ReturnType | null = null; let subscribers = 0; let visibilityListenerBound = false; +let runtimeInvalidationUnsubscribe: (() => void) | null = null; function isPageVisible(): boolean { return typeof document === "undefined" || document.visibilityState === "visible"; @@ -95,6 +96,18 @@ function bindVisibilityListener() { visibilityListenerBound = true; } +function bindRuntimeInvalidationListener() { + if (runtimeInvalidationUnsubscribe || !window.desktopBridge?.app?.onRuntimeInvalidated) { + return; + } + + runtimeInvalidationUnsubscribe = window.desktopBridge.app.onRuntimeInvalidated(() => { + if (isPageVisible()) { + void refreshRuntimeSnapshot(); + } + }); +} + function unbindVisibilityListener() { if (!visibilityListenerBound || typeof document === "undefined") { return; @@ -104,10 +117,16 @@ function unbindVisibilityListener() { visibilityListenerBound = false; } +function unbindRuntimeInvalidationListener() { + runtimeInvalidationUnsubscribe?.(); + runtimeInvalidationUnsubscribe = null; +} + export function useDesktopRuntime() { onMounted(() => { subscribers += 1; bindVisibilityListener(); + bindRuntimeInvalidationListener(); if (!snapshot.value && !loading.value) { void refreshRuntimeSnapshot(); } @@ -119,6 +138,7 @@ export function useDesktopRuntime() { if (subscribers === 0) { stopPolling(); unbindVisibilityListener(); + unbindRuntimeInvalidationListener(); return; } diff --git a/apps/desktop-client/src/renderer/env.d.ts b/apps/desktop-client/src/renderer/env.d.ts index 3aa4c20..19ba0e6 100644 --- a/apps/desktop-client/src/renderer/env.d.ts +++ b/apps/desktop-client/src/renderer/env.d.ts @@ -29,6 +29,9 @@ declare global { retryPublishTask(taskId: string): Promise; syncRuntimeSession(session: DesktopRuntimeSessionSyncRequest | null): Promise; releaseRuntimeSession(revoke?: boolean): Promise; + onRuntimeInvalidated( + listener: (event: { reason: "account-health"; at: number }) => void, + ): () => void; }; }; } diff --git a/apps/desktop-client/src/renderer/types.ts b/apps/desktop-client/src/renderer/types.ts index 155ca17..28d6df6 100644 --- a/apps/desktop-client/src/renderer/types.ts +++ b/apps/desktop-client/src/renderer/types.ts @@ -19,12 +19,29 @@ export interface RuntimeAccount { platformUid: string; avatarUrl: string | null; health: "live" | "expired" | "captcha" | "risk"; + authState: "unknown" | "active" | "expiring_soon" | "challenge_required" | "expired" | "revoked"; + probeState: "idle" | "probing" | "network_error"; + authReason: + | "bind_success" + | "probe_success" + | "login_redirect" + | "http_401" + | "captcha_gate" + | "missing_partition" + | "silent_refresh_failed" + | "manual_unbind" + | "too_many_failures" + | "network_error" + | null; tags: string[]; syncVersion: number; clientId: string; partition: string; sessionState: "hot" | "warm" | "cold"; lastSyncAt: number; + lastVerifiedAt: number | null; + lastProbeAt: number | null; + nextProbeAt: number | null; queueDepth: number; online: boolean; } diff --git a/apps/desktop-client/src/renderer/views/AccountsView.vue b/apps/desktop-client/src/renderer/views/AccountsView.vue index 6a3d6a8..186a8a7 100644 --- a/apps/desktop-client/src/renderer/views/AccountsView.vue +++ b/apps/desktop-client/src/renderer/views/AccountsView.vue @@ -6,7 +6,7 @@ import StatusBadge from "../components/StatusBadge.vue"; import SurfaceCard from "../components/SurfaceCard.vue"; import { useDesktopRuntime } from "../composables/useDesktopRuntime"; import { showClientActionError } from "../lib/client-errors"; -import { formatDateTime, titleCaseToken } from "../lib/formatters"; +import { formatDateTime, formatRelativeTime, titleCaseToken } from "../lib/formatters"; import { desktopPublishMediaCatalog } from "../lib/media-catalog"; import type { RuntimeAccount } from "../types"; @@ -70,12 +70,13 @@ function platformMeta(platform: string) { } function authState(account: AccountRow): AccountAuthState { - switch (account.health) { - case "live": + switch (account.authState) { + case "active": return "authorized"; case "expired": + case "revoked": return "expired"; - case "captcha": + case "challenge_required": return "attention"; default: return "risk"; @@ -89,9 +90,9 @@ function authStateLabel(account: AccountRow): string { case "expired": return "授权过期"; case "attention": - return "待补登"; + return "需人工验证"; default: - return "风险观察"; + return account.probeState === "network_error" ? "最近校验失败" : "待重新校验"; } } @@ -123,6 +124,16 @@ function sessionStateLabel(account: AccountRow): string { } } +function verificationTimeLabel(account: AccountRow): string { + if (account.lastVerifiedAt) { + return `${formatRelativeTime(account.lastVerifiedAt)}校验通过`; + } + if (account.probeState === "probing") { + return "正在校验"; + } + return "尚未校验"; +} + const filteredAccounts = computed(() => { const keyword = searchQuery.value.trim().toLowerCase(); @@ -382,7 +393,7 @@ const tableColumns = [ @@ -399,8 +410,8 @@ const tableColumns = [ diff --git a/apps/desktop-client/src/renderer/views/AiPlatformsView.vue b/apps/desktop-client/src/renderer/views/AiPlatformsView.vue index ad81092..683a8e4 100644 --- a/apps/desktop-client/src/renderer/views/AiPlatformsView.vue +++ b/apps/desktop-client/src/renderer/views/AiPlatformsView.vue @@ -11,7 +11,7 @@ import { computed, ref } from "vue"; import { useDesktopRuntime } from "../composables/useDesktopRuntime"; import { showClientActionError } from "../lib/client-errors"; -import { formatDateTime } from "../lib/formatters"; +import { formatDateTime, formatRelativeTime } from "../lib/formatters"; import { desktopMonitoringMediaCatalog } from "../lib/media-catalog"; import type { RuntimeAccount } from "../types"; @@ -52,15 +52,19 @@ function authLabel(account: AccountRow | null) { return "未绑定"; } - switch (account.health) { - case "live": + switch (account.authState) { + case "active": return "授权正常"; - case "captcha": - return "待处理"; + case "challenge_required": + return "需人工验证"; case "expired": return "授权过期"; + case "expiring_soon": + return account.probeState === "network_error" ? "最近校验失败" : "待重新校验"; + case "revoked": + return "已停用"; default: - return "风险观察"; + return account.probeState === "probing" ? "校验中" : "待校验"; } } @@ -69,18 +73,38 @@ function authColor(account: AccountRow | null) { return "default"; } - switch (account.health) { - case "live": + switch (account.authState) { + case "active": return "success"; - case "captcha": + case "challenge_required": return "warning"; case "expired": + case "revoked": return "error"; + case "expiring_soon": + return account.probeState === "network_error" ? "warning" : "default"; default: return "default"; } } +function verificationLabel(account: AccountRow): string { + if (account.lastVerifiedAt) { + return `${formatRelativeTime(account.lastVerifiedAt)}校验通过`; + } + if (account.probeState === "probing") { + return "正在执行首轮校验"; + } + return "尚未完成校验"; +} + +function nextProbeLabel(account: AccountRow): string { + if (!account.nextProbeAt) { + return "未排程"; + } + return formatRelativeTime(account.nextProbeAt); +} + function sessionLabel(account: AccountRow): string { switch (account.sessionState) { case "hot": @@ -243,6 +267,14 @@ async function unbindPlatform(account: AccountRow) { 最近同步 {{ formatDateTime(platform.account.lastSyncAt) }} +
+ 校验状态 + {{ verificationLabel(platform.account) }} +
+
+ 下次巡检 + {{ nextProbeLabel(platform.account) }} +
diff --git a/apps/desktop-client/src/renderer/views/HomeView.vue b/apps/desktop-client/src/renderer/views/HomeView.vue index 951cf6e..9ed33bf 100644 --- a/apps/desktop-client/src/renderer/views/HomeView.vue +++ b/apps/desktop-client/src/renderer/views/HomeView.vue @@ -236,7 +236,7 @@ const subsystemCards = computed(() => { /* Metrics Top Grid */ .metrics-grid { display: grid; - grid-template-columns: repeat(auto-fit, minmax(260px, 1fr)); + grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 24px; } diff --git a/findings.md b/findings.md index 7904571..842053b 100644 --- a/findings.md +++ b/findings.md @@ -4,6 +4,7 @@ - Verify the repository against the 9-step implementation plan provided by the user. - Use `/Users/liangxu/Documents/test/geo-rankly/docs/superpowers/specs/2026-03-31-admin-web-backend-core-design.md` as the source of truth for the next development slice. - Continue development immediately after identifying the earliest unfinished step. +- For desktop AI monitoring, put scheduling authority on the client, keep scraping silent/background-only, allow stale cross-day tasks to be dropped, and avoid fanning one question to all models at once. ## Research Findings - The repository already contains core backend scaffolding under `server/`, including `cmd/tenant-api`, `internal/bootstrap`, `shared`, `tenant`, `migrations`, and SQL query files. @@ -62,6 +63,22 @@ - `template_prompt.go`, `template_assist.go`, `prompt_generate_service.go`, and `knowledge_service.go` now assemble prompts from the centralized prompt package, so runtime business logic no longer embeds the long prompt text directly. - The prompt centralization also corrected invalid JSON quoting in the seed card config `custom_placeholder` fields, which previously contained unescaped inner quotes. - Targeted verification passed with `go test ./internal/tenant/... ./cmd/dev-seed` from `server/`, confirming the refactor compiles cleanly across the affected runtime and seed paths. +- `desktop-client` now includes an AI platform management view with all six target platforms and the existing bind/open/unbind flows wired through the desktop bridge. +- `account-binder.ts` already supports all six AI platforms through generic binding definitions, but identity detection is heuristic and not yet platform-specific. +- The current desktop runtime executes tasks through a simple in-memory FIFO guarded by a single `currentTaskId`, so all desktop tasks are effectively global-serial today. +- Monitor-task dropping by `business_date` is already implemented in the runtime controller; stale tasks are returned as `unknown` with a client-drop reason instead of being retried on the next day. +- There is currently no durable local monitor-task queue/cache in `desktop-client`; if the process exits, queued work only survives on the server side. +- `process-metrics.ts` already samples Electron process CPU and memory, which gives us enough local telemetry to drive a small adaptive-concurrency policy without introducing a new metrics dependency. +- `session-registry.ts` already persists per-account Electron session partitions to `desktop-session-partitions.json`, showing that lightweight file-backed persistence is already accepted in this runtime. +- The only implemented monitor adapter is `doubao`, and it still uses the existing hidden `WebContentsView` path rather than Playwright. +- `playwright-core` is installed in `apps/desktop-client`, but there is no `connectOverCDP()` runtime integration yet. +- Hidden execution primitives already exist in two forms: hidden `BrowserWindow` instances for silent account checks and pooled hidden `WebContentsView` instances for existing adapters. +- Runtime diagnostics already expose scheduler and process-metrics state through the desktop snapshot, so new scheduler/CDP fields can be surfaced without inventing a new renderer path. +- A new local monitor scheduler now persists queued monitor tasks to `app.getPath("userData")/desktop-monitor-scheduler.json`, recovers same-day tasks after restart, and drops stale cross-day tasks during hydration. +- The new scheduler is intentionally conservative when a task has no question-level metadata: unknown-question monitor tasks do not consume the second concurrency slot, which avoids accidental same-question fan-out until server event metadata is available. +- Electron process metrics are sufficient to drive a two-tier adaptive concurrency policy: default to `1`, raise to `2` only when the machine looks healthy enough and multiple live AI accounts exist. +- Desktop task SSE events now carry optional `business_date`, `scheduler_group_key`, `question_text`, and `title` metadata derived from the task payload, allowing the local scheduler to defer same-question fan-out before leasing. +- A new hidden Playwright manager now connects to Electron Chromium through a local CDP endpoint and leases hidden pages bound to the existing account session partition, while leaving the legacy hidden-view path available for current adapters. ## Technical Decisions | Decision | Rationale | @@ -78,6 +95,9 @@ | Introduce a minimal `vue-i18n` foundation now instead of leaving hard-coded strings in new pages | The user explicitly called out the current i18n approach as non-standard | | Keep `AppShell.vue` on the user-preferred version and avoid structural/style rewrites there | The user said that file is managed elsewhere and only allowed inline-style cleanup | | Create a dedicated `internal/tenant/prompts` package for prompt text and seed definitions | This keeps prompt tuning in one place while minimizing refactor risk in the current dirty worktree | +| Add the local scheduler before wiring more monitor adapters | This aligns with the user's requirement that the client, not the server, decides actual execution timing | +| Keep publish-task execution behavior intact while evolving monitor-task scheduling | Publish is already working and should not be destabilized by monitor-specific policy changes | +| Build the hidden Playwright layer as reusable infrastructure instead of burying it inside one adapter | Qwen is the immediate motivation, but the same layer will likely be needed by other complex AI platforms | ## Issues Encountered | Issue | Resolution | @@ -86,6 +106,7 @@ | `make sqlc-generate` fails before any code is generated | Repair the `sqlc.yaml` schema path and then build repository wrappers around generated queries | | `go run` migrate initially failed with `unknown driver postgres` | Added `-tags postgres` to the Makefile-managed migrate command | | Playwright MCP could not open a browser session locally because it tried to create `/.playwright-mcp` | Use `vite preview` plus `curl` verification for this turn instead of spending time on browser tool plumbing | +| `desktop-client` has no existing SQLite/local DB layer despite `better-sqlite3` being installed | Decide between a tiny file-backed queue or adding a new DB-backed task store during implementation | ## Resources - `/Users/liangxu/Documents/test/geo-rankly/docs/superpowers/specs/2026-03-31-admin-web-backend-core-design.md` diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index 62dcea1..d2627bf 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -240,6 +240,10 @@ export interface DesktopTaskEventMessage { target_account_id: string; target_client_id: string; platform: string; + title?: string; + business_date?: string; + scheduler_group_key?: string; + question_text?: string; status: | "queued" | "in_progress" diff --git a/progress.md b/progress.md index 16a4378..404f588 100644 --- a/progress.md +++ b/progress.md @@ -1,5 +1,54 @@ # Progress Log +## Session: 2026-04-20 + +### Phase 15: Desktop Scheduler Design & Integration +- **Status:** complete +- Actions taken: + - Re-read the planning files and recovered the current desktop AI monitoring scope from the existing repo changes. + - Audited the desktop runtime controller, process metrics sampler, runtime snapshot, session registry, and monitor adapter surface. + - Confirmed the current desktop task executor is global-serial, monitor-task staleness dropping is already implemented, and no durable local queue/cache exists yet. + - Confirmed `playwright-core` is installed but `connectOverCDP()` is not wired, while hidden Electron execution already exists via `BrowserWindow` and `WebContentsView`. + - Added `monitor-scheduler.ts` as a durable file-backed local scheduler for monitor tasks, including same-day persistence, stale cross-day pruning, per-platform mutual exclusion, question-level cooldowns, and lease-miss backoff. + - Refactored `runtime-controller.ts` so publish tasks keep a conservative explicit queue while monitor tasks are selected through the local scheduler with adaptive max concurrency from Electron process metrics. + - Extended runtime diagnostics to surface the monitor scheduler snapshot and made the lease manager track multiple active leases instead of a single global lease placeholder. +- Files created/modified: + - `task_plan.md` (updated) + - `findings.md` (updated) + - `progress.md` (updated) + - `apps/desktop-client/src/main/monitor-scheduler.ts` (created) + - `apps/desktop-client/src/main/runtime-controller.ts` (updated) + - `apps/desktop-client/src/main/lease-manager.ts` (updated) + - `apps/desktop-client/src/main/runtime-snapshot.ts` (updated) + +### Phase 16: Hidden Playwright CDP Execution Layer +- **Status:** complete +- Actions taken: + - Added `playwright-cdp.ts` as a reusable hidden-browser manager that launches hidden Electron `BrowserWindow` instances on account-bound sessions and attaches Playwright via `chromium.connectOverCDP()`. + - Enabled Electron remote debugging in bootstrap so the app exposes a CDP endpoint for local Playwright attachment. + - Extended adapter context with an optional Playwright `{ browser, page }` handle and wired the runtime controller to acquire/release hidden Playwright pages when an adapter declares `executionMode: "playwright"`. + - Kept the existing hidden `WebContentsView` path intact so current adapters like Doubao continue to work unchanged. +- Files created/modified: + - `apps/desktop-client/src/main/playwright-cdp.ts` (created) + - `apps/desktop-client/src/main/bootstrap.ts` (updated) + - `apps/desktop-client/src/main/adapters/base.ts` (updated) + - `apps/desktop-client/src/main/runtime-controller.ts` (updated) + +### Phase 17: Verification & Delivery +- **Status:** in_progress +- Actions taken: + - Added optional scheduler metadata to desktop task events in shared types and server publishers so the local scheduler can make safer fan-out decisions before leasing. + - Ran `gofmt` on the modified Go files after the event metadata changes. + - Verified `pnpm --filter @geo/desktop-client typecheck` passes. + - Verified `pnpm --filter @geo/desktop-client build` passes. + - Verified `pnpm --filter admin-web typecheck` still passes after the shared-type event additions. + - Verified `go test ./internal/tenant/app/...` passes from `server/`. +- Files created/modified: + - `packages/shared-types/src/index.ts` (updated) + - `server/internal/tenant/app/desktop_task_events.go` (updated) + - `server/internal/tenant/app/desktop_task_service.go` (updated) + - `server/internal/tenant/app/publish_job_service.go` (updated) + ## Session: 2026-04-04 ### Phase 13: Prompt Centralization diff --git a/server/internal/tenant/app/desktop_task_events.go b/server/internal/tenant/app/desktop_task_events.go index dfc1472..1f899bd 100644 --- a/server/internal/tenant/app/desktop_task_events.go +++ b/server/internal/tenant/app/desktop_task_events.go @@ -2,23 +2,29 @@ package app import ( "context" + "crypto/sha1" "encoding/json" + "fmt" "time" "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" ) type DesktopTaskEvent struct { - Type string `json:"type"` - TaskID string `json:"task_id"` - JobID string `json:"job_id"` - WorkspaceID int64 `json:"workspace_id"` - TargetAccountID string `json:"target_account_id"` - TargetClientID string `json:"target_client_id"` - Platform string `json:"platform"` - Status string `json:"status"` - Kind string `json:"kind"` - UpdatedAt time.Time `json:"updated_at"` + Type string `json:"type"` + TaskID string `json:"task_id"` + JobID string `json:"job_id"` + WorkspaceID int64 `json:"workspace_id"` + TargetAccountID string `json:"target_account_id"` + TargetClientID string `json:"target_client_id"` + Platform string `json:"platform"` + Title *string `json:"title,omitempty"` + BusinessDate *string `json:"business_date,omitempty"` + SchedulerGroupKey *string `json:"scheduler_group_key,omitempty"` + QuestionText *string `json:"question_text,omitempty"` + Status string `json:"status"` + Kind string `json:"kind"` + UpdatedAt time.Time `json:"updated_at"` } func publishDesktopTaskEvent(ctx context.Context, client *rabbitmq.Client, event DesktopTaskEvent) error { @@ -33,3 +39,88 @@ func publishDesktopTaskEvent(ctx context.Context, client *rabbitmq.Client, event return client.PublishDesktopTaskEvent(ctx, payload) } + +func deriveDesktopTaskEventMetadataFromPayload( + kind string, + payload []byte, +) (title *string, businessDate *string, schedulerGroupKey *string, questionText *string) { + if len(payload) == 0 { + return nil, nil, nil, nil + } + + var envelope map[string]any + if err := json.Unmarshal(payload, &envelope); err != nil { + return nil, nil, nil, nil + } + + title = firstDesktopTaskEventString(envelope, + "title", + "question_title", + "questionTitle", + ) + businessDate = firstDesktopTaskEventString(envelope, + "business_date", + "businessDate", + "metric_date", + "date", + ) + questionText = firstDesktopTaskEventString(envelope, + "question_text", + "questionText", + "query", + "question", + "prompt", + "content", + ) + + if explicit := firstDesktopTaskEventString(envelope, + "scheduler_group_key", + "schedulerGroupKey", + ); explicit != nil { + return title, businessDate, explicit, questionText + } + + if questionHash := firstDesktopTaskEventString(envelope, "question_hash", "questionHash"); questionHash != nil { + return title, businessDate, questionHash, questionText + } + + if questionID := firstDesktopTaskEventString(envelope, "question_id", "questionId"); questionID != nil { + value := fmt.Sprintf("qid:%s", *questionID) + return title, businessDate, &value, questionText + } + + if kind == "monitor" && questionText != nil { + sum := sha1.Sum([]byte(*questionText)) + value := fmt.Sprintf("qtxt:%x", sum) + return title, businessDate, &value, questionText + } + + return title, businessDate, nil, questionText +} + +func firstDesktopTaskEventString(payload map[string]any, keys ...string) *string { + for _, key := range keys { + value, exists := payload[key] + if !exists { + continue + } + + switch typed := value.(type) { + case string: + if typed != "" { + return &typed + } + case float64: + text := fmt.Sprintf("%.0f", typed) + return &text + case int64: + text := fmt.Sprintf("%d", typed) + return &text + case int: + text := fmt.Sprintf("%d", typed) + return &text + } + } + + return nil +} diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 1d81bc5..d3be3a8 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -725,17 +725,22 @@ func (s *DesktopTaskService) publishTaskEvent(ctx context.Context, task *reposit return } + title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(task.Kind, task.Payload) err := publishDesktopTaskEvent(ctx, s.messaging, DesktopTaskEvent{ - Type: eventType, - TaskID: task.DesktopID.String(), - JobID: task.JobID.String(), - WorkspaceID: task.WorkspaceID, - TargetAccountID: task.TargetAccountID.String(), - TargetClientID: task.TargetClientID.String(), - Platform: task.Platform, - Status: task.Status, - Kind: task.Kind, - UpdatedAt: task.UpdatedAt, + Type: eventType, + TaskID: task.DesktopID.String(), + JobID: task.JobID.String(), + WorkspaceID: task.WorkspaceID, + TargetAccountID: task.TargetAccountID.String(), + TargetClientID: task.TargetClientID.String(), + Platform: task.Platform, + Title: title, + BusinessDate: businessDate, + SchedulerGroupKey: schedulerGroupKey, + QuestionText: questionText, + Status: task.Status, + Kind: task.Kind, + UpdatedAt: task.UpdatedAt, }) if err != nil { s.logWarn("desktop task event publish failed", err, zap.String("task_id", task.DesktopID.String()), zap.String("event_type", eventType)) diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index e686d0d..cc78b42 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/server/internal/tenant/app/publish_job_service.go @@ -213,17 +213,22 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID) for _, task := range createdTasks { + title, businessDate, schedulerGroupKey, questionText := deriveDesktopTaskEventMetadataFromPayload(task.Kind, task.Payload) if publishErr := publishDesktopTaskEvent(context.Background(), s.messaging, DesktopTaskEvent{ - Type: "task_available", - TaskID: task.DesktopID.String(), - JobID: task.JobID.String(), - WorkspaceID: task.WorkspaceID, - TargetAccountID: task.TargetAccountID.String(), - TargetClientID: task.TargetClientID.String(), - Platform: task.Platform, - Status: task.Status, - Kind: task.Kind, - UpdatedAt: task.UpdatedAt, + Type: "task_available", + TaskID: task.DesktopID.String(), + JobID: task.JobID.String(), + WorkspaceID: task.WorkspaceID, + TargetAccountID: task.TargetAccountID.String(), + TargetClientID: task.TargetClientID.String(), + Platform: task.Platform, + Title: title, + BusinessDate: businessDate, + SchedulerGroupKey: schedulerGroupKey, + QuestionText: questionText, + Status: task.Status, + Kind: task.Kind, + UpdatedAt: task.UpdatedAt, }); publishErr != nil { s.logWarn("desktop task available event publish failed", publishErr, zap.String("task_id", task.DesktopID.String())) } diff --git a/task_plan.md b/task_plan.md index 2d093c6..6eb9032 100644 --- a/task_plan.md +++ b/task_plan.md @@ -1,10 +1,10 @@ # Task Plan: GEO admin-web backend completion and frontend foundation ## Goal -Finish the completed backend verification/implementation chain, then resume the interrupted work by establishing the missing frontend foundation for `admin-web` so it can authenticate against `tenant-api` and render the first dashboard slice. +Continue the desktop AI monitoring implementation by first adding a client-side local scheduler with durable queueing and adaptive execution policy, then adding a reusable hidden Playwright CDP execution layer that attaches to Electron Chromium and reuses existing account partitions. ## Current Phase -Phase 14 +Phase 17 ## Phases ### Phase 1: Progress Verification @@ -94,10 +94,30 @@ Phase 14 - [x] Deliver the extraction result and any remaining gaps - **Status:** complete +### Phase 15: Desktop Scheduler Design & Integration +- [x] Audit the current desktop runtime queue/task execution path and lock the scheduler constraints +- [x] Implement a durable local monitor-task queue/cache with stale-task cleanup +- [x] Implement platform-level mutual exclusion and adaptive global concurrency policy +- [x] Integrate the scheduler into runtime leasing/execution without regressing publish tasks +- **Status:** complete + +### Phase 16: Hidden Playwright CDP Execution Layer +- [x] Add a reusable hidden-browser manager that attaches Playwright to Electron Chromium over CDP +- [x] Reuse desktop account partition/session state for hidden pages +- [x] Expose the hidden Playwright context/page lifecycle to future monitor adapters +- [x] Keep the current hidden-view path working for existing adapters during the transition +- **Status:** complete + +### Phase 17: Verification & Delivery +- [x] Run targeted desktop type/build verification +- [ ] Inspect runtime snapshot output for new scheduler/CDP state +- [ ] Summarize what is production-ready versus still scaffolded +- **Status:** in_progress + ## Key Questions -1. Which of the 9 steps are already implemented versus partially complete? -2. What is the earliest step whose acceptance criteria are not yet met? -3. What concrete code changes are required now to advance that step safely? +1. How should the desktop client defer and locally optimize monitor-task execution without violating one-task-per-platform serialism? +2. What is the smallest durable local cache that allows same-day resume while safely dropping stale next-day monitor tasks? +3. How can Playwright CDP attach cleanly to Electron Chromium while reusing existing account session partitions? ## Decisions Made | Decision | Rationale | @@ -115,6 +135,9 @@ Phase 14 | Reconstruct the screenshot layouts from visual position data, not OCR text alone | The user explicitly asked to follow image design including element placement | | Move the quota card into the left sidebar footer and use page-level hero sections | This matches the reference screenshots more closely than the previous generic topbar layout | | Centralize prompt text in a dedicated package instead of leaving it inside business logic functions | The user explicitly asked to extract hard-coded prompts for easier future optimization | +| Move scheduling authority fully into `desktop-client` | The user explicitly wants the server to dispatch tasks while the client decides when to execute them | +| Allow stale cross-day monitor tasks to be dropped on the client | The user explicitly allows漏采 and does not want next-day catch-up for unfinished tasks | +| Implement hidden browser infrastructure before expanding more adapters | Qwen and similar platforms cannot reliably use direct API calls and need browser-native execution | ## Errors Encountered | Error | Attempt | Resolution | @@ -125,9 +148,12 @@ Phase 14 | Gemini CLI `gemini-3.1-pro-preview` returned `429 MODEL_CAPACITY_EXHAUSTED` repeatedly | 1 | Stopped the Gemini path after the user requested not to use Gemini anymore | | Playwright MCP navigation failed locally because it attempted to create `/.playwright-mcp` | 1 | Fell back to preview HTTP checks instead of blocking the turn on Playwright environment setup | | Login button click did not trigger the sign-in flow in browser testing even though the backend endpoint worked | 1 | Bound the primary login button directly to `handleSubmit` and re-ran browser verification until the route change succeeded | +| Desktop monitor execution is currently global-serial and lacks durable local queueing | 1 | Replace the ad-hoc in-memory FIFO with a scheduler module that owns persistence, concurrency, and stale-task cleanup | +| `connectOverCDP()` is not wired yet even though `playwright-core` is installed | 1 | Add a dedicated hidden-browser manager instead of pushing CDP code directly into adapters or the runtime controller | ## Notes - Re-check task_plan.md before major implementation decisions. - Record concrete evidence for each claimed completed step. - Frontend scope for this turn is limited to the tenant-facing `admin-web` shell, not the platform ops console. - Visual references for this turn must be treated as layout guides, not just copy decks. +- Desktop AI monitoring scope for this turn is infrastructure-first: scheduler plus hidden browser layer, not all six adapters.