the bot answers its own messages in an endless loop

"""Find Slack channels where the app is replying to its own messages.

Read only. Three GET methods and no writes: a bot token with channels:read and
channels:history is enough. The repair is printed, never performed, because this
token can post into the same channels it is reading.
"""
import argparse
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("slack_echo_loop_audit")

API = "https://slack.com/api"


def is_self(message, identity):
    """True when this message was authored by the app we authenticated as.

    Pure, and deliberately narrow. Matching on "has a bot_id" would flag every
    other integration in the channel and report a busy alerts channel as a loop,
    so the comparison is against our own ids from auth.test. Both are checked
    because a modern app-authored message carries bot_id while a message posted
    with a user token carries only `user`.

    This is the same predicate the repair puts in the event handler.
    """
    bot_id = identity.get("bot_id")
    user_id = identity.get("user_id")
    if bot_id and message.get("bot_id") == bot_id:
        return True
    if user_id and message.get("user") == user_id:
        return True
    return False


def verdict(messages, identity, *, min_run=4, burst=2.0):
    """Classify one channel by its longest run of self-authored messages.

    Pure, so the thresholds are visible and testable rather than buried in a
    request loop. `messages` are history items in any order; they are sorted by
    ts here because a run is a property of posting order.

    Returns (state, detail). Length alone is not the signal: a digest job posting
    a dozen messages in a row is not a loop, so a long run whose internal gaps
    are wider than `burst` seconds gets its own state rather than being reported
    as one.
    """
    ordered = sorted(messages, key=lambda m: float(m.get("ts") or 0))

    best, best_gaps = [], []
    run, gaps = [], []
    for m in ordered:
        if is_self(m, identity):
            if run:
                gaps.append(float(m.get("ts") or 0) - float(run[-1].get("ts") or 0))
            run.append(m)
        else:
            if len(run) > len(best):
                best, best_gaps = run, gaps
            run, gaps = [], []
    if len(run) > len(best):
        best, best_gaps = run, gaps

    n = len(best)
    if n <= 1:
        return ("quiet",
                "longest self-authored run is %d. Every reply is answering "
                "somebody else." % n)

    widest = max(best_gaps) if best_gaps else 0.0

    if n < min_run:
        return ("short-run",
                "%d in a row, %.1fs apart at widest. A threaded reply or a "
                "two-part message, not a loop." % (n, widest))

    if widest >= burst:
        return ("batch",
                "%d in a row but %.1fs apart at widest. That is a poster, not a "
                "loop: a digest or a backlog being drained. Worth confirming it "
                "is deliberate." % (n, widest))

    return ("echo-loop",
            "%d consecutive self-authored messages, none more than %.2fs apart, "
            "with no human message in the run. The handler is hearing itself."
            % (n, widest))


def call(session, method, **params):
    """One Web API read. Slack answers almost every failure with HTTP 200 and
    puts the error in the body, so the body is what gets asserted on."""
    r = session.get("%s/%s" % (API, method), params=params, timeout=30)
    r.raise_for_status()
    body = r.json()
    if not body.get("ok"):
        raise SystemExit("%s: %s (needed=%s provided=%s)"
                         % (method, body.get("error"), body.get("needed"),
                            body.get("provided")))
    return body


def channels(session, explicit):
    if explicit:
        return [{"id": c, "name": c} for c in explicit]
    out, cursor = [], ""
    while True:
        body = call(session, "users.conversations", limit=200,
                    types="public_channel,private_channel", cursor=cursor)
        out.extend(body.get("channels", []))
        cursor = (body.get("response_metadata") or {}).get("next_cursor") or ""
        if not cursor:
            return out


def history(session, channel_id, limit):
    body = call(session, "conversations.history", channel=channel_id,
                limit=min(200, limit))
    return body.get("messages", [])


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--channel", action="append", default=[],
                    help="channel id to read; repeatable. Default: every channel "
                         "the bot is a member of")
    ap.add_argument("--limit", type=int, default=200,
                    help="messages to read per channel")
    ap.add_argument("--min-run", type=int, default=4,
                    help="runs shorter than this are never reported as a loop")
    ap.add_argument("--burst", type=float, default=2.0,
                    help="seconds; a run spaced wider than this is a batch")
    args = ap.parse_args()

    token = os.environ.get("SLACK_BOT_TOKEN")
    if not token:
        log.error("set SLACK_BOT_TOKEN (a bot token with channels:read and "
                  "channels:history is enough)")
        return 2

    session = requests.Session()
    session.headers.update({"Authorization": "Bearer " + token})

    me = call(session, "auth.test")
    identity = {"bot_id": me.get("bot_id"), "user_id": me.get("user_id")}
    log.info("authenticated as %s (bot_id=%s) in %s",
             me.get("user"), identity["bot_id"], me.get("team"))

    targets = channels(session, args.channel)
    if not targets:
        log.info("the bot is not a member of any conversation")
        return 0

    loops = longest = 0
    for ch in targets:
        messages = history(session, ch["id"], args.limit)
        state, detail = verdict(messages, identity,
                                min_run=args.min_run, burst=args.burst)
        name = ch.get("name", ch["id"])
        if state in ("quiet", "short-run"):
            log.info("%-10s #%s  %s", state, name, detail)
            continue
        if state == "batch":
            log.info("%-10s #%s  %s", state, name, detail)
            continue
        loops += 1
        log.warning("%-10s #%s  %s", state, name, detail)
        log.warning("  repair: in the handler, return early when event.bot_id is "
                    "set, when event.subtype is bot_message, or when event.user "
                    "== %s.", identity["user_id"])
        log.warning("  better: subscribe to app_mention instead of "
                    "message.channels so your own posts never reach the handler.")

    log.info("%d channel(s) checked, %d loop(s)", len(targets), loops)
    return 1 if loops else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Find Slack channels where the app is replying to its own messages.
 *
 * Read only. Three GET methods and no writes: a bot token with channels:read
 * and channels:history is enough. The repair is printed, never performed.
 */
