openclaw/extensions/signal/src/monitor.ts

// Signal plugin module implements monitor behavior.
import { CHANNEL_APPROVAL_NATIVE_RUNTIME_CONTEXT_CAPABILITY } from "openclaw/plugin-sdk/approval-handler-adapter-runtime";
import type { PluginRuntime } from "openclaw/plugin-sdk/channel-core";
import { resolveChannelStreamingBlockEnabled } from "openclaw/plugin-sdk/channel-outbound";
import { registerChannelRuntimeContext } from "openclaw/plugin-sdk/channel-runtime-context";
import type {
  OpenClawConfig,
  ReplyToMode,
  SignalReactionNotificationMode,
} from "openclaw/plugin-sdk/config-contracts";
import {
  canonicalizeBase64,
  detectMime,
  estimateBase64DecodedBytes,
  saveMediaBuffer,
} from "openclaw/plugin-sdk/media-runtime";
import { DEFAULT_GROUP_HISTORY_LIMIT, type HistoryEntry } from "openclaw/plugin-sdk/reply-history";
import {
  deliverTextOrMediaReply,
  resolveSendableOutboundReplyParts,
} from "openclaw/plugin-sdk/reply-payload";
import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
import {
  chunkTextWithMode,
  resolveChunkMode,
  resolveTextChunkLimit,
} from "openclaw/plugin-sdk/reply-runtime";
import { getRuntimeConfig } from "openclaw/plugin-sdk/runtime-config-snapshot";
import {
  createNonExitingRuntime,
  type BackoffPolicy,
  type RuntimeEnv,
} from "openclaw/plugin-sdk/runtime-env";
import {
  resolveAllowlistProviderRuntimeGroupPolicy,
  resolveDefaultGroupPolicy,
  warnMissingProviderGroupPolicyFallbackOnce,
} from "openclaw/plugin-sdk/runtime-group-policy";
import {
  normalizeOptionalString,
  normalizeStringEntries,
} from "openclaw/plugin-sdk/string-coerce-runtime";
import { normalizeE164 } from "openclaw/plugin-sdk/text-utility-runtime";
import { waitForTransportReady } from "openclaw/plugin-sdk/transport-ready-runtime";
import { resolveSignalAccount, resolveSignalReplyToMode } from "./accounts.js";
import { isSignalNativeApprovalHandlerConfigured } from "./approval-native.js";
import { addSignalApprovalReactionHintToStructuredPayload } from "./approval-reactions.js";
import { signalRpcRequest } from "./client-adapter.js";
import type { SignalTransportKind } from "./client-adapter.js";
import { createSignalDaemonLifecycle } from "./daemon-lifecycle.js";
import {
  assertSignalDaemonEndpointAvailable,
  spawnSignalDaemon,
  type SignalDaemonHandle,
  waitForSignalDaemonReady,
} from "./daemon.js";
import { isSignalSenderAllowed, type resolveSignalSender } from "./identity.js";
import { createSignalEventHandler } from "./monitor/event-handler.js";
import type {
  SignalAttachment,
  SignalNativeReplyContext,
  SignalReactionMessage,
  SignalReactionTarget,
} from "./monitor/event-handler.types.js";
import { createSignalNativeReplyIdResolver } from "./native-reply.js";
import { materializeSignalPresentationFallback } from "./presentation-fallback.js";
import { registerSignalReactionTargetsForDeliveredPayload } from "./reaction-targets.js";
import { sendMessageSignal } from "./send.js";
import { startSignalIngressMonitor, type SignalIngressMonitor } from "./signal-ingress.js";
import {
  publishSignalRecovering,
  runSignalSseLoop,
  type SignalStatusSink,
} from "./sse-reconnect.js";
import { normalizeSignalTransportHost } from "./transport-url.js";

