channel-pull

1import type { Register } from "claude-code";
2
3/**
4 * Pulls Telegram messages from this session's channel-relay and submits each
5 * as a prompt, for sessions that can't receive channel notifications (the
6 * Claude desktop app). The relay only offers the pull socket when its parent
7 * claude runs without channels, so a CLI `cc` session is left alone.
8 */
9
10type RelayPortFile = {
11  sessionId?: string;
12  pullSocket?: string;
13  channelsEnabled?: boolean;
14  startedAt?: string;
15};
16
17type Inbound = { content: string; meta: Record<string, string> };
18
19// The relay caps a long-poll at 8s.
20const POLL_MS = 8_000;
21const RETRY_MS = 5_000;
22
23const escapeAttr = (v: string) =>
24  v.replace(/&/g, "&amp;").replace(/"/g, "&quot;").replace(/</g, "&lt;");
25
26/**
27 * The same `<channel>` tag a channels-enabled session sees. A `<channel` or
28 * `</channel` inside the message is escaped so the text can't close the tag
29 * and forge a second one (another `user=`).
30 */
31export const toChannelPrompt = (m: Inbound): string => {
32  const attrs = Object.entries({ source: "channel-relay", ...m.meta })
33    .map(([k, v]) => `${k}="${escapeAttr(String(v))}"`)
34    .join(" ");
35  const body = m.content.replace(/<(\/?channel)/gi, "&lt;$1");
36  return `<channel ${attrs}>\n${body}\n</channel>`;
37};
38
39/** This session's relay: the newest port file naming its id (a SIGKILLed relay leaves a stale one). */
40export const pickRelay = (
41  ports: RelayPortFile[],
42  sessionId: string,
43): RelayPortFile | undefined =>
44  ports
45    .filter((p) => p.sessionId === sessionId)
46    .sort((a, b) => (b.startedAt ?? "").localeCompare(a.startedAt ?? ""))[0];
47
48export const register: Register = (on) => {
49  on("session.start", async ($, e, next) => {
50    const started = await next(e);
51
52    const findRelay = async (): Promise<RelayPortFile | undefined> => {
53      const dir =
54        (await $.env.get("CLAUDE_TELEGRAM_STATE_DIR")) ??
55        `${await $.env.get("HOME")}/.claude-mobile-bridge`;
56      const sessionId = await $.session.id();
57      const files = (await $.fs.list(dir)).filter(
58        (f) => f.kind === "file" && /^channel-relay-.*\.json$/.test(f.name),
59      );
60      const ports: RelayPortFile[] = [];
61      for (const f of files) {
62        try {
63          ports.push(JSON.parse(await $.fs.read(`${dir}/${f.name}`)));
64        } catch {
65          // a port file mid-write or already gone; next pass reads it
66        }
67      }
68      return pickRelay(ports, sessionId);
69    };
70
71    let connected: string | undefined;
72    const drop = (why: string) => {
73      if (connected)
74        $.ui.log(`channel-pull: relay lost (${why})`, { to: "debug" });
75      connected = undefined;
76      $.ui.status(undefined);
77    };
78
79    /** One poll; resolves the delay before the next, or undefined to stop. */
80    const tick = async (): Promise<number | undefined> => {
81      const relay = await findRelay().catch(() => undefined);
82      if (relay?.channelsEnabled) {
83        $.ui.log("channel-pull: channels enabled, relay notifies directly", {
84          to: "debug",
85        });
86        return undefined;
87      }
88      if (!relay?.pullSocket) {
89        drop("no relay for this session");
90        return RETRY_MS;
91      }
92      try {
93        const res = await $.http.fetch(
94          `http://relay/inbox?timeout=${POLL_MS}`,
95          { socketPath: relay.pullSocket },
96        );
97        if (!res.ok) throw new Error(`HTTP ${res.status}`);
98        if (connected !== relay.pullSocket) {
99          connected = relay.pullSocket;
100          $.ui.status("📡 telegram");
101          $.ui.log(`channel-pull: polling ${connected}`, { to: "debug" });
102        }
103        const { messages } = JSON.parse(res.text) as { messages: Inbound[] };
104        for (const m of messages) {
105          void $.prompt
106            .submit({ text: toChannelPrompt(m), asUser: true })
107            .catch((err) => $.ui.log(`channel-pull: submit failed: ${err}`));
108        }
109        return 0;
110      } catch (err) {
111        drop(String(err));
112        return RETRY_MS;
113      }
114    };
115
116    // A timer chain, not a loop: the engine drops a module's timers when it
117    // reloads, so a reloaded copy never polls alongside the old one.
118    const schedule = (ms: number): void => {
119      $.clock.after(ms, () => {
120        void tick().then(
121          (delay) => {
122            if (delay !== undefined) schedule(delay);
123          },
124          (err) => $.ui.log(`channel-pull: stopped: ${err}`),
125        );
126      });
127    };
128    schedule(0);
129    return started;
130  });
131};
132