diff --git a/apps/admin-web/src/lib/desktop-account-runtime.ts b/apps/admin-web/src/lib/desktop-account-runtime.ts index 69449c6..2db25c7 100644 --- a/apps/admin-web/src/lib/desktop-account-runtime.ts +++ b/apps/admin-web/src/lib/desktop-account-runtime.ts @@ -1,6 +1,12 @@ import type { DesktopAccountInfo } from "@geo/shared-types"; export function resolveAccountHealth(account: DesktopAccountInfo): DesktopAccountInfo["health"] { + if ( + account.runtime_auth_state === "active" || + (account.runtime_auth_state === "expiring_soon" && account.runtime_probe_state !== "network_error") + ) { + return "live"; + } return account.runtime_health ?? account.health; } diff --git a/apps/desktop-client/src/main/account-health.ts b/apps/desktop-client/src/main/account-health.ts index 54d8d75..aaee217 100644 --- a/apps/desktop-client/src/main/account-health.ts +++ b/apps/desktop-client/src/main/account-health.ts @@ -22,13 +22,11 @@ import { 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 STALE_AFTER_MS = 45 * 60_000; const REGULAR_PROBE_MIN_MS = 30 * 60_000; -const REGULAR_PROBE_MAX_MS = 60 * 60_000; +const REGULAR_PROBE_MAX_MS = STALE_AFTER_MS; 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; const MAX_CONCURRENT_ACCOUNT_PROBES = 3; const MANUAL_PROBE_CACHE_TTL_MS = 5 * 60_000; @@ -100,7 +98,7 @@ function resolvePartition(accountId: string): string { } function nextInitialProbeAt(now = Date.now()): number { - return now + randomInt(FIRST_PROBE_MIN_MS, FIRST_PROBE_MAX_MS); + return now; } function nextRegularProbeAt(now = Date.now()): number { @@ -113,9 +111,9 @@ function normalizePersisted( return { accountId: record.accountId, platform: record.platform, - authState: "unknown", + authState: record.lastVerifiedAt ? "active" : "unknown", probeState: "idle", - authReason: null, + authReason: record.lastVerifiedAt ? "probe_success" : null, lastVerifiedAt: record.lastVerifiedAt ?? null, lastProbeAt: null, nextProbeAt: nextInitialProbeAt(), @@ -176,15 +174,7 @@ function ensureProbeTimer(): void { }, 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"; - } - +function coerceDerivedAuthState(record: AccountHealthRecord): AccountAuthState { return record.authState; } @@ -442,6 +432,18 @@ function markProbeQueued(record: AccountHealthRecord): AccountHealthSnapshot { return commitRecord(record, previousSignature); } +function isProbeInFlight(record: AccountHealthRecord): boolean { + return record.probeState === "queued" || record.probeState === "probing"; +} + +function isStaleActiveRecord(record: AccountHealthRecord, now = Date.now()): boolean { + return ( + record.authState === "active" + && Boolean(record.lastVerifiedAt) + && now - (record.lastVerifiedAt ?? 0) >= STALE_AFTER_MS + ); +} + function maxAttemptsForTrigger(trigger: ProbeTrigger): number { return trigger === "preflight" ? 1 : ACCOUNT_PROBE_RETRY_BACKOFF_MS.length + 1; } @@ -569,7 +571,10 @@ 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); + if (!record || isProbeInFlight(record)) { + return false; + } + return Boolean((record.nextProbeAt && record.nextProbeAt <= now) || isStaleActiveRecord(record, now)); }); for (const account of dueAccounts) { @@ -653,6 +658,11 @@ function syncTrackedAccountsInternal( if (removed) { emitRuntimeInvalidated("account-health"); } + if (pruneMissing) { + queueMicrotask(() => { + void runDueProbes(); + }); + } } export function forgetTrackedAccountHealth( diff --git a/apps/desktop-client/src/main/bootstrap.ts b/apps/desktop-client/src/main/bootstrap.ts index edcc9fe..abe43e7 100644 --- a/apps/desktop-client/src/main/bootstrap.ts +++ b/apps/desktop-client/src/main/bootstrap.ts @@ -337,7 +337,7 @@ function registerBridgeHandlers(): void { return null; }); safeHandle("desktop:probe-runtime-account", async (_event, accountId: string) => { - requestRuntimeAccountProbe(accountId); + await requestRuntimeAccountProbe(accountId); return captureRuntimeAccountSnapshot(accountId); }); safeHandle( diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index be4d210..52e6186 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -83,6 +83,7 @@ import { startHiddenPlaywrightReaper, } from "./playwright-cdp"; import { getProcessMetricsSnapshot } from "./process-metrics"; +import { onRuntimeInvalidated } from "./runtime-events"; import { initScheduler, noteSchedulerAccountsSync, @@ -97,7 +98,7 @@ import { setSchedulerPhase, setSchedulerQueueDepth, } from "./scheduler"; -import { clearSessionHandle, createSessionHandle } from "./session-registry"; +import { clearSessionHandle, createSessionHandle, getPersistedPartition, getSessionHandle } from "./session-registry"; import { cancelDesktopTask, completeDesktopTask, @@ -239,6 +240,8 @@ interface RuntimeState { pullTimer: ReturnType | null; accountHealthReportTimer: ReturnType | null; accountHealthReportQueue: Map; + accountHealthReportBadAccounts: Set; + accountRemoteHealth: Map; dispatchWsClient: DispatchWsClient | null; dispatchWsConnected: boolean; lastHeartbeatAt: number; @@ -266,6 +269,8 @@ const state: RuntimeState = { pullTimer: null, accountHealthReportTimer: null, accountHealthReportQueue: new Map(), + accountHealthReportBadAccounts: new Set(), + accountRemoteHealth: new Map(), dispatchWsClient: null, dispatchWsConnected: false, lastHeartbeatAt: 0, @@ -277,6 +282,7 @@ const state: RuntimeState = { lastError: null, activitySeq: 0, }; +let accountHealthReportBridgeUnsubscribe: (() => void) | null = null; function accountIdentityFromDesktopAccount(account: DesktopAccountInfo): PublishAccountIdentity { return { @@ -309,6 +315,14 @@ function isDefinitiveAuthState(authState: string): boolean { return ["active", "expired", "challenge_required", "revoked"].includes(authState); } +function hasLocalAccountSession(accountId: string): boolean { + return Boolean(getSessionHandle(accountId) || getPersistedPartition(accountId)); +} + +function shouldReportAccountPresence(account: DesktopAccountInfo): boolean { + return hasLocalAccountSession(account.id) || account.client_id === state.client?.id; +} + function runtimePlatformLabel(platform: string): string { return getAIPlatformCatalogItem(platform)?.label ?? platform; } @@ -473,6 +487,7 @@ function startRuntime(): void { initMonitorScheduler(); initPublishScheduler(); startHiddenPlaywrightReaper(); + ensureAccountHealthReportBridge(); state.running = true; setSchedulerPhase("running"); setSchedulerError(null); @@ -488,6 +503,19 @@ function startRuntime(): void { pumpExecutionLoop(); } +function ensureAccountHealthReportBridge(): void { + if (accountHealthReportBridgeUnsubscribe) { + return; + } + + accountHealthReportBridgeUnsubscribe = onRuntimeInvalidated((event) => { + if (event.reason !== "account-health" || !event.accountId) { + return; + } + enqueueAccountHealthReport(event.accountId); + }); +} + function stopRuntime(): void { state.running = false; state.leaseInFlightKinds.clear(); @@ -532,6 +560,8 @@ function clearRuntimeState(): void { state.accounts = []; state.accountProfiles.clear(); state.accountHealthReportQueue.clear(); + state.accountHealthReportBadAccounts.clear(); + state.accountRemoteHealth.clear(); state.activeExecutions.clear(); state.lastHeartbeatAt = 0; state.lastHeartbeatStatus = "idle"; @@ -860,14 +890,7 @@ async function sendHeartbeat(source: "startup" | "timer"): Promise { channel: state.client?.channel ?? "dev", startup: source === "startup", account_ids: state.accounts - .filter((account) => - getProjectedAccountHealth({ - accountId: account.id, - platform: account.platform, - health: account.health, - verifiedAt: account.verified_at, - }).health === "live", - ) + .filter((account) => shouldReportAccountPresence(account)) .map((account) => account.id), }); @@ -918,6 +941,7 @@ async function syncAccounts(source: "startup" | "manual"): Promise { const identities = accounts.map((account) => accountIdentityFromDesktopAccount(account)); syncTrackedAccounts(identities); + state.accountRemoteHealth = new Map(accounts.map((account) => [account.id, account.health])); state.accountProfiles.clear(); const syncedAccounts: Array = accounts.map((account) => { @@ -1036,11 +1060,15 @@ function buildAccountHealthReport(account: DesktopAccountInfo): DesktopAccountHe health: account.health, verifiedAt: account.verified_at, }); + const health = + projected.authState === "expiring_soon" && projected.probeState !== "network_error" + ? "live" + : projected.health; return { account_id: account.id, platform: account.platform, platform_uid: normalizeAccountPlatformUid(account.platform_uid) || account.platform_uid, - health: projected.health, + health, auth_state: projected.authState, probe_state: projected.probeState, auth_reason: projected.authReason, @@ -1051,13 +1079,47 @@ function buildAccountHealthReport(account: DesktopAccountInfo): DesktopAccountHe }; } +function isAccountHealthFailureReport(report: DesktopAccountHealthReport): boolean { + return ( + report.health !== "live" + || report.auth_state === "expired" + || report.auth_state === "challenge_required" + || report.auth_state === "revoked" + || report.probe_state === "network_error" + ); +} + +function shouldQueueAccountHealthReport( + account: DesktopAccountInfo, + report: DesktopAccountHealthReport, +): boolean { + if (isAccountHealthFailureReport(report)) { + return true; + } + + if (state.accountHealthReportBadAccounts.has(account.id)) { + return true; + } + + return (state.accountRemoteHealth.get(account.id) ?? account.health) !== "live"; +} + function enqueueAccountHealthReport(accountId: string): void { const account = state.accounts.find((item) => item.id === accountId) ?? null; if (!account) { return; } - state.accountHealthReportQueue.set(account.id, buildAccountHealthReport(account)); + const report = buildAccountHealthReport(account); + if (!shouldQueueAccountHealthReport(account, report)) { + return; + } + + if (isAccountHealthFailureReport(report)) { + state.accountHealthReportBadAccounts.add(account.id); + } + + state.accountHealthReportQueue.set(account.id, report); if (state.accountHealthReportTimer) { return; } @@ -1080,6 +1142,14 @@ async function flushAccountHealthReports(): Promise { try { await reportDesktopAccountHealth({ reports }); + for (const report of reports) { + state.accountRemoteHealth.set(report.account_id, report.health); + if (isAccountHealthFailureReport(report)) { + state.accountHealthReportBadAccounts.add(report.account_id); + } else { + state.accountHealthReportBadAccounts.delete(report.account_id); + } + } } catch (error) { console.warn("[desktop-runtime] account health report failed", { count: reports.length, @@ -1156,32 +1226,32 @@ export async function refreshRuntimeAccounts(): Promise { await syncAccounts("manual"); } -export function requestRuntimeAccountProbe( +export async function requestRuntimeAccountProbe( accountId: string, options: { account?: PublishAccountIdentity; force?: boolean } = {}, -): void { +): Promise { const existingAccount = state.accounts.find((item) => item.id === accountId) ?? null; const account = options.account ?? (existingAccount ? accountIdentityFromDesktopAccount(existingAccount) : null); if (!account) { throw new Error("runtime_account_not_found"); } - void probeTrackedAccount(account, { - trigger: "manual", - allowSilentRefresh: true, - force: options.force ?? true, - }) - .then(async () => { - enqueueAccountHealthReport(account.id); - await mirrorProjectedAccountProfile(account.id); - }) - .catch((error) => { - console.warn("[desktop-runtime] account probe failed", { - accountId: account.id, - platform: account.platform, - message: errorMessage(error), - }); + try { + await probeTrackedAccount(account, { + trigger: "manual", + allowSilentRefresh: true, + force: options.force ?? true, }); + enqueueAccountHealthReport(account.id); + await mirrorProjectedAccountProfile(account.id); + } catch (error) { + console.warn("[desktop-runtime] account probe failed", { + accountId: account.id, + platform: account.platform, + message: errorMessage(error), + }); + throw error; + } } export function noteRuntimeAccountBound(account: DesktopAccountInfo): void { diff --git a/server/internal/tenant/app/desktop_account_service.go b/server/internal/tenant/app/desktop_account_service.go index eede445..cbe68d3 100644 --- a/server/internal/tenant/app/desktop_account_service.go +++ b/server/internal/tenant/app/desktop_account_service.go @@ -505,7 +505,7 @@ func (s *DesktopAccountService) applyRuntimeHealth(ctx context.Context, workspac continue } - health := report.Health + health := effectiveDesktopAccountRuntimeHealth(report) authState := report.AuthState probeState := report.ProbeState checkedAt := report.CheckedAt @@ -523,6 +523,20 @@ func (s *DesktopAccountService) applyRuntimeHealth(ctx context.Context, workspac } } +func effectiveDesktopAccountRuntimeHealth(report bufferedDesktopAccountHealthReport) string { + authState := strings.TrimSpace(report.AuthState) + probeState := strings.TrimSpace(report.ProbeState) + switch authState { + case "active": + return "live" + case "expiring_soon": + if probeState != "network_error" { + return "live" + } + } + return report.Health +} + func (s *DesktopAccountService) buildDesktopAccountView( account *repository.DesktopAccount, clientMap map[uuid.UUID]*repository.DesktopClient, diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index 0bf006e..074b8ff 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/server/internal/tenant/app/publish_job_service.go @@ -121,7 +121,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr } accountHealth := account.Health if report, ok := runtimeHealth[account.DesktopID]; ok && strings.TrimSpace(report.Health) != "" { - accountHealth = report.Health + accountHealth = effectiveDesktopAccountRuntimeHealth(report) } if accountHealth != "live" { return nil, response.ErrConflict(40993, "desktop_account_not_publishable", "one or more selected desktop accounts require re-authorization before publishing")