telegram-reaction

1import type { EngineInterface, Register } from 'claude-code'
2
3// telegram-reaction -- an OPT-IN Claude Code mod: the progress of a Telegram
4// turn shows as a reaction on the user's OWN message, instead of a separate
5// "✍️ working on it…" placeholder message:
6//
7//   👀  received        prompt.submit from the Telegram channel
8//   🤔  working         turn.start, or the first tool call that is not the reply
9//   ✍️  writing         the Telegram `reply` tool is about to run
10//   👨‍💻  after           the `reply` tool SUCCEEDED -- the answer is out, the turn is
11//                       still running; it STAYS while the turn goes on (more tool
12//                       calls, more replies)
13//   (cleared)           the turn completed -- not the first reply: an agent often
14//                       answers in several messages, and the user reads the
15//                       reaction as "still working" until the turn is over
16//
17// The reaction only ever moves forward (👀 -> 🤔 -> ✍️ -> 👨‍💻); it never flips back.
18// The `after` step is switched off with MARVEEN_TELEGRAM_REACTION_EMOJIS='after=': then the 0.3.2 behaviour is
19// unchanged to the call (✍️ stays to the turn end; the step does not even move the pending stage).
20//
21// Each step's emoji can be changed or the step skipped:
22// MARVEEN_TELEGRAM_REACTION_EMOJIS='received=👀,working=🤔,writing=✍️,after=👨‍💻' -- a step left out keeps its default; an empty
23// value ('working=') skips the step: no change at that point (the previous reaction stays; at the first step: none
24// yet). The first three empty turn the mod off (the cleanup still runs; `after` alone does not keep it on). An unknown step, a malformed item or an emoji
25// Telegram does not accept as a reaction is a loud warning in the session transcript ($.ui.log), and that step keeps
26// its default. Read at session start: a change needs a restart of the main agent.
27//
28// The agent's OWN reaction (its react tool) is cleared at the turn end too, by default (a reaction left on a message
29// is listed by Telegram as unread); MARVEEN_TELEGRAM_REACTION_KEEP_AGENT=1 keeps it (the 0.2.x behaviour). Either way
30// the message stays the agent's: no stage of ours on it again.
31//
32// Where it is loaded it is ON: when installed as a plugin next to an unmodified engine, it is switched on for the
33// main agent through the engine's .env CHANNEL_PLUGINS_EXTRA; MARVEEN_TELEGRAM_REACTION=0
34// turns it off, FLEET_FORBID_LIVE=1 (a test process tree) makes it inert. The
35// engine's stock placeholder hook (scripts/hooks/telegram_progress.py) is kept off
36// through its own dedup marker, and the mod records the pending reply in the
37// stock hook's format, so the Stop hook's delivery enforcement and the watchdog keep
38// working unchanged (see claimForFactoryHook).
39//
40// What it never does: it never denies or rewrites an event, and it delays only
41// one: prompt.submit waits for the claim. Every Bot API call is fire-and-forget
42// on ONE serial chain, and it sends the DESIRED reaction of its message as it is
43// when the call runs (our stage, the agent's own emoji once the agent reacted on
44// the message itself, or none) -- never an earlier snapshot. When the agent's
45// own react succeeds, the chain writes the agent's emoji again after everything
46// it holds. A call with no certain answer leaves the message "unknown": a call
47// that timed out syncs its message again when it finally settles (it may have
48// landed after newer calls), and like a failed clear it is retried at the next
49// turn end and the next session start. So once the chain is drained and every
50// request has settled, each message carries its desired reaction.
51//
52// The bot token comes from <TELEGRAM_STATE_DIR>/.env, the same file the
53// telegram plugin and the placeholder hooks read. It is held in memory only:
54// never stored, logged or shown.
55
56// The steps of a turn, in their only order (the rank is the STEP's, not the emoji's: two steps may share an emoji).
57export const STEPS = ['received', 'working', 'writing', 'after'] as const
58export type Step = (typeof STEPS)[number]
59const RANK: Record<Step, number> = { received: 0, working: 1, writing: 2, after: 3 }
60export const DEFAULT_EMOJIS: Readonly<Record<Step, string>> = { received: '👀', working: '🤔', writing: '✍️', after: '👨‍💻' }
61// the steps of the 0.3.2 ladder: all of them empty turns the mod off (the `after` step alone never keeps it on)
62const LADDER_032 = ['received', 'working', 'writing'] as const
63// The Bot API ReactionTypeEmoji whitelist, as the telegram plugin 0.0.7's own dependency types it (@grammyjs/types
64// 3.25.0, message.d.ts, read 2026-10-12). Compared without the variation selector U+FE0F: the default ✍️ carries it,
65// the list does not, and Telegram takes both.
66const ALLOWED_REACTIONS = [
67  '👍', '👎', '❤', '🔥', '🥰', '👏', '😁', '🤔', '🤯', '😱', '🤬', '😢', '🎉', '🤩', '🤮', '💩', '🙏', '👌', '🕊', '🤡',
68  '🥱', '🥴', '😍', '🐳', '❤‍🔥', '🌚', '🌭', '💯', '🤣', '⚡', '🍌', '🏆', '💔', '🤨', '😐', '🍓', '🍾', '💋', '🖕', '😈',
69  '😴', '😭', '🤓', '👻', '👨‍💻', '👀', '🎃', '🙈', '😇', '😨', '🤝', '✍', '🤗', '🫡', '🎅', '🎄', '☃', '💅', '🤪', '🗿',
70  '🆒', '💘', '🙉', '🦄', '😘', '💊', '🙊', '😎', '👾', '🤷‍♂', '🤷', '🤷‍♀', '😡',
71]
72const noVs = (e: string) => e.replace(/\uFE0F/g, '')
73const ALLOWED_BY_FORM = new Map(ALLOWED_REACTIONS.map(e => [noVs(e), e]))
74/** User text in a warning is cut to WARN_ITEM_MAX code points -- a 5000-emoji item must not
75 *  write itself into the transcript at full length */
76const WARN_ITEM_MAX = 64
77const WARN_LINES_MAX = 8
78const cut = (t: string) => { const cp = [...t]; return cp.length > WARN_ITEM_MAX ? cp.slice(0, WARN_ITEM_MAX).join('') + '…' : t }
79
80/** MARVEEN_TELEGRAM_REACTION_EMOJIS -> the emoji of each step ('' = the step is skipped) and the warnings to show.
81 *  Missing or empty: the defaults. Each item is <step>=<emoji>; an unknown step or a malformed item is skipped, an emoji
82 *  Telegram does not accept leaves that step at its default -- all of them with a warning, never silently. */
83export function emojisFrom(value: unknown): { emojis: Record<Step, string>; warnings: string[] } {
84  const emojis: Record<Step, string> = { ...DEFAULT_EMOJIS }
85  const warnings: string[] = []
86  let text = typeof value === 'string' ? value.trim() : ''
87  // a value written shell-style ('...' or "...") is taken without its outer quotes
88  if (text.length >= 2 && (text[0] === "'" || text[0] === '"') && text[text.length - 1] === text[0]) text = text.slice(1, -1).trim()
89  if (!text) return { emojis, warnings }
90  const given = new Set<string>()
91  for (const raw of text.split(',')) {
92    const item = raw.trim()
93    if (!item) continue
94    const eq = item.indexOf('=')
95    if (eq < 1) {
96      warnings.push(`"${cut(item)}" is not <step>=<emoji> -- ignored`)
97      continue
98    }
99    const name = item.slice(0, eq).trim()
100    const emoji = item.slice(eq + 1).trim()
101    if (!(STEPS as readonly string[]).includes(name)) {
102      warnings.push(`unknown step "${cut(name)}" (received, working, writing, after) -- ignored`)
103      continue
104    }
105    const step = name as Step
106    if (given.has(step)) warnings.push(`"${step}" is given more than once -- the last one counts`)
107    given.add(step)
108    if (emoji === '') {
109      emojis[step] = ''
110      continue
111    }
112    const allowed = ALLOWED_BY_FORM.get(noVs(emoji))
113    if (!allowed) {
114      emojis[step] = DEFAULT_EMOJIS[step]
115      warnings.push(`"${step}=${cut(emoji)}": Telegram does not accept this as a reaction -- the default ${DEFAULT_EMOJIS[step]} stays`)
116      continue
117    }
118    // the default keeps its exact form (the one measured live); any other emoji is sent in the Bot API list's form
119    emojis[step] = noVs(emoji) === noVs(DEFAULT_EMOJIS[step]) ? DEFAULT_EMOJIS[step] : allowed
120  }
121  // the number of lines is capped as well: 5000 malformed items must not write 5000 lines into the transcript
122  if (warnings.length > WARN_LINES_MAX) warnings.splice(WARN_LINES_MAX, warnings.length - WARN_LINES_MAX, `... and ${warnings.length - WARN_LINES_MAX} more`)
123  return { emojis, warnings }
124}
125
126// A module constant on purpose: a base taken from the environment could carry
127// the bot token to another host. Tests answer $.http.fetch beneath the mod.
128const API_BASE = 'https://api.telegram.org'
129const FETCH_MS = 8_000
130const FETCH_MIN_MS = 100
131const RETRY_MAX_TRIES = 20
132const RETRY_MAX_AGE_MS = 48 * 3600 * 1000
133const RETRY_PER_RUN = 5
134// the closing pass of a turn end writes every uncertain message of the turn (up to this bound), not 5
135const RETRY_CLOSE_MAX = 50
136// records are kept past the turn end -- a message the agent took over must not be taken back by a later
137// re-delivery, and a recently delivered message stays known -- bounded: at most this many, none untouched for longer
138// than RETRY_MAX_AGE_MS (the oldest go first)
139const KEEP_MAX = 100
140
141// A channel delivery STARTS with a Telegram channel tag; a batched delivery then
142// carries one tag per message (the placeholder hook records all of them, too).
143// Text that does not start with a tag -- e.g. a pasted or quoted tag -- does not
144// count: reacting to it would mark an old, already answered message.
145const LEADING_TAG = /^\s*<channel\b[^>]*\bsource="[^"]*telegram[^"]*"[^>]*>/
146const ANY_TAG = /<channel\b[^>]*\bsource="[^"]*telegram[^"]*"[^>]*>/g
147
148type Target = { chat_id: string; message_id: number; stage: Step }
149type Retry = { chat_id: string; message_id: number; tries: number; first: number; emoji?: string | null; agent?: boolean }
150type Rec = { chat_id: string; message_id: number; desired: string | null; confirmed: string | null; known: boolean; inflight: number; agent: boolean; gen: number; since?: number }
151
152const key = (t: { chat_id: string; message_id: number }) => `${t.chat_id}:${t.message_id}`
153
154type Src = { chat_id: string; message_id: number; message_thread_id?: number }
155
156/** The (chat_id, message_id) of every message in a Telegram channel delivery -- plus its message_thread_id when it came
157 *  from a topic (TELEGRAM_TOPICS_ENABLED, the same attribute and pattern as the stock placeholder hook); [] for anything
158 *  else. */
159export function targetsOf(text: unknown): Src[] {
160  const t = typeof text === 'string' ? text : ''
161  if (!LEADING_TAG.test(t)) return []
162  const out: Src[] = []
163  for (const tag of t.match(ANY_TAG) ?? []) {
164    const chat = /\bchat_id="([^"]+)"/.exec(tag)
165    const mid = /\bmessage_id="(-?\d+)"/.exec(tag)
166    const tid = /\bmessage_thread_id="(\d+)"/.exec(tag)
167    if (chat && mid) out.push({ chat_id: chat[1]!, message_id: Number(mid[1]), ...(tid ? { message_thread_id: Number(tid[1]) } : {}) })
168  }
169  return out
170}
171
172/** The Telegram channel's reply tool, e.g. mcp__plugin_telegram_telegram__reply. */
173export function isReplyTool(tool: unknown): boolean {
174  return /telegram.*__reply$/.test(String(tool ?? ''))
175}
176
177/** The Telegram channel's own react tool: the agent setting a reaction itself. */
178export function isReactTool(tool: unknown): boolean {
179  return /telegram.*__react$/.test(String(tool ?? ''))
180}
181
182/** The chat a reply tool call targets: its input's chat_id (flat as a fallback). */
183export function replyChat(e: any): string | null {
184  for (const v of [e?.input?.chat_id, e?.chat_id]) {
185    if (v !== undefined && v !== null && v !== '') return String(v)
186  }
187  return null
188}
189
190/** A tool result that clearly failed (deny, isError, or an error-shaped result). */
191export function toolSucceeded(r: any): boolean {
192  if (!r || r.deny !== undefined || r.isError) return false
193  const x = r.result
194  if (x && typeof x === 'object' && !Array.isArray(x) && (x.is_error || x.isError || x.ok === false)) return false
195  return true
196}
197
198/** The per-call timeout: MARVEEN_TELEGRAM_REACTION_FETCH_MS clamped to 100..8000 ms, else 8000. */
199export function fetchMs(value: unknown): number {
200  const n = Number(value)
201  return Number.isFinite(n) && n > 0 ? Math.min(FETCH_MS, Math.max(FETCH_MIN_MS, n)) : FETCH_MS
202}
203
204export function tokenFrom(envText: unknown): string | null {
205  for (const line of String(envText ?? '').split('\n')) {
206    const s = line.trim()
207    if (s.startsWith('TELEGRAM_BOT_TOKEN=')) return s.slice('TELEGRAM_BOT_TOKEN='.length).trim() || null
208  }
209  return null
210}
211
212function withTimeout<T>(p: Promise<T>, ms: number): Promise<T | null> {
213  let timer: ReturnType<typeof setTimeout> | undefined
214  const late = new Promise<null>(resolve => { timer = setTimeout(() => resolve(null), ms) })
215  return Promise.race([p, late]).finally(() => clearTimeout(timer))
216}
217
218// Module state starts over on every load; register() resets it explicitly too.
219let enabled = false
220let token: string | null = null
221let timeoutMs = FETCH_MS
222let emojis: Record<Step, string> = { ...DEFAULT_EMOJIS }
223let keepAgent = false
224// the messages the agent reacted on (successfully) during the current main turn: cleared at its end unless keepAgent
225let turnAgent = new Set<string>()
226// The store keys carry the bot's state dir: two agents sharing a config dir
227// must not clean up each other's reactions with the wrong bot.
228let storeKey = (name: string) => name
229let pending = new Map<string, Target>()
230let retries: Retry[] = []
231// ONE WRITER, DESIRED STATE. Every message we ever reacted on has a record: the
232// reaction it SHOULD carry (our stage, the agent's own emoji once it took the message over, or none), and what Telegram
233// is KNOWN to carry. Every setMessageReaction goes on the one serial chain and sends the desired value as it is when the
234// call runs -- never a snapshot taken earlier. A call with no certain answer (timed out, failed) leaves the message
235// unknown; a timed-out call that settles later (it may have landed after newer calls) syncs the message again, and it
236// is retried like a failed clear. The invariant: once the chain is drained and every request settled, the reaction on
237// each message is the desired one.
238let records = new Map<string, Rec>()
239let chain: Promise<void> = Promise.resolve()
240
241type Outcome = 'ok' | 'failed' | 'timeout'
242
243/** Key names of a result object for the diagnostic line: only schema-like names (a key that looks like data is shown as
244 *  '?'), at most 8 -- never a value. */
245export function keyNames(v: unknown): string {
246  if (!v || typeof v !== 'object') return typeof v
247  const ks = Object.keys(v as object)
248  const shown = ks.slice(0, 8).map(k => (/^[A-Za-z_][A-Za-z0-9_]{0,31}$/.test(k) ? k : '?'))
249  return shown.join(',') + (ks.length > 8 ? ',…' : '')
250}
251
252/** One setMessageReaction call, settled: the fetch is never cut short (no abort exists), only no longer waited for. */
253function call($: EngineInterface, chat_id: string, message_id: number, emoji: string | null): Promise<'ok' | 'failed'> {
254  if (!token) return Promise.resolve('failed')
255  return Promise.resolve()
256    .then(() => $.http.fetch(`${API_BASE}/bot${token}/setMessageReaction`, {
257      method: 'POST',
258      headers: { 'Content-Type': 'application/json' },
259      body: JSON.stringify({ chat_id, message_id, reaction: emoji ? [{ type: 'emoji', emoji }] : [] }),
260    }))
261    .then((r: any) => (r && r.status === 200 ? 'ok' : 'failed'), () => 'failed' as const)
262}
263
264/** Waits at most one fetch timeout; a request still out is counted (`out`) BEFORE its settling is hooked (`late`). */
265async function send(p: Promise<'ok' | 'failed'>, out: () => void, late: () => void): Promise<Outcome> {
266  const r = await withTimeout(p, timeoutMs)
267  if (r !== null) return r
268  out()
269  p.then(late, late)
270  return 'timeout'
271}
272
273function persist($: EngineInterface): Promise<void> {
274  // a retry carries the reaction its message must end with: the agent's own emoji where the agent took it over, else none
275  // (a stage of a dead turn is never restored after a crash)
276  // The agent's emoji is carried only with KEEP_AGENT -- by default a dead turn's agent emoji is not
277  // restored either (it would never be cleared); the ownership itself travels apart ('agent'), so a re-delivery is no stage
278  const out = retries.map(r => {
279    const rec = records.get(key(r))
280    return rec ? { ...r, emoji: rec.agent && keepAgent ? rec.desired : null, agent: rec.agent } : r
281  })
282  return Promise.resolve()
283    .then(() => $.store.set(storeKey('pending'), [...pending.values()]))
284    .then(() => $.store.set(storeKey('retries'), out))
285    .catch(() => {})
286}
287
288/** Queue work on the one chain; the hook that called it does not wait. */
289function enqueue($: EngineInterface, work: () => Promise<void>): void {
290  chain = chain.then(work).then(() => persist($)).catch(() => {})
291}
292
293const settled = (r: Rec) => r.known && r.confirmed === r.desired && r.inflight === 0
294
295// While a retry pass runs (it awaits writes), a retryLater from a hook must not be lost: the pass
296// replaces the queue with what IT keeps at its end, and an X it already processed and dropped is still in the old array
297// (so the dedup would skip it). Every call during a pass is ALSO collected here and merged into the pass's result.
298let incoming: Retry[] | null = null
299
300function retryLater(r: { chat_id: string; message_id: number }, emoji: string | null = null): void {
301  const e: Retry = { chat_id: r.chat_id, message_id: r.message_id, tries: 0, first: Date.now(), emoji }
302  if (incoming && !incoming.some(x => key(x) === key(r))) incoming.push(e)
303  if (!retries.some(x => key(x) === key(r))) retries.push(e)
304}
305
306/** The message's state is no longer known (a late answer, a failed or uncertain call): a NEW generation -- a write still
307 *  on the wire may not confirm the old one. */
308function unknown(r: Rec): void {
309  r.gen += 1
310  r.known = false
311}
312
313/** The one write: Telegram gets the record's desired reaction AS IT IS NOW. Runs on the chain only. */
314async function write($: EngineInterface, k: string, force = false): Promise<Outcome | 'skip'> {
315  const r = records.get(k)
316  if (!r || (!force && r.known && r.confirmed === r.desired)) return 'skip'
317  const want = r.desired
318  const gen = r.gen
319  const out = await send(call($, r.chat_id, r.message_id, want), () => { r.inflight += 1 }, () => {
320    // the timed-out request is back (landed or not, maybe after newer calls): the message is unknown again -> sync
321    r.inflight -= 1
322    unknown(r)
323    if (records.get(k) === r) sync($, k)
324    else retryLater(r, r.agent ? r.desired : null)   // the module was reloaded meanwhile: the new one repeats it
325  })
326  if (out === 'ok' && r.gen === gen) {
327    r.confirmed = want
328    r.known = true
329  } else if (out !== 'ok') {
330    unknown(r)
331    retryLater(r)   // repeated in the closing pass of the turn end, and later, like a failed clear
332  }
333  return out
334}
335
336/** Bring one message to its desired reaction, on the chain. */
337function sync($: EngineInterface, k: string): void {
338  enqueue($, async () => { await write($, k) })
339}
340
341function record(t: { chat_id: string; message_id: number }): Rec {
342  const k = key(t)
343  let r = records.get(k)
344  if (!r) {
345    r = { chat_id: t.chat_id, message_id: t.message_id, desired: null, confirmed: null, known: false, inflight: 0, agent: false, gen: 0 }
346    records.set(k, r)
347  }
348  return r
349}
350
351function stage($: EngineInterface, t: { chat_id: string; message_id: number }, step: Step): void {
352  if (step === 'after' && !emojis.after) return         // the step is off: nothing changes, not even the pending stage (0.3.2 to the call)
353  const k = key(t)
354  if (records.get(k)?.agent) return                     // the agent took the message over: no stages again (a re-delivery)
355  const cur = pending.get(k)
356  if (cur && RANK[cur.stage] >= RANK[step]) return      // forward only
357  pending.set(k, { chat_id: t.chat_id, message_id: t.message_id, stage: step })
358  const emoji = emojis[step]
359  if (!emoji) return   // a skipped step: no change here -- the previous reaction stays (at the first: none yet)
360  const r = record(t)
361  r.desired = emoji
362  r.since = Date.now()
363  sync($, k)
364}
365
366function clear($: EngineInterface, targets: Target[]): void {
367  for (const t of targets) {
368    pending.delete(key(t))
369    const r = records.get(key(t))
370    if (r && !r.agent) r.desired = null
371    sync($, key(t))
372  }
373}
374
375/** A queued message without a record (an earlier session's leftover) gets one -- its stored desired reaction
376 *  (the agent's emoji, or none) -- so it is written on the one chain like any other, and a later agent react hands it over. */
377function adopt(r: Retry): Rec {
378  const rec = record(r)
379  // An earlier session's agent emoji is restored only with KEEP_AGENT; by default it is cleared
380  rec.desired = keepAgent ? r.emoji ?? null : null
381  rec.agent = r.agent === true || r.emoji != null
382  if (rec.agent) rec.since = r.first
383  return rec
384}
385
386/** An agent emoji still shown that the main turn end must clear (KEEP_AGENT=0) -- its queue entry
387 *  stays (persisted with no emoji) even while the message is settled, so a session dying before that turn end clears it. */
388const owed = (rec: Rec) => !keepAgent && rec.agent && rec.desired !== null
389
390function retryClears($: EngineInterface, perRun = RETRY_PER_RUN): void {
391  enqueue($, async () => {
392    const now = Date.now()
393    const keep: Retry[] = []
394    let run = 0
395    incoming = []
396    let kesz = false
397    try {
398    for (const r of retries) {
399      const k = key(r)
400      const rec = records.get(k)
401      if (rec && settled(rec)) {                                                       // known right: done
402        if (owed(rec)) keep.push(r)                                                    // ...but still to clear at the turn end
403        continue
404      }
405      if (r.tries >= RETRY_MAX_TRIES || now - r.first > RETRY_MAX_AGE_MS) {            // give up -- its record too
406        if (rec && !pending.has(k)) records.delete(k)
407        continue
408      }
409      if (!rec && pending.has(k)) continue                                            // live again: not an orphan
410      if (run >= perRun) { keep.push(r); continue }
411      run += 1
412      const rr = rec ?? adopt(r)   // one write path for every message: bounded, generation-checked, late answers synced
413      const out = await write($, k, true)
414      if (out !== 'ok' || !settled(rr)) keep.push({ ...r, tries: r.tries + 1 })
415      else if (owed(rr)) keep.push(r)
416    }
417    kesz = true
418    } finally {
419      // the pass's result + every retryLater that arrived DURING it (one entry per message; the next pass decides). A
420      // throw inside the pass keeps the WHOLE queue, as before (the not yet visited entries must not be lost).
421      const inc = incoming ?? []
422      incoming = null
423      const alap = kesz ? keep : retries
424      retries = alap.concat(inc.filter(x => !alap.some(k => key(k) === key(x))))
425    }
426  })
427}
428
429/** Records stay past the turn end (the agent's take-over survives a later re-delivery, a recently delivered
430 *  message stays known); only a known-right, idle record goes, and only when untouched for RETRY_MAX_AGE_MS or beyond
431 *  the KEEP_MAX newest. */
432function prune($: EngineInterface): void {
433  enqueue($, async () => {
434    const now = Date.now()
435    const idle = (k: string, r: Rec) => settled(r) && !pending.has(k) && !retries.some(x => key(x) === k)
436    for (const [k, r] of [...records]) if (idle(k, r) && now - (r.since ?? now) > RETRY_MAX_AGE_MS) records.delete(k)
437    const sor = [...records].sort((a, b) => (a[1].since ?? 0) - (b[1].since ?? 0))
438    let tobb = sor.length - KEEP_MAX
439    for (const [k, r] of sor) {
440      if (tobb <= 0) break
441      if (idle(k, r)) { records.delete(k); tobb -= 1 }
442    }
443  })
444}
445
446async function stateDir($: EngineInterface): Promise<string> {
447  const dir = (await $.env.get('TELEGRAM_STATE_DIR')) || `${(await $.env.get('HOME')) ?? ''}/.claude/channels/telegram`
448  return dir.replace(/\/+$/, '')
449}
450
451// Plugin install path: the engine's stock placeholder hook (scripts/hooks/telegram_progress.py) does NOT know about
452// the reaction mod, and a plugin does not write the engine's tracked files. The hook does keep a documented dedup
453// contract, though: when an O_EXCL marker (<state>/progress/seen-<sid>-<src_mid>.marker) exists it exits with a
454// "dedup skip" and posts no placeholder. next(e) of prompt.submit runs the UserPromptSubmit settings hooks (the API's
455// contract), so if the mod writes the marker BEFORE that, the stock hook deterministically skips the placeholder. The
456// mod itself records the pending entry in the stock format (chat_id + src_mid, WITHOUT a placeholder message_id), so
457// the stock Stop enforcement (blocks once, then delivers as a fallback) and the watchdog keep working. A message from a
458// topic (TELEGRAM_TOPICS_ENABLED) also carries its message_thread_id, as the stock entry does, so the fallback reply
459// goes back into that topic; without a topic the entry is byte for byte the old one. On an error nothing blocks: the
460// stock hook posts its placeholder the old way, and the reaction goes out regardless.
461export async function claimForFactoryHook($: EngineInterface, targets: Src[]): Promise<void> {
462  if (!targets.length) return
463  const dir = await stateDir($)
464  const sid = await $.session.id()
465  const pdir = `${dir}/progress`
466  const pendPath = `${pdir}/${sid}.json`
467  let pend: any[] = []
468  try {
469    const v = JSON.parse(await $.fs.read(pendPath))
470    if (Array.isArray(v)) pend = v
471  } catch {}
472  for (const t of targets) {
473    if (!pend.some(p => p && String(p.chat_id) === t.chat_id && String(p.src_mid) === String(t.message_id))) {
474      pend.push({ chat_id: t.chat_id, src_mid: String(t.message_id),
475                  ...(t.message_thread_id !== undefined ? { message_thread_id: t.message_thread_id } : {}) })
476    }
477  }
478  // The pending entry FIRST, atomically (a tmp file renamed over it; $.fs.write creates the progress/ dir).
479  // Only then the dedup marker: if the pending write fails, no marker is written either, so the stock hook posts its
480  // placeholder and records its own pending entry (the Stop enforcement keeps working) instead of a skipped placeholder
481  // with no pending entry. The read-modify-write is not locked against the stock hooks (a short window; README).
482  await writePending($, pendPath, pend)
483  for (const t of targets) await $.fs.write(`${pdir}/seen-${sid}-${t.message_id}.marker`, '')
484  // Accumulated, not replaced: two deliveries close together may both wait for their classic event, and the transcript
485  // path is the SESSION's, so either event stamps both. Capped: a classic event that never comes must not grow it.
486  claimed = claimed.concat(targets.map(t => ({ chat_id: t.chat_id, src_mid: String(t.message_id) }))).slice(-50)
487}
488
489// The atomic pending write: a tmp file renamed over the pending file.
490async function writePending($: EngineInterface, pendPath: string, pend: any[]): Promise<void> {
491  const tmp = `${pendPath}.tmp-${Date.now()}-${Math.floor(Math.random() * 1e9)}`
492  await $.fs.write(tmp, JSON.stringify(pend))
493  const mv = await $.process.run(['/bin/mv', '-f', tmp, pendPath])
494  if (mv.exitCode !== 0) {
495    // No tmp file left behind in progress/ ($.fs has no delete)
496    try { await $.process.run(['/bin/rm', '-f', tmp]) } catch {}
497    throw new Error('pending rename failed')
498  }
499}
500
501// The pending entries THIS prompt's claim wrote (the stock submit hook skips on the marker, so it never stamps them).
502// prompt.submit's input has no transcript path; the classic UserPromptSubmit input does (BaseHookInput), and that event
503// runs inside next(e) of prompt.submit, ahead of the settings hooks. stampTranscript() puts the path on those entries.
504let claimed: { chat_id: string; src_mid: string }[] = []
505
506// The stock watchdog finds the agent's transcript ONLY through the transcript_path on a pending entry. Without it a
507// round the Stop hook never closes (agent died, reply tool hung) cannot recover the agent's answer.
508export async function stampTranscript($: EngineInterface, transcriptPath: unknown): Promise<void> {
509  const mine = claimed
510  claimed = []
511  if (!mine.length || typeof transcriptPath !== 'string' || !transcriptPath) return
512  const pendPath = `${await stateDir($)}/progress/${await $.session.id()}.json`
513  let pend: any
514  try { pend = JSON.parse(await $.fs.read(pendPath)) } catch { return }
515  if (!Array.isArray(pend)) return
516  let changed = false
517  for (const p of pend) {
518    if (p && !p.transcript_path && mine.some(m => String(p.chat_id) === m.chat_id && String(p.src_mid) === m.src_mid)) {
519      p.transcript_path = transcriptPath
520      changed = true
521    }
522  }
523  if (changed) await writePending($, pendPath, pend)
524}
525
526async function readToken($: EngineInterface, dir: string): Promise<string | null> {
527  try {
528    return tokenFrom(await $.fs.read(`${dir}/.env`))
529  } catch {
530    return null
531  }
532}
533
534export const register: Register = on => {
535  enabled = false
536  claimed = []
537  token = null
538  timeoutMs = FETCH_MS
539  emojis = { ...DEFAULT_EMOJIS }
540  keepAgent = false
541  turnAgent = new Set()
542  storeKey = (name: string) => name
543  pending = new Map()
544  retries = []
545  incoming = null   // a register() during a pass starts with no half-collected list
546  records = new Map()
547  chain = Promise.resolve()
548
549  on('session.start', async ($, e, next) => {
550    const r = await next(e)
551    let warnings: string[] = []
552    try {
553      // When installed as a plugin next to an unmodified engine, the mod is switched on for the main agent through the
554      // engine's .env CHANNEL_PLUGINS_EXTRA (enabledPlugins) -- there is no supported way to set the main agent's process
555      // env. Hence: LOADED = ON; MARVEEN_TELEGRAM_REACTION=0 explicitly turns it off ('1' and unset: on).
556      const on_ = (await $.env.get('MARVEEN_TELEGRAM_REACTION')) !== '0'
557      // fleet-forbid-live: in a test process tree (FLEET_FORBID_LIVE=1) the mod is INERT -- it makes no Bot API calls
558      const env = { FLEET_FORBID_LIVE: await $.env.get('FLEET_FORBID_LIVE') }
559      enabled = false
560      if (env.FLEET_FORBID_LIVE === '1') return r
561      // Reactions left behind by an earlier session are cleaned up even while the mod is switched OFF (a reaction must
562      // not stay orphaned just because the mod was turned off in the meantime); no new reactions while it is off
563      const dir = await stateDir($)
564      storeKey = (name: string) => `${name}@${dir}`
565      token = await readToken($, dir)
566      timeoutMs = fetchMs(await $.env.get('MARVEEN_TELEGRAM_REACTION_FETCH_MS'))
567      // The emoji of each step; a bad value is a loud warning (never silent), that step keeps its default
568      const cfg = emojisFrom(await $.env.get('MARVEEN_TELEGRAM_REACTION_EMOJIS'))
569      emojis = cfg.emojis
570      warnings = cfg.warnings.map(w => `MARVEEN_TELEGRAM_REACTION_EMOJIS ${w}`)
571      const keepRaw = await $.env.get('MARVEEN_TELEGRAM_REACTION_KEEP_AGENT')
572      keepAgent = keepRaw === '1'
573      // A value other than 0/1 ('true', 'yes', '1 ') is not silently the default -- named,
574      // without the value
575      if (keepRaw != null && keepRaw !== '' && keepRaw !== '0' && keepRaw !== '1') {
576        warnings.push('MARVEEN_TELEGRAM_REACTION_KEEP_AGENT is neither 0 nor 1 -- the default applies (the agent\'s emoji is cleared)')
577      }
578      // A previous session that ended without its turn end (crash, restart)
579      // may have left reactions behind: clear them now.
580      const left = await $.store.get(storeKey('pending'))
581      const old = await $.store.get(storeKey('retries'))
582      if (Array.isArray(old)) {
583        for (const x of old) if (x && typeof x.chat_id === 'string' && Number.isFinite(x.message_id)) retryLater(x, x.emoji ?? null)
584        retries = retries.map(r => ({ ...r, ...(old.find((x: any) => x && key(x) === key(r)) ?? {}) }))   // keep tries/first
585      }
586      if (Array.isArray(left)) {
587        // one entry per message: a message already queued keeps its own (desired) entry
588        for (const t of left) if (t && typeof t.chat_id === 'string' && Number.isFinite(t.message_id)) retryLater(t, null)
589      }
590      retryClears($)
591      // the first three steps skipped = off (`after` alone does not keep it on); the cleanup above still ran
592      enabled = on_ && LADDER_032.some(s => emojis[s] !== '')
593      // the warnings LAST and not awaited: an op that never returns must not hold back the cleanup or the switch
594      for (const w of warnings) void Promise.resolve().then(() => $.ui.log(`telegram-reaction: ${w}`)).catch(() => {})
595    } catch {
596      enabled = false
597    }
598    return r
599  })
600
601  // Inside next(e) of prompt.submit, ahead of the settings hooks: the input carries the session's transcript path
602  on('classic.UserPromptSubmit', async ($, e, next) => {
603    try { await stampTranscript($, (e as any).transcript_path) } catch {}
604    return next(e)
605  })
606
607  on('prompt.submit', async ($, e, next) => {
608    try {
609      const o: any = e.origin
610      if (enabled && o && o.kind === 'channel' && /telegram/i.test(String(o.server ?? ''))) {
611        const targets = targetsOf(e.text)
612        for (const t of targets) stage($, t, 'received')   // synchronous (the stages queued on the chain are still coalesced)
613        // BEFORE the stock placeholder hook (next(e) runs the settings hooks): dedup claim + pending
614        try { await claimForFactoryHook($, targets) } catch {}
615      }
616    } catch {}
617    return next(e)
618  })
619
620  on('turn.start', async ($, e, next) => {
621    try {
622      if (enabled) for (const t of [...pending.values()]) if (t.stage === 'received') stage($, t, 'working')
623    } catch {}
624    return next(e)
625  })
626
627  on('tool.call', async ($, e, next) => {
628    let replyTo: string | null = null
629    let handOver: string | null = null
630    let agentEmoji: string | null = null
631    try {
632      if (enabled && isReactTool(e.tool)) {
633        // The agent sets a reaction on a message itself: that message is its
634        // own from now on -- no more stages, and no clear at the turn end.
635        // Only once the react call SUCCEEDED (below): a failed one (e.g. an
636        // emoji Telegram does not allow) leaves our reaction in place, and the
637        // turn end must still clear it. Nothing waits here.
638        const input: any = e.input ?? {}
639        if (input.chat_id !== undefined && Number.isFinite(Number(input.message_id))) {
640          handOver = key({ chat_id: String(input.chat_id), message_id: Number(input.message_id) })
641          agentEmoji = typeof input.emoji === 'string' && input.emoji ? input.emoji : null
642        }
643      } else if (enabled && pending.size) {
644        if (isReplyTool(e.tool)) {
645          replyTo = replyChat(e)
646          for (const t of [...pending.values()]) if (replyTo !== null && t.chat_id === replyTo) stage($, t, 'writing')
647        } else {
648          for (const t of [...pending.values()]) if (t.stage === 'received') stage($, t, 'working')
649        }
650      }
651    } catch {}
652    // The reply is not the end of the turn: the reaction stays (✍️, after a successful reply 👨‍💻) until
653    // turn.complete clears it. A session that dies after a reply but before the
654    // turn end leaves it in the store; the next session start clears it.
655    const r = await next(e)
656    try {
657      // the reply SUCCEEDED -- the answer is out while the turn still runs: the reply's own chat moves to `after`
658      // (forward only; a failed reply stays at ✍️; an agent-owned message is left alone by stage())
659      if (replyTo !== null && toolSucceeded(r)) {
660        for (const t of [...pending.values()]) if (t.chat_id === replyTo) stage($, t, 'after')
661      }
662      const rec = handOver !== null ? records.get(handOver) : undefined
663      if (!rec && handOver !== null && toolSucceeded(r)) {
664        // A message the mod does not (or no longer) track: the agent's reaction is on it now (its react
665        // succeeded) -- recorded as the agent's, KNOWN (no write needed), so a later re-delivery does not stage it
666        const input: any = e.input ?? {}
667        const nr = record({ chat_id: String(input.chat_id), message_id: Number(input.message_id) })
668        nr.agent = true
669        nr.desired = agentEmoji
670        nr.confirmed = agentEmoji
671        nr.known = true
672        nr.since = Date.now()
673        // With skipped steps a message can be pending WITHOUT a record -- it is the agent's now: out of
674        // pending, and the store updated (sync -> persist; the write itself is skipped, the record is already known)
675        if (pending.delete(handOver) || !keepAgent) sync($, handOver)
676        turnAgent.add(handOver)
677        // By default the agent's emoji must be cleared even if the session dies before the turn end --
678        // queued (persisted with no emoji), so the next session start clears it; the closing pass drops it once settled
679        if (!keepAgent) retryLater(nr)
680      } else if (rec && toolSucceeded(r)) {
681        // The agent's reaction becomes the DESIRED one, and our chain writes it again AFTER everything it
682        // already holds -- a set of ours still running (or timed out, landing late) can never be the last word
683        pending.delete(handOver!)
684        rec.agent = true
685        rec.since = Date.now()
686        rec.desired = agentEmoji
687        if (!keepAgent) retryLater(rec)   // a crash before the turn end still clears it (see above)
688        // (KEEP_AGENT=0 property) the agent's react CHANGED what Telegram shows -- our confirmed value
689        // is stale, so the message is unknown (a new generation): a later desired '' must not be skipped as "already right"
690        unknown(rec)
691        sync($, handOver!)
692        turnAgent.add(handOver!)
693      } else if (rec) {
694        // a FAILED react may still have landed: the message is uncertain -> the closing pass writes
695        // its desired reaction again (the turn-end clear of ours, or the agent's earlier emoji)
696        unknown(rec)
697        retryLater(rec)
698      } else if (handOver !== null && !keepAgent) {
699        // A react judged FAILED on a message the mod never tracked may still have LANDED (a
700        // misjudged success included) -- by default no emoji stays at the turn end, without exception: the message
701        // becomes the agent's with no desired reaction, UNKNOWN (so the clear is written), queued (persisted, a crash
702        // still clears it) and cleared at the turn end. The bot only ever writes its own reaction: a clear is harmless.
703        const input: any = e.input ?? {}
704        const nr = record({ chat_id: String(input.chat_id), message_id: Number(input.message_id) })
705        // a message still RUNNING (pending, no record only because its steps so far were skipped)
706        // stays OURS: its ladder goes on, and the turn-end clear of pending writes the clear (unknown: not skipped)
707        if (!pending.has(handOver)) {
708          nr.agent = true
709          turnAgent.add(handOver)
710        }
711        nr.desired = null
712        nr.since = Date.now()
713        unknown(nr)
714        retryLater(nr)
715        enqueue($, async () => {})   // persist only (the queue entry); the clear itself is written at the turn end
716      }
717      if (handOver !== null && !toolSucceeded(r)) {
718        // Diagnostic: the agent's react was judged FAILED -- the message is not handed over, so our turn-end
719        // clear also removes what the react may have set. The shape only (key names), never the content.
720        const shape = keyNames(r)
721        const inner = r && typeof r === 'object' && (r as any).result && typeof (r as any).result === 'object'
722          ? keyNames((r as any).result) : typeof (r as any)?.result
723        const what = !rec ? (keepAgent ? 'untracked -- the mod leaves it alone (KEEP_AGENT=1)' : pending.has(handOver) ? 'ours, still running (no record yet) -- our turn-end clear also removes what the react may have set' : 'untracked -- cleared at the turn end')
724          : rec.agent ? (keepAgent ? "already the agent's -- its earlier emoji is written again" : "already the agent's -- cleared at the turn end")
725          : 'ours -- our turn-end clear also removes what the react may have set'
726        void Promise.resolve().then(() => $.ui.log(`telegram-reaction: the agent's react was judged FAILED (result keys: ${shape}; result.result: ${inner}); the message is ${what}`)).catch(() => {})
727      }
728    } catch {}
729    return r
730  })
731
732  on('turn.complete', async ($, e, next) => {
733    const r = await next(e)
734    try {
735      // A subagent's turn ending (agentId set) does not end the main loop's turn.
736      if (enabled && !e.agentId) {
737        // The CLOSING write: the clears first, then the retry pass -- queued after EVERYTHING this turn queued
738        // (the clears included), it writes the desired reaction of every message left uncertain (a timed-out or failed
739        // call, a failed clear of this turn, a hand-over); what is still uncertain stays queued for the next turn end
740        // and session start (at most 20 tries / 48 h, 5 messages per pass).
741        clear($, [...pending.values()])
742        // The agent's own reactions of this turn are cleared too, unless KEEP_AGENT (it stays the agent's message)
743        if (!keepAgent) {
744          for (const k of turnAgent) {
745            const rec = records.get(k)
746            if (rec && rec.agent) {
747              rec.desired = null
748              sync($, k)
749            }
750          }
751        }
752        turnAgent = new Set()
753        retryClears($, RETRY_CLOSE_MAX)
754        prune($)
755      }
756    } catch {}
757    return r
758  })
759}
760