const API = 'https://slack.com/api';

/**
 * True when this message was authored by the app we authenticated as.
 *
 * Pure, and deliberately narrow: matching on "has a bot_id" would flag every
 * other integration in the channel. Both ids are checked because a modern
 * app-authored message carries bot_id while one posted with a user token
 * carries only `user`. This is the same predicate the repair puts in the
 * event handler.
 */
export function isSelf(message, identity) {
  if (identity.bot_id && message.bot_id === identity.bot_id) return true;
  if (identity.user_id && message.user === identity.user_id) return true;
  return false;
}

/**
 * Classify one channel by its longest run of self-authored messages.
 *
 * Pure, so the thresholds are visible and testable. Length alone is not the
 * signal: a digest job posting a dozen messages in a row is not a loop, so a
 * long run with wide internal gaps gets its own state.
 */
export function verdict(messages, identity, { minRun = 4, burst = 2.0 } = {}) {
  const ordered = [...messages].sort((a, b) => Number(a.ts ?? 0) - Number(b.ts ?? 0));

  let best = [];
  let bestGaps = [];
  let run = [];
  let gaps = [];
  for (const m of ordered) {
    if (isSelf(m, identity)) {
      if (run.length) gaps.push(Number(m.ts ?? 0) - Number(run[run.length - 1].ts ?? 0));
      run.push(m);
    } else {
      if (run.length > best.length) { best = run; bestGaps = gaps; }
      run = [];
      gaps = [];
    }
  }
  if (run.length > best.length) { best = run; bestGaps = gaps; }

  const n = best.length;
  if (n <= 1) {
    return ['quiet',
      `longest self-authored run is ${n}. Every reply is answering somebody else.`];
  }

  const widest = bestGaps.length ? Math.max(...bestGaps) : 0;

  if (n < minRun) {
    return ['short-run',
      `${n} in a row, ${widest.toFixed(1)}s apart at widest. A threaded reply ` +
      'or a two-part message, not a loop.'];
  }

  if (widest >= burst) {
    return ['batch',
      `${n} in a row but ${widest.toFixed(1)}s apart at widest. That is a ` +
      'poster, not a loop: a digest or a backlog being drained. Worth ' +
      'confirming it is deliberate.'];
  }

  return ['echo-loop',
    `${n} consecutive self-authored messages, none more than ` +
    `${widest.toFixed(2)}s apart, with no human message in the run. The ` +
    'handler is hearing itself.'];
}

async function call(token, method, params = {}) {
  const url = new URL(`${API}/${method}`);
  for (const [k, v] of Object.entries(params)) {
    if (v !== undefined && v !== null && v !== '') url.searchParams.set(k, v);
  }
  const res = await fetch(url, { headers: { Authorization: `Bearer ${token}` } });
  if (!res.ok) throw new Error(`${res.status} from ${method}`);
  const body = await res.json();
  // Slack answers almost every failure with HTTP 200 and puts the error in the
  // body, so the body is what gets asserted on.
  if (!body.ok) {
    throw new Error(`${method}: ${body.error} (needed=${body.needed} ` +
                    `provided=${body.provided})`);
  }
  return body;
}

async function channels(token, explicit) {
  if (explicit.length) return explicit.map((id) => ({ id, name: id }));
  const out = [];
  let cursor = '';
  for (;;) {
    const body = await call(token, 'users.conversations',
      { limit: 200, types: 'public_channel,private_channel', cursor });
    out.push(...(body.channels ?? []));
    cursor = body.response_metadata?.next_cursor ?? '';
    if (!cursor) return out;
  }
}

