import path from 'node:path'; import { spawn } from 'node:child_process'; import { PACKAGE_ROOT } from './config-loader.mjs'; let processorBridgePromise = null; function loadProcessorBridge() { if (!processorBridgePromise) processorBridgePromise = import('./processor-bridge.mjs'); return processorBridgePromise; } const defaultListenerApi = { getStatus: async () => (await loadProcessorBridge()).getPersonalListenerStatus(), recoverLogin: async () => (await loadProcessorBridge()).recoverPersonalLogin(), start: async () => (await loadProcessorBridge()).startPersonalListener(), stop: async options => (await loadProcessorBridge()).stopPersonalListener(options), }; export class PersonalPollingRuntime { constructor({ config, workspaceRoot, onState = () => {}, listenerApi = {} }) { this.config = config; this.workspaceRoot = workspaceRoot; this.onState = onState; this.listenerApi = { getStatus: listenerApi.getStatus || defaultListenerApi.getStatus, recoverLogin: listenerApi.recoverLogin || defaultListenerApi.recoverLogin, start: listenerApi.start || defaultListenerApi.start, stop: listenerApi.stop || defaultListenerApi.stop, }; this.running = false; this.listenerStarted = false; this.friendWorker = null; this.loopPromise = null; this.cancelWait = null; this.lastRecoveryAttemptAt = 0; this.lastRecoveryError = ''; } wait(ms) { return new Promise(resolve => { const timer = setTimeout(() => { this.cancelWait = null; resolve(); }, ms); this.cancelWait = () => { clearTimeout(timer); this.cancelWait = null; resolve(); }; }); } startFriendWorker() { if (!this.config.friendPollingEnabled || this.friendWorker) return; const scriptPath = path.join(PACKAGE_ROOT, 'scripts', 'friend-polling-worker.js'); this.friendWorker = spawn(process.execPath, [scriptPath], { cwd: this.workspaceRoot, env: { ...process.env, QIWEI_WORKSPACE_ROOT: this.workspaceRoot }, stdio: 'inherit', windowsHide: true, }); this.onState({ friendPolling: { status: 'running', pid: this.friendWorker.pid } }); this.friendWorker.once('exit', (code, signal) => { this.friendWorker = null; this.onState({ friendPolling: { status: this.running ? 'error' : 'stopped', code, signal } }); }); } recoveryDue() { const cooldownMs = Math.max(30000, Number(this.config.retryMs) || 10000); return Date.now() - this.lastRecoveryAttemptAt >= cooldownMs; } async recoverAndRestart(listener = {}) { if (!this.recoveryDue()) return null; this.lastRecoveryAttemptAt = Date.now(); this.onState({ personalPolling: { status: 'reconnecting' } }); const recovered = await this.listenerApi.recoverLogin(); const recoveredCode = Number(recovered?.summary?.statusCode); if (!recovered?.summary?.loggedIn && recoveredCode !== 2) { this.lastRecoveryError = String( recovered?.errors?.[0]?.message || recovered?.assistantMessage || 'Automatic login recovery is pending.', ); return null; } this.lastRecoveryError = ''; if (listener.running) await this.listenerApi.stop({ preserveAgentState: true }); this.listenerStarted = false; const started = await this.listenerApi.start(); this.listenerStarted = Boolean(started?.data?.running); return started; } async pollOnce() { let result = await this.listenerApi.getStatus(); let listener = result?.data?.listener || {}; let account = result?.data?.account || {}; let requiresLogin = account.reason === 'upstream_device_missing'; if (account.online === false && !requiresLogin) { await this.recoverAndRestart(listener); result = await this.listenerApi.getStatus(); listener = result?.data?.listener || {}; account = result?.data?.account || {}; requiresLogin = account.reason === 'upstream_device_missing'; } if (!listener.running && account.online !== false) { try { const started = await this.listenerApi.start(); this.listenerStarted = Boolean(started?.data?.running); } catch (error) { const recovered = await this.recoverAndRestart(listener); if (!recovered) throw error; } result = await this.listenerApi.getStatus(); listener = result?.data?.listener || {}; account = result?.data?.account || {}; } this.listenerStarted = Boolean(listener.running); if (this.listenerStarted) this.startFriendWorker(); const listenerHealthy = listener.running && account.online !== false; if (listenerHealthy) this.lastRecoveryError = ''; this.onState({ personalPolling: { status: listenerHealthy ? 'running' : (requiresLogin ? 'needs_login' : 'waiting'), syncKey: Number(listener.syncKey || 0), startedAt: listener.startedAt || null, lastError: listenerHealthy ? '' : (requiresLogin ? account.statusText : '') || listener.lastError || this.lastRecoveryError, }, }); return listener; } async loop() { while (this.running) { try { await this.pollOnce(); } catch (error) { this.listenerStarted = false; this.onState({ personalPolling: { status: 'waiting', lastError: error.message } }); } if (this.running) await this.wait(this.config.retryMs); } } start() { if (this.running) return; this.running = true; this.onState({ personalPolling: { status: 'starting', lastError: '' } }); this.loopPromise = this.loop(); } async stop() { this.running = false; if (this.cancelWait) this.cancelWait(); if (this.listenerStarted) { try { await this.listenerApi.stop({ preserveAgentState: true }); } catch {} } this.listenerStarted = false; if (this.friendWorker) { try { this.friendWorker.kill(); } catch {} this.friendWorker = null; } this.onState({ personalPolling: { status: 'stopped' }, friendPolling: { status: 'stopped' }, }); await this.loopPromise; } }