/** * Gmail Watcher Service * * Automatically starts `gog gmail watch serve` when the gateway starts, * if hooks.gmail is configured with an account. */ import { type ChildProcess, spawn } from "node:child_process"; import process from "node:process"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { releaseChildProcessOutputAfterExit } from "../process/child-process.js"; import { formatCommandResult } from "../process/command-error.js"; import { runCommandWithTimeout } from "../process/exec.js"; import { killProcessTree } from "../process/kill-tree.js"; import { hasBinary } from "../skills/loading/config.js"; import { ensureTailscaleEndpoint } from "./gmail-setup-utils.js"; import { isAddressInUseError } from "./gmail-watcher-errors.js"; import { buildGogWatchServeLogArgs, buildGogWatchServeArgs, buildGogWatchStartArgs, type GmailHookRuntimeConfig, resolveGogExecutable, resolveGogServeInvocation, resolveGmailHookRuntimeConfig, } from "./gmail.js"; const log = createSubsystemLogger("gmail-watcher"); const GMAIL_WATCHER_STDERR_TAIL_CHARS = 512; let watcherProcess: ChildProcess | null = null; let renewInterval: ReturnType | null = null; let renewalInFlight: Promise | null = null; let renewalAbortController: AbortController | null = null; let shuttingDown = false; let currentConfig: GmailHookRuntimeConfig | null = null; let respawnTimeout: ReturnType | null = null; /** * Start the Gmail watch (registers with Gmail API) */ async function startGmailWatch( cfg: Pick, options: { signal?: AbortSignal } = {}, ): Promise { const args = [resolveGogExecutable(), ...buildGogWatchStartArgs(cfg)]; try { const result = await runCommandWithTimeout(args, { timeoutMs: 120_000, signal: options.signal, }); if (result.code !== 0) { log.error(formatCommandResult("gog gmail watch start", result)); return false; } log.info(`watch started for ${cfg.account}`); return true; } catch (err) { log.error(`watch start error: ${String(err)}`); return false; } } /** * Spawn the gog gmail watch serve process */ function spawnGogServe(cfg: GmailHookRuntimeConfig): ChildProcess { const args = buildGogWatchServeArgs(cfg); log.info(`starting gog ${buildGogWatchServeLogArgs(cfg).join(" ")}`); let addressInUse = false; let spawnFailed = false; // Carry a bounded tail so bind markers split across stderr chunks survive until close. let stderrTail = ""; const invocation = resolveGogServeInvocation(args); const child = spawn(invocation.command, invocation.args, { stdio: ["ignore", "pipe", "pipe"], // Own process group on Unix so killProcessTree can reach descendants on shutdown. detached: process.platform !== "win32", windowsHide: invocation.windowsHide, windowsVerbatimArguments: invocation.windowsVerbatimArguments, }); child.stdout?.on("error", (err) => { log.error(`gog stdout error: ${String(err)}`); }); child.stdout?.on("data", (data: Buffer) => { const line = data.toString().trim(); if (line) { log.info(`[gog] ${line}`); } }); child.stderr?.on("error", (err) => { log.error(`gog stderr error: ${String(err)}`); }); child.stderr?.on("data", (data: Buffer) => { const chunk = data.toString(); // Classify before truncation so a marker completed across the retention boundary survives. const combined = stderrTail + chunk; if (!addressInUse && isAddressInUseError(combined)) { addressInUse = true; } stderrTail = combined.slice(-GMAIL_WATCHER_STDERR_TAIL_CHARS); const line = chunk.trim(); if (!line) { return; } log.warn(`[gog] ${line}`); }); child.on("error", (err) => { // Failed spawn emits close without a pid; later errors on a running child remain retryable. if (child.pid === undefined) { spawnFailed = true; } log.error(`gog process error: ${String(err)}`); }); const releaseOutput = releaseChildProcessOutputAfterExit(child); child.once("exit", () => { // The detached POSIX group remains ours after its leader dies. Windows // taskkill requires a live root PID; only bound inherited-pipe drain there. if (!shuttingDown && watcherProcess === child && process.platform !== "win32" && child.pid) { killProcessTree(child.pid, { force: true, detached: true }); } }); // `close` follows bounded stdio drain, so late stderr still classifies bind failures. child.once("close", (code, signal) => { releaseOutput(); if (shuttingDown || watcherProcess !== child) { return; } if (spawnFailed) { watcherProcess = null; return; } if (addressInUse) { log.warn( "gog serve failed to bind (address already in use); stopping restarts. " + "Another watcher is likely running. Set OPENCLAW_SKIP_GMAIL_WATCHER=1 or stop the other process.", ); watcherProcess = null; return; } log.warn(`gog exited (code=${code}, signal=${signal}); restarting in 5s`); watcherProcess = null; respawnTimeout = setTimeout(() => { respawnTimeout = null; if (shuttingDown || !currentConfig) { return; } watcherProcess = spawnGogServe(currentConfig); }, 5000); }); return child; } /** * Signal the gog process tree to exit gracefully (SIGTERM, SIGKILL after 3 s) * and resolve on exit/close/error or a final 8 s safety timeout. */ function settleProcess(proc: ChildProcess): Promise { // A Windows root PID can be reused after exit, even while inherited pipes // delay close. The spawn owner's bounded drain releases those pipes safely. if (process.platform === "win32" && (proc.exitCode != null || proc.signalCode != null)) { return Promise.resolve(); } return new Promise((resolve) => { let settled = false; let processSettled = false; let graceElapsed = false; let graceTimer: ReturnType | undefined; const settle = () => { if (settled) { return; } settled = true; if (graceTimer) { clearTimeout(graceTimer); } if (finalTimeout) { clearTimeout(finalTimeout); } proc.removeListener("exit", settleAfterEscalation); proc.removeListener("close", settleAfterEscalation); proc.removeListener("error", settleAfterEscalation); resolve(); }; const settleAfterEscalation = () => { processSettled = true; if (graceElapsed) { settle(); } }; const finalTimeout = setTimeout(() => { if (!settled) { log.warn("gog process did not exit after SIGKILL; giving up"); settle(); } }, 8_000); proc.on("exit", settleAfterEscalation); proc.on("close", settleAfterEscalation); proc.on("error", settleAfterEscalation); // killProcessTree sends SIGTERM to the process group (Unix) or uses taskkill /T // (Windows) and escalates to SIGKILL after graceMs, reaching any descendants // spawned by gog that plain proc.kill() would miss. if (typeof proc.pid === "number") { const graceMs = 3_000; killProcessTree(proc.pid, { graceMs, detached: process.platform !== "win32", }); // killProcessTree owns escalation but intentionally unrefs its timer. // Keep shutdown referenced until that escalation has had a chance to run. graceTimer = setTimeout(() => { graceElapsed = true; if (processSettled) { settle(); } }, graceMs + 25); } else { // pid absent means spawn never started; direct kill clears any lingering state. try { proc.kill("SIGTERM"); } catch { /* process may not exist */ } graceElapsed = true; } }); } async function stopPeriodicRenewal(): Promise { if (renewInterval) { clearInterval(renewInterval); renewInterval = null; } const renewal = renewalInFlight; const controller = renewalAbortController; if (!renewal) { renewalAbortController = null; return; } controller?.abort(); await renewal; if (renewalInFlight === renewal) { renewalInFlight = null; } if (renewalAbortController === controller) { renewalAbortController = null; } } type GmailWatcherStartResult = { started: boolean; reason?: string; }; type GmailWatcherStartOptions = { signal?: AbortSignal; }; function cancelledGmailWatcherStart( expectedConfig: GmailHookRuntimeConfig, ): GmailWatcherStartResult { if (currentConfig === expectedConfig) { currentConfig = null; } return { started: false, reason: "startup cancelled" }; } /** * Start the Gmail watcher service. * Called automatically by the gateway if hooks.gmail is configured. */ export async function startGmailWatcher( cfg: OpenClawConfig, options: GmailWatcherStartOptions = {}, ): Promise { // Check if gmail hooks are configured if (!cfg.hooks?.enabled) { return { started: false, reason: "hooks not enabled" }; } if (!cfg.hooks?.gmail?.account) { return { started: false, reason: "no gmail account configured" }; } // Check if gog is available if (!hasBinary("gog")) { return { started: false, reason: "gog binary not found" }; } // Resolve the full runtime config const resolved = resolveGmailHookRuntimeConfig(cfg, {}); if (!resolved.ok) { return { started: false, reason: resolved.error }; } return startGmailWatcherService(resolved.value, options); } /** Start the shared watcher lifecycle after the caller resolves config and prerequisites. */ export async function startGmailWatcherService( runtimeConfig: GmailHookRuntimeConfig, options: GmailWatcherStartOptions = {}, ): Promise { if (options.signal?.aborted) { return cancelledGmailWatcherStart(runtimeConfig); } currentConfig = runtimeConfig; // Stop any existing watcher before doing async setup so a re-entry // does not orphan the old serve process or leave a dangling timer. // This must run before Tailscale/watch-start to prevent the old // process from exiting and queuing a respawn during async work. if (watcherProcess || renewInterval || renewalInFlight || respawnTimeout) { shuttingDown = true; if (respawnTimeout) { clearTimeout(respawnTimeout); respawnTimeout = null; } await stopPeriodicRenewal(); if (watcherProcess) { const oldProcess = watcherProcess; watcherProcess = null; await settleProcess(oldProcess); } shuttingDown = false; } // Set up Tailscale endpoint if needed if (runtimeConfig.tailscale.mode !== "off") { try { await ensureTailscaleEndpoint({ mode: runtimeConfig.tailscale.mode, path: runtimeConfig.tailscale.path, port: runtimeConfig.serve.port, signal: options.signal, target: runtimeConfig.tailscale.target, }); log.info( `tailscale ${runtimeConfig.tailscale.mode} configured for port ${runtimeConfig.serve.port}`, ); if (options.signal?.aborted) { return cancelledGmailWatcherStart(runtimeConfig); } } catch (err) { if (options.signal?.aborted) { return cancelledGmailWatcherStart(runtimeConfig); } log.error(`tailscale setup failed: ${String(err)}`); return { started: false, reason: `tailscale setup failed: ${String(err)}`, }; } } // Start the Gmail watch (register with Gmail API) const watchStarted = await startGmailWatch(runtimeConfig, { signal: options.signal }); if (options.signal?.aborted) { return cancelledGmailWatcherStart(runtimeConfig); } if (!watchStarted) { log.warn("gmail watch start failed, but continuing with serve"); } // Spawn the gog serve process shuttingDown = false; watcherProcess = spawnGogServe(runtimeConfig); const renewMs = runtimeConfig.renewEveryMinutes * 60_000; renewInterval = setInterval(() => { if (shuttingDown || renewalInFlight) { return; } const controller = new AbortController(); renewalAbortController = controller; const renewal = startGmailWatch(runtimeConfig, { signal: controller.signal }).finally(() => { if (renewalInFlight === renewal) { renewalInFlight = null; } if (renewalAbortController === controller) { renewalAbortController = null; } }); renewalInFlight = renewal; }, renewMs); log.info( `gmail watcher started for ${runtimeConfig.account} (renew every ${runtimeConfig.renewEveryMinutes}m)`, ); return { started: true }; } /** * Stop the Gmail watcher service. */ export async function stopGmailWatcher(): Promise { shuttingDown = true; if (respawnTimeout) { clearTimeout(respawnTimeout); respawnTimeout = null; } await stopPeriodicRenewal(); if (watcherProcess) { log.info("stopping gmail watcher"); const proc = watcherProcess; watcherProcess = null; await settleProcess(proc); } currentConfig = null; log.info("gmail watcher stopped"); }