export type MonitorSignalOpts = {
  runtime?: RuntimeEnv;
  abortSignal?: AbortSignal;
  account?: string;
  accountId?: string;
  config?: OpenClawConfig;
  baseUrl?: string;
  channelRuntime?: PluginRuntime["channel"];
  autoStart?: boolean;
  startupTimeoutMs?: number;
  cliPath?: string;
  configPath?: string;
  httpHost?: string;
  httpPort?: number;
  receiveMode?: "on-start" | "manual";
  ignoreAttachments?: boolean;
  ignoreStories?: boolean;
  sendReadReceipts?: boolean;
  allowFrom?: Array<string | number>;
  groupAllowFrom?: Array<string | number>;
  mediaMaxMb?: number;
  reconnectPolicy?: Partial<BackoffPolicy>;
  waitForTransportReady?: typeof waitForTransportReady;
  statusSink?: SignalStatusSink;
};

function createSignalMonitorTaskRunner(runtime: RuntimeEnv) {
  const inFlight = new Set<Promise<void>>();
  return {
    runTask(task: () => Promise<void>): Promise<void> {
      const trackedTask = Promise.resolve().then(task);
      inFlight.add(trackedTask);
      void trackedTask.catch((err: unknown) =>
        runtime.error?.(`signal monitor task failed: ${String(err)}`),
      );
      void trackedTask.finally(() => inFlight.delete(trackedTask)).catch(() => undefined);
      return trackedTask;
    },
    async waitForIdle(): Promise<void> {
      while (inFlight.size > 0) {
        await Promise.allSettled(inFlight);
      }
    },
  };
}

function resolveSignalReactionTargets(reaction: SignalReactionMessage): SignalReactionTarget[] {
  const targets: SignalReactionTarget[] = [];
  const uuid = reaction.targetAuthorUuid?.trim();
  if (uuid) {
    targets.push({ kind: "uuid", id: uuid, display: `uuid:${uuid}` });
  }
  const author = reaction.targetAuthor?.trim();
  if (author) {
    const normalized = normalizeE164(author);
    targets.push({ kind: "phone", id: normalized, display: normalized });
  }
  return targets;
}

function isSignalReactionMessage(
  reaction: SignalReactionMessage | null | undefined,
): reaction is SignalReactionMessage {
  if (!reaction) {
    return false;
  }
  const emoji = reaction.emoji?.trim();
  const timestamp = reaction.targetSentTimestamp;
  const hasTarget = Boolean(
    normalizeOptionalString(reaction.targetAuthor) ||
    normalizeOptionalString(reaction.targetAuthorUuid),
  );
  return Boolean(emoji && typeof timestamp === "number" && timestamp > 0 && hasTarget);
}

function shouldEmitSignalReactionNotification(params: {
  mode?: SignalReactionNotificationMode;
  account?: string | null;
  accountUuid?: string | null;
  targets?: SignalReactionTarget[];
  sender?: ReturnType<typeof resolveSignalSender> | null;
  allowlist?: string[];
}) {
  const { mode, account, accountUuid, targets, sender, allowlist } = params;
  const effectiveMode = mode ?? "own";
  if (effectiveMode === "off") {
    return false;
  }
  if (effectiveMode === "own") {
    const accountId = normalizeOptionalString(account);
    const normalizedAccountUuid = normalizeOptionalString(accountUuid);
    if ((!accountId && !normalizedAccountUuid) || !targets || targets.length === 0) {
      return false;
    }
    const normalizedAccount = accountId ? normalizeE164(accountId) : undefined;
    return targets.some((target) => {
      if (target.kind === "uuid") {
        // UUID-only reaction payloads omit the phone identity carried by account.
        return [accountId, normalizedAccountUuid].some(
          (candidate) => candidate === target.id || candidate === `uuid:${target.id}`,
        );
      }
      return Boolean(normalizedAccount) && normalizedAccount === target.id;
    });
  }
  if (effectiveMode === "allowlist") {
    if (!sender || !allowlist || allowlist.length === 0) {
      return false;
    }
    return isSignalSenderAllowed(sender, allowlist);
  }
  return true;
}

function buildSignalReactionSystemEventText(params: {
  emojiLabel: string;
  actorLabel: string;
  messageId: string;
  targetLabel?: string;
  groupLabel?: string;
}) {
  const base = `Signal reaction added: ${params.emojiLabel} by ${params.actorLabel} msg ${params.messageId}`;
  const withTarget = params.targetLabel ? `${base} from ${params.targetLabel}` : base;
  return params.groupLabel ? `${withTarget} in ${params.groupLabel}` : withTarget;
}