async function main() {
  const token = process.env.SLACK_BOT_TOKEN;
  if (!token) {
    console.error('set SLACK_BOT_TOKEN (channels:read and channels:history)');
    process.exitCode = 2;
    return;
  }

  const explicit = process.argv.slice(2).filter((a) => !a.startsWith('-'));

  const me = await call(token, 'auth.test');
  const identity = { bot_id: me.bot_id, user_id: me.user_id };
  console.log(`authenticated as ${me.user} (bot_id=${me.bot_id}) in ${me.team}`);

  const targets = await channels(token, explicit);
  if (targets.length === 0) {
    console.log('the bot is not a member of any conversation');
    return;
  }

  let loops = 0;
  for (const ch of targets) {
    const body = await call(token, 'conversations.history',
      { channel: ch.id, limit: 200 });
    const [state, detail] = verdict(body.messages ?? [], identity);
    const name = ch.name ?? ch.id;
    if (state !== 'echo-loop') {
      console.log(`${state.padEnd(10)} #${name}  ${detail}`);
      continue;
    }
    loops += 1;
    console.warn(`${state.padEnd(10)} #${name}  ${detail}`);
    console.warn('  repair: in the handler, return early when event.bot_id is ' +
                 `set, when event.subtype is bot_message, or when event.user == ${identity.user_id}.`);
    console.warn('  better: subscribe to app_mention instead of ' +
                 'message.channels so your own posts never reach the handler.');
  }

  console.log(`${targets.length} channel(s) checked, ${loops} loop(s)`);
  process.exitCode = loops ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and
// without the guard main() would run there too, fail on the missing token, and
// set a non-zero exit code that fails the whole test file even as every test
// passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from slack_echo_loop_audit import is_self, verdict

ME = {"bot_id": "B111", "user_id": "U111"}


def msg(ts, *, bot=None, user=None):
    m = {"ts": str(ts)}
    if bot:
        m["bot_id"] = bot
    if user:
        m["user"] = user
    return m


def test_another_apps_bot_message_is_not_ours():
    assert is_self(msg(1, bot="B999"), ME) is False


def test_our_bot_id_and_our_user_id_both_count():
    assert is_self(msg(1, bot="B111"), ME) is True
    assert is_self(msg(2, user="U111"), ME) is True


def test_replies_interleaved_with_humans_are_quiet():
    messages = [msg(1, user="U777"), msg(2, bot="B111"),
                msg(3, user="U777"), msg(4, bot="B111")]
    state, _ = verdict(messages, ME)
    assert state == "quiet"


def test_a_fast_unbroken_run_is_the_loop():
    messages = [msg(1000 + i * 0.3, bot="B111") for i in range(12)]
    state, detail = verdict(messages, ME)
    assert state == "echo-loop"
    assert "12" in detail


def test_a_slow_long_run_is_a_batch_not_a_loop():
    # A digest posting one message every five seconds. Reporting this is how
    # the check gets switched off.
    messages = [msg(1000 + i * 5.0, bot="B111") for i in range(12)]
    state, _ = verdict(messages, ME)
    assert state == "batch"


def test_history_arriving_newest_first_is_still_measured_correctly():
    newest_first = [msg(1000 + i * 0.3, bot="B111") for i in range(9)][::-1]
    assert verdict(newest_first, ME)[0] == "echo-loop"


def test_two_in_a_row_is_a_short_run():
    messages = [msg(1, user="U777"), msg(2, bot="B111"), msg(2.4, bot="B111")]
    state, _ = verdict(messages, ME)
    assert state == "short-run"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { isSelf, verdict } from './slack-echo-loop-audit.mjs';

const ME = { bot_id: 'B111', user_id: 'U111' };

const msg = (ts, { bot, user } = {}) => {
  const m = { ts: String(ts) };
  if (bot) m.bot_id = bot;
  if (user) m.user = user;
  return m;
};

test('another app bot message is not ours', () => {
  assert.equal(isSelf(msg(1, { bot: 'B999' }), ME), false);
});

test('our bot id and our user id both count', () => {
  assert.equal(isSelf(msg(1, { bot: 'B111' }), ME), true);
  assert.equal(isSelf(msg(2, { user: 'U111' }), ME), true);
});

test('replies interleaved with humans are quiet', () => {
  const messages = [msg(1, { user: 'U777' }), msg(2, { bot: 'B111' }),
    msg(3, { user: 'U777' }), msg(4, { bot: 'B111' })];
  assert.equal(verdict(messages, ME)[0], 'quiet');
});

test('a fast unbroken run is the loop', () => {
  const messages = Array.from({ length: 12 }, (_, i) => msg(1000 + i * 0.3, { bot: 'B111' }));
  const [state, detail] = verdict(messages, ME);
  assert.equal(state, 'echo-loop');
  assert.match(detail, /12/);
});

test('a slow long run is a batch, not a loop', () => {
  const messages = Array.from({ length: 12 }, (_, i) => msg(1000 + i * 5, { bot: 'B111' }));
  assert.equal(verdict(messages, ME)[0], 'batch');
});

test('history arriving newest first is still measured correctly', () => {
  const messages = Array.from({ length: 9 }, (_, i) => msg(1000 + i * 0.3, { bot: 'B111' })).reverse();
  assert.equal(verdict(messages, ME)[0], 'echo-loop');
});

test('two in a row is a short run', () => {
  const messages = [msg(1, { user: 'U777' }), msg(2, { bot: 'B111' }), msg(2.4, { bot: 'B111' })];
  assert.equal(verdict(messages, ME)[0], 'short-run');
});