const SIGNAL_ATTACHMENT_RPC_RESPONSE_HEADROOM_BYTES = 64 * 1024;
const SIGNAL_BASE64_OVERHEAD_NUMERATOR = 4;
const SIGNAL_BASE64_OVERHEAD_DENOMINATOR = 3;

function deriveSignalAttachmentRpcMaxResponseBytes(maxBytes: number): number | undefined {
  if (!Number.isFinite(maxBytes) || maxBytes <= 0) {
    return undefined;
  }
  const base64Bytes = Math.ceil(
    (maxBytes * SIGNAL_BASE64_OVERHEAD_NUMERATOR) / SIGNAL_BASE64_OVERHEAD_DENOMINATOR,
  );
  return base64Bytes + SIGNAL_ATTACHMENT_RPC_RESPONSE_HEADROOM_BYTES;
}

async function fetchAttachment(params: {
  baseUrl: string;
  account?: string;
  transportKind?: SignalTransportKind;
  attachment: SignalAttachment;
  sender?: string;
  groupId?: string;
  maxBytes: number;
}): Promise<{ path: string; contentType?: string } | null> {
  const { attachment } = params;
  if (!attachment?.id) {
    return null;
  }
  if (typeof attachment.size === "number" && attachment.size > params.maxBytes) {
    throw new Error(
      `Signal attachment ${attachment.id} exceeds ${(params.maxBytes / (1024 * 1024)).toFixed(0)}MB limit`,
    );
  }
  const rpcParams: Record<string, unknown> = {
    id: attachment.id,
  };
  if (params.account) {
    rpcParams.account = params.account;
  }
  if (params.groupId) {
    rpcParams.groupId = params.groupId;
  } else if (params.sender) {
    rpcParams.recipient = params.sender;
  } else {
    return null;
  }

  const result = await signalRpcRequest<{ data?: string }>("getAttachment", rpcParams, {
    baseUrl: params.baseUrl,
    maxResponseBytes: deriveSignalAttachmentRpcMaxResponseBytes(params.maxBytes),
    transportKind: params.transportKind,
  });
  if (!result?.data) {
    return null;
  }
  if (estimateBase64DecodedBytes(result.data) > params.maxBytes) {
    throw new Error(
      `Signal attachment ${attachment.id} exceeds ${(params.maxBytes / (1024 * 1024)).toFixed(0)}MB limit`,
    );
  }
  const canonicalData = canonicalizeBase64(result.data);
  if (!canonicalData) {
    throw new Error(`Signal attachment ${attachment.id} returned malformed base64 data`);
  }
  const buffer = Buffer.from(canonicalData, "base64");
  const originalFilename = normalizeOptionalString(attachment.filename ?? undefined);
  const contentType =
    normalizeOptionalString(attachment.contentType ?? undefined) ??
    (await detectMime({ buffer, filePath: originalFilename }));
  const saved = await saveMediaBuffer(
    buffer,
    contentType,
    "inbound",
    params.maxBytes,
    originalFilename,
  );
  return { path: saved.path, contentType: saved.contentType };
}

export async function deliverReplies(params: {
  cfg: OpenClawConfig;
  replies: ReplyPayload[];
  target: string;
  baseUrl: string;
  account?: string;
  accountUuid?: string;
  accountId?: string;
  runtime: RuntimeEnv;
  maxBytes: number;
  textLimit: number;
  chunkMode: "length" | "newline";
  replyContext?: SignalNativeReplyContext;
  chatType?: "direct" | "group";
}) {
  const {
    replies,
    target,
    baseUrl,
    account,
    accountUuid,
    accountId,
    runtime,
    maxBytes,
    textLimit,
    chunkMode,
  } = params;
  const replyToMode = resolveSignalReplyToMode({
    cfg: params.cfg,
    accountId,
    chatType: params.chatType,
  });
  for (const payload of replies) {
    const deliveryResults: Array<{
      channel: "signal";
      messageId: string;
      meta: { signalVisibleText: string };
    }> = [];
    const presentationPayload = materializeSignalPresentationFallback(payload);
    const deliveredPayload =
      addSignalApprovalReactionHintToStructuredPayload({
        cfg: params.cfg,
        accountId,
        to: target,
        payload: presentationPayload,
        targetAuthor: account,
        targetAuthorUuid: accountUuid,
      }) ?? presentationPayload;
    const reply = resolveSendableOutboundReplyParts(deliveredPayload);
    const nextNativeReply = createSignalNativeReplyResolver({
      payload: deliveredPayload,
      replyContext: params.replyContext,
      replyToMode,
    });
    const recordDeliveryResult = (
      result: Awaited<ReturnType<typeof sendMessageSignal>>,
      visibleText: string,
    ) => {
      const messageId =
        typeof result?.messageId === "string" && result.messageId.trim()
          ? result.messageId.trim()
          : null;
      if (messageId) {
        deliveryResults.push({
          channel: "signal",
          messageId,
          meta: { signalVisibleText: visibleText },
        });
      }
    };
    const delivered = await deliverTextOrMediaReply({
      payload: deliveredPayload,
      text: reply.text,
      chunkText: (value) => chunkTextWithMode(value, textLimit, chunkMode),
      sendText: async (chunk) => {
        recordDeliveryResult(
          await sendMessageSignal(target, chunk, {
            cfg: params.cfg,
            baseUrl,
            account,
            maxBytes,
            accountId,
            ...nextNativeReply(),
          }),
          chunk,
        );
      },
      sendMedia: async ({ mediaUrl, caption }) => {
        const visibleText = caption ?? "";
        recordDeliveryResult(
          await sendMessageSignal(target, visibleText, {
            cfg: params.cfg,
            baseUrl,
            account,
            mediaUrl,
            maxBytes,
            accountId,
            ...nextNativeReply(),
          }),
          visibleText,
        );
      },
    });
    if (delivered !== "empty") {
      registerSignalReactionTargetsForDeliveredPayload({
        cfg: params.cfg,
        target: {
          channel: "signal",
          to: target,
          accountId,
        },
        payload: deliveredPayload,
        results: deliveryResults,
        targetAuthor: account,
        targetAuthorUuid: accountUuid,
      });
      runtime.log?.(`delivered reply to ${target}`);
    }
  }
}

function createSignalNativeReplyResolver(params: {
  payload: ReplyPayload;
  replyContext?: SignalNativeReplyContext;
  replyToMode: ReplyToMode;
}): () => Pick<
  Parameters<typeof sendMessageSignal>[2],
  "replyToId" | "replyToAuthor" | "replyToBody"
> {
  const nextReplyToId = createSignalNativeReplyIdResolver(params);
  return () => {
    const replyToId = nextReplyToId();
    if (!replyToId) {
      return {};
    }
    const replyToAuthor = normalizeOptionalString(params.replyContext?.author);
    return {
      replyToId,
      ...(replyToAuthor ? { replyToAuthor, replyToBody: params.replyContext?.body ?? "" } : {}),
    };
  };
}

export async function monitorSignalProvider(opts: MonitorSignalOpts = {}): Promise<void> {
  const runtime = opts.runtime ?? createNonExitingRuntime();
  const cfg = opts.config ?? getRuntimeConfig();
  const accountInfo = resolveSignalAccount({
    cfg,
    accountId: opts.accountId,
  });
  const historyLimit = Math.max(
    0,
    accountInfo.config.historyLimit ??
      cfg.messages?.groupChat?.historyLimit ??
      DEFAULT_GROUP_HISTORY_LIMIT,
  );
  const groupHistories = new Map<string, HistoryEntry[]>();
  const textLimit = resolveTextChunkLimit(cfg, "signal", accountInfo.accountId);
  const chunkMode = resolveChunkMode(cfg, "signal", accountInfo.accountId);
  const baseUrl = normalizeOptionalString(opts.baseUrl) ?? accountInfo.baseUrl;
  const account =
    normalizeOptionalString(opts.account) ?? normalizeOptionalString(accountInfo.config.account);
  const dmPolicy = accountInfo.config.dmPolicy ?? "pairing";
  const allowFrom = normalizeStringEntries(opts.allowFrom ?? accountInfo.config.allowFrom);
  const groupAllowFrom = normalizeStringEntries(
    opts.groupAllowFrom ??
      accountInfo.config.groupAllowFrom ??
      (accountInfo.config.allowFrom && accountInfo.config.allowFrom.length > 0
        ? accountInfo.config.allowFrom
        : []),
  );
  const defaultGroupPolicy = resolveDefaultGroupPolicy(cfg);
  const { groupPolicy, providerMissingFallbackApplied } =
    resolveAllowlistProviderRuntimeGroupPolicy({
      providerConfigPresent: cfg.channels?.signal !== undefined,
      groupPolicy: accountInfo.config.groupPolicy,
      defaultGroupPolicy,
    });
  warnMissingProviderGroupPolicyFallbackOnce({
    providerMissingFallbackApplied,
    providerKey: "signal",
    accountId: accountInfo.accountId,
    log: (message) => runtime.log?.(message),
  });
  const reactionMode = accountInfo.config.reactionNotifications ?? "own";
  const reactionAllowlist = normalizeStringEntries(accountInfo.config.reactionAllowlist);
  const mediaMaxBytes = (opts.mediaMaxMb ?? accountInfo.config.mediaMaxMb ?? 8) * 1024 * 1024;
  const transportKind = accountInfo.transport.kind;
  const managedTransport =
    accountInfo.transport.kind === "managed-native" ? accountInfo.transport : undefined;
  const ignoreAttachments = opts.ignoreAttachments ?? accountInfo.config.ignoreAttachments ?? false;
  const sendReadReceipts = Boolean(opts.sendReadReceipts ?? accountInfo.config.sendReadReceipts);
  const waitForTransportReadyFn = opts.waitForTransportReady ?? waitForTransportReady;

  const autoStart = Boolean(managedTransport) && (opts.autoStart ?? true);
  const startupTimeoutMs = Math.min(
    120_000,
    Math.max(1_000, opts.startupTimeoutMs ?? managedTransport?.startupTimeoutMs ?? 30_000),
  );
  const readReceiptsViaDaemon = autoStart && sendReadReceipts;
  const daemonLifecycle = createSignalDaemonLifecycle({ abortSignal: opts.abortSignal });
  const monitorTaskRunner = createSignalMonitorTaskRunner(runtime);
  let daemonHandle: SignalDaemonHandle | null = null;
  let ingressMonitor: SignalIngressMonitor | undefined;
  const startupDeadline = Date.now() + startupTimeoutMs;

  if (autoStart) {
    const cliPath = opts.cliPath ?? managedTransport?.cliPath ?? "signal-cli";
    const configPath =
      normalizeOptionalString(opts.configPath) ??
      normalizeOptionalString(managedTransport?.configPath);
    const httpHost = normalizeSignalTransportHost(
      opts.httpHost ?? managedTransport?.httpHost ?? "127.0.0.1",
    );
    const httpPort = opts.httpPort ?? managedTransport?.httpPort ?? 8080;
    const startupTimeoutSignal = AbortSignal.timeout(startupTimeoutMs);
    const endpointProbeSignal = opts.abortSignal
      ? AbortSignal.any([opts.abortSignal, startupTimeoutSignal])
      : startupTimeoutSignal;
    // Readiness alone cannot prove ownership: an unrelated service can answer /api/v1/check
    // while signal-cli exits on EADDRINUSE. Probe the configured bind before starting it.
    try {
      await assertSignalDaemonEndpointAvailable({
        httpHost,
        httpPort,
        abortSignal: endpointProbeSignal,
      });
    } catch (error) {
      if (opts.abortSignal?.aborted) {
        return;
      }
      if (startupTimeoutSignal.aborted || Date.now() >= startupDeadline) {
        throw new Error(
          `signal daemon startup timed out after ${startupTimeoutMs}ms while checking its endpoint`,
          { cause: error },
        );
      }
      throw error;
    }
    // Abort can land after the probe resolves but before this continuation resumes.
    // Recheck at the spawn boundary so a cancelled monitor never creates a daemon.
    if (opts.abortSignal?.aborted) {
      return;
    }
    if (Date.now() >= startupDeadline) {
      throw new Error(
        `signal daemon startup timed out after ${startupTimeoutMs}ms before starting`,
      );
    }
    daemonHandle = spawnSignalDaemon({
      cliPath,
      ...(configPath ? { configPath } : {}),
      account,
      httpHost,
      httpPort,
      receiveMode: opts.receiveMode ?? managedTransport?.receiveMode,
      ignoreAttachments: opts.ignoreAttachments ?? accountInfo.config.ignoreAttachments,
      ignoreStories: opts.ignoreStories ?? managedTransport?.ignoreStories,
      sendReadReceipts,
      runtime,
    });
    daemonLifecycle.attach(daemonHandle);
  }

  const onAbort = () => void daemonLifecycle.stop();
  opts.abortSignal?.addEventListener("abort", onAbort, { once: true });

  try {
    if (daemonHandle) {
      await waitForSignalDaemonReady({
        baseUrl,
        abortSignal: daemonLifecycle.abortSignal,
        startupDeadlineMs: startupDeadline,
        logAfterMs: 10_000,
        logIntervalMs: 10_000,
        runtime,
        waitForTransportReadyFn,
      });
      const daemonExitError = daemonLifecycle.getExitError();
      if (daemonExitError) {
        throw daemonExitError;
      }
    }

    registerChannelRuntimeContext({
      channelRuntime: opts.channelRuntime,
      channelId: "signal",
      accountId: accountInfo.accountId,
      capability: CHANNEL_APPROVAL_NATIVE_RUNTIME_CONTEXT_CAPABILITY,
      context: isSignalNativeApprovalHandlerConfigured({
        cfg,
        accountId: accountInfo.accountId,
      })
        ? {
            accountId: accountInfo.accountId,
            baseUrl,
            account,
            accountUuid: accountInfo.config.accountUuid,
          }
        : null,
      abortSignal: opts.abortSignal,
    });

    const handleEvent = createSignalEventHandler({
      runtime,
      channelRuntime: opts.channelRuntime,
      abortSignal: daemonLifecycle.abortSignal,
      runTrackedTask: (task) => {
        void monitorTaskRunner.runTask(task);
      },
      cfg,
      baseUrl,
      account,
      accountUuid: accountInfo.config.accountUuid,
      accountId: accountInfo.accountId,
      blockStreaming: resolveChannelStreamingBlockEnabled(accountInfo.config),
      historyLimit,
      groupHistories,
      textLimit,
      dmPolicy,
      allowFrom,
      groupAllowFrom,
      groupPolicy,
      reactionMode,
      reactionAllowlist,
      mediaMaxBytes,
      ignoreAttachments,
      sendReadReceipts,
      readReceiptsViaDaemon,
      fetchAttachment: (params) => fetchAttachment({ ...params, transportKind }),
      deliverReplies: (params) => deliverReplies({ ...params, cfg, chunkMode }),
      resolveSignalReactionTargets,
      isSignalReactionMessage,
      shouldEmitSignalReactionNotification,
      buildSignalReactionSystemEventText,
    });

    ingressMonitor = await startSignalIngressMonitor({
      accountId: accountInfo.accountId,
      dispatch: handleEvent,
      runtime,
    });

    await runSignalSseLoop({
      baseUrl,
      account,
      abortSignal: daemonLifecycle.abortSignal,
      runtime,
      // signal-cli can keep the SSE event endpoint idle until the next inbound event.
      timeoutMs: 0,
      transportKind,
      policy: opts.reconnectPolicy,
      statusSink: opts.statusSink,
      onEvent: (event) =>
        monitorTaskRunner.runTask(async () => await ingressMonitor?.receive(event)),
    });
    const daemonExitError = daemonLifecycle.getExitError();
    if (daemonExitError) {
      throw daemonExitError;
    }
  } catch (err) {
    const daemonExitError = daemonLifecycle.getExitError();
    if (opts.abortSignal?.aborted && !daemonExitError) {
      return;
    }
    if (daemonExitError) {
      publishSignalRecovering(opts.statusSink, daemonExitError.message);
    }
    throw err;
  } finally {
    await ingressMonitor?.stop();
    // Daemon attachment finishes before monitor tasks start. Keep teardown open until both the
    // child has exited and already-started reply work has drained.
    await Promise.all([daemonLifecycle.stop(), monitorTaskRunner.waitForIdle()]);
    opts.abortSignal?.removeEventListener("abort", onAbort);
  }
}