a rising share of outbound calls end in status failed

"""Report the outbound call failure rate, bucketed by direction and destination.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The repair is printed, never performed,
because this script holds a credential to an account that can place calls and
spend money.
"""
import argparse
import datetime as dt
import logging
import os
import sys

import requests

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

HOST = "https://api.twilio.com"
BASE = HOST + "/2010-04-01"
MONITOR = "https://monitor.twilio.com/v1"

# Statuses that are an outcome. queued, ringing and in-progress are calls that
# have not finished yet: counting them would move the rate purely by when the
# script happened to run.
OUTCOMES = ("completed", "failed", "busy", "no-answer", "canceled")


def dial_prefix(to, digits=3):
    """Bucket a destination by its leading digits.

    SIP URIs and client identities get their own buckets rather than being
    stripped down to whatever digits they happen to contain, because a failure
    rate on sip: destinations is a completely different investigation from one
    on a country prefix.
    """
    v = str(to or "").strip()
    if not v:
        return "unknown"
    low = v.lower()
    if low.startswith("sip:") or low.startswith("sips:"):
        return "sip"
    if low.startswith("client:"):
        return "client"
    d = "".join(c for c in v if c.isdigit())
    if not d:
        return "unknown"
    return "+" + d[:digits]


def summarise(calls, digits=3):
    """Group outbound calls into (direction, prefix) buckets of outcomes.

    Pure, and deliberately tolerant: an unexpected status is skipped rather than
    counted as a failure, because a status this script does not know about is
    not evidence of anything.
    """
    buckets = {}
    for c in calls:
        status = str(c.get("status") or "").strip().lower()
        if status not in OUTCOMES:
            continue
        direction = str(c.get("direction") or "unknown").strip().lower()
        if not direction.startswith("outbound"):
            continue
        key = (direction, dial_prefix(c.get("to"), digits))
        b = buckets.setdefault(key, {"total": 0, "completed": 0, "failed": 0,
                                     "busy": 0, "no_answer": 0, "canceled": 0})
        b["total"] += 1
        b[status.replace("-", "_")] += 1
    return buckets


def verdict(bucket, floor=20, threshold=0.10):
    """Judge one bucket. Pure, and the thresholds are arguments so the boundary
    cases can be tested rather than argued about.

    Returns (state, detail).
    """
    total = bucket.get("total", 0)
    failed = bucket.get("failed", 0)
    share = (failed / total) if total else 0.0
    pct = "%.1f%%" % (share * 100)

    if total < floor:
        return ("low-volume",
                "%d call(s) is too few to read a rate from: %d failed, which is "
                "%s of nothing much." % (total, failed, pct))
    if failed == total:
        return ("total-failure",
                "every one of %d call(s) failed. This is not a rate, it is a "
                "destination or a permission that is off." % total)
    if share >= threshold:
        return ("elevated",
                "%d of %d call(s) failed (%s), against a threshold of %.0f%%. "
                "busy=%d no-answer=%d."
                % (failed, total, pct, threshold * 100,
                   bucket.get("busy", 0), bucket.get("no_answer", 0)))
    return ("ok", "%d of %d call(s) failed (%s)" % (failed, total, pct))


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def list_calls(session, account, since, limit, status=None):
    """Page the Calls list. next_page_uri here is a path, not an absolute URL.

    There is no ErrorCode filter on this resource, and StartTime>= is the only
    way to bound the window, so everything else is done client-side.
    """
    url = "%s/Accounts/%s/Calls.json" % (BASE, account)
    params = {"StartTime>=": since, "PageSize": 1000}
    if status:
        params["Status"] = status
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("calls", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def alert_codes(session, since, limit, levels):
    """Error codes seen in the window, counted, across both log levels.

    Sweeping error alone is the mistake worth avoiding here: some of the codes
    that explain a voice failure rate, including several 132xx Dial attribute
    errors, are logged at warning.
    """
    seen = {}
    for level in levels:
        url = MONITOR + "/Alerts"
        params = {"LogLevel": level, "StartDate": since, "PageSize": 1000}
        got = 0
        while url and got < limit:
            page = get(session, url, **params)
            for a in page.get("alerts", []):
                seen.setdefault(a.get("sid"), str(a.get("error_code") or "?"))
                got += 1
            url = (page.get("meta") or {}).get("next_page_url")
            params = {}
    counts = {}
    for code in seen.values():
        counts[code] = counts.get(code, 0) + 1
    return counts


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7, help="window size in days")
    ap.add_argument("--prefix-digits", type=int, default=3,
                    help="how many leading digits of `to` make a bucket")
    ap.add_argument("--floor", type=int, default=20,
                    help="minimum calls before a bucket's rate is judged")
    ap.add_argument("--threshold", type=float, default=0.10,
                    help="failure share at which a bucket is elevated")
    ap.add_argument("--max-calls", type=int, default=20000,
                    help="stop after this many calls per sweep")
    ap.add_argument("--all-statuses", action="store_true",
                    help="one unfiltered sweep, so busy and no-answer are in "
                         "the denominator too")
    ap.add_argument("--with-alerts", action="store_true",
                    help="also count Debugger alerts in the window, at both "
                         "the error and warning log levels")
    args = ap.parse_args()

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)
    since = (dt.date.today() - dt.timedelta(days=args.days)).isoformat()

    if args.all_statuses:
        calls = list_calls(session, account, since, args.max_calls)
    else:
        calls = (list_calls(session, account, since, args.max_calls, "failed")
                 + list_calls(session, account, since, args.max_calls, "completed"))
        log.info("busy and no-answer are not in this denominator: "
                 "re-run with --all-statuses for the full outcome mix")

    buckets = summarise(calls, args.prefix_digits)
    if not buckets:
        log.info("no outbound calls in the last %d day(s)", args.days)
        return 0

    total = sum(b["total"] for b in buckets.values())
    failed = sum(b["failed"] for b in buckets.values())
    elevated = 0
    for key in sorted(buckets, key=lambda k: -buckets[k]["failed"]):
        direction, prefix = key
        state, detail = verdict(buckets[key], args.floor, args.threshold)
        line = "%-14s %-14s %-8s %s" % (state, direction, prefix, detail)
        if state in ("elevated", "total-failure"):
            elevated += 1
            log.warning(line)
        else:
            log.info(line)

    if args.with_alerts:
        counts = alert_codes(session, since, 10000, ["error", "warning"])
        top = sorted(counts.items(), key=lambda kv: -kv[1])[:8]
        log.info("alerts in window (error and warning): %s",
                 ", ".join("%s=%d" % kv for kv in top) or "none")

    share = (failed / total * 100) if total else 0.0
    log.info("%d outbound call(s), %d failed (%.1f%%), %d elevated bucket(s)",
             total, failed, share, elevated)
    if elevated:
        log.warning("  repair: pull the signalling detail for a call in the worst "
                    "bucket with GET %s/Accounts/%s/Calls/{CallSid}/Events.json, "
                    "then fix the cause it points at: geo permissions, E.164 "
                    "normalisation, or caller ID reputation", BASE, account)
    return 1 if elevated else 0


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

/**
 * Report the outbound call failure rate, bucketed by direction and destination.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is printed, never
 * performed.
 */
const HOST = 'https://api.twilio.com';
const BASE = `${HOST}/2010-04-01`;
const MONITOR = 'https://monitor.twilio.com/v1';

// Statuses that are an outcome. queued, ringing and in-progress have not
// finished, and counting them moves the rate by when the script ran.
const OUTCOMES = new Set(['completed', 'failed', 'busy', 'no-answer', 'canceled']);

/**
 * Bucket a destination by its leading digits. SIP URIs and client identities
 * get their own buckets rather than being stripped to whatever digits they
 * contain.
 */
export function dialPrefix(to, digits = 3) {
  const v = String(to ?? '').trim();
  if (!v) return 'unknown';
  const low = v.toLowerCase();
  if (low.startsWith('sip:') || low.startsWith('sips:')) return 'sip';
  if (low.startsWith('client:')) return 'client';
  const d = v.replace(/[^0-9]/g, '');
  if (!d) return 'unknown';
  return `+${d.slice(0, digits)}`;
}

/**
 * Group outbound calls into direction/prefix buckets of outcomes. Pure, and
 * deliberately tolerant: an unexpected status is skipped rather than counted as
 * a failure. Returns a Map keyed by `${direction}|${prefix}`.
 */
export function summarise(calls, digits = 3) {
  const buckets = new Map();
  for (const c of calls) {
    const status = String(c.status ?? '').trim().toLowerCase();
    if (!OUTCOMES.has(status)) continue;
    const direction = String(c.direction ?? 'unknown').trim().toLowerCase();
    if (!direction.startsWith('outbound')) continue;
    const key = `${direction}|${dialPrefix(c.to, digits)}`;
    if (!buckets.has(key)) {
      buckets.set(key, { total: 0, completed: 0, failed: 0, busy: 0,
                         no_answer: 0, canceled: 0 });
    }
    const b = buckets.get(key);
    b.total += 1;
    b[status.replace('-', '_')] += 1;
  }
  return buckets;
}

/**
 * Judge one bucket. Pure, and the thresholds are arguments so the boundary
 * cases can be tested. Returns [state, detail].
 */
export function verdict(bucket, floor = 20, threshold = 0.10) {
  const total = bucket.total ?? 0;
  const failed = bucket.failed ?? 0;
  const share = total ? failed / total : 0;
  const pct = `${(share * 100).toFixed(1)}%`;

  if (total < floor) {
    return ['low-volume',
      `${total} call(s) is too few to read a rate from: ${failed} failed, ` +
      `which is ${pct} of nothing much.`];
  }
  if (failed === total) {
    return ['total-failure',
      `every one of ${total} call(s) failed. This is not a rate, it is a ` +
      'destination or a permission that is off.'];
  }
  if (share >= threshold) {
    return ['elevated',
      `${failed} of ${total} call(s) failed (${pct}), against a threshold of ` +
      `${(threshold * 100).toFixed(0)}%. busy=${bucket.busy ?? 0} ` +
      `no-answer=${bucket.no_answer ?? 0}.`];
  }
  return ['ok', `${failed} of ${total} call(s) failed (${pct})`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

/** Page the Calls list. next_page_uri here is a path, not an absolute URL. */
export async function listCalls(auth, account, since, limit, status = null) {
  let url = `${BASE}/Accounts/${account}/Calls.json`;
  let params = { 'StartTime>=': since, PageSize: 1000 };
  if (status) params.Status = status;
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.calls ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

/** Error codes in the window, across both log levels, de-duplicated on sid. */
export async function alertCodes(auth, since, limit, levels) {
  const seen = new Map();
  for (const level of levels) {
    let url = `${MONITOR}/Alerts`;
    let params = { LogLevel: level, StartDate: since, PageSize: 1000 };
    while (url && seen.size < limit) {
      const page = await get(auth, url, params);
      for (const a of page.alerts ?? []) {
        if (!seen.has(a.sid)) seen.set(a.sid, String(a.error_code ?? '?'));
      }
      url = page.meta?.next_page_url ?? null;
      params = {};
    }
  }
  const counts = new Map();
  for (const code of seen.values()) counts.set(code, (counts.get(code) ?? 0) + 1);
  return counts;
}

function arg(name, fallback) {
  const i = process.argv.indexOf(name);
  return i === -1 ? fallback : Number(process.argv[i + 1]);
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const days = arg('--days', 7);
  const digits = arg('--prefix-digits', 3);
  const floor = arg('--floor', 20);
  const threshold = arg('--threshold', 0.10);
  const maxCalls = arg('--max-calls', 20000);
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  let calls;
  if (process.argv.includes('--all-statuses')) {
    calls = await listCalls(auth, account, since, maxCalls);
  } else {
    calls = [...await listCalls(auth, account, since, maxCalls, 'failed'),
             ...await listCalls(auth, account, since, maxCalls, 'completed')];
    console.log('busy and no-answer are not in this denominator: ' +
                're-run with --all-statuses for the full outcome mix');
  }

  const buckets = summarise(calls, digits);
  if (buckets.size === 0) {
    console.log(`no outbound calls in the last ${days} day(s)`);
    return;
  }

  let total = 0;
  let failed = 0;
  let elevated = 0;
  const keys = [...buckets.keys()].sort((a, b) => buckets.get(b).failed - buckets.get(a).failed);
  for (const k of keys) {
    const b = buckets.get(k);
    total += b.total;
    failed += b.failed;
    const [direction, prefix] = k.split('|');
    const [state, detail] = verdict(b, floor, threshold);
    const line = `${state.padEnd(14)} ${direction.padEnd(14)} ${prefix.padEnd(8)} ${detail}`;
    if (state === 'elevated' || state === 'total-failure') { elevated += 1; console.warn(line); }
    else console.log(line);
  }

  if (process.argv.includes('--with-alerts')) {
    const counts = await alertCodes(auth, since, 10000, ['error', 'warning']);
    const top = [...counts.entries()].sort((a, b) => b[1] - a[1]).slice(0, 8);
    console.log(`alerts in window (error and warning): ${
      top.map(([c, n]) => `${c}=${n}`).join(', ') || 'none'}`);
  }

  const share = total ? (failed / total) * 100 : 0;
  console.log(`${total} outbound call(s), ${failed} failed (${share.toFixed(1)}%), ` +
              `${elevated} elevated bucket(s)`);
  if (elevated) {
    console.warn('  repair: pull the signalling detail for a call in the worst bucket ' +
                 `with GET ${BASE}/Accounts/${account}/Calls/{CallSid}/Events.json, then ` +
                 'fix the cause it points at: geo permissions, E.164 normalisation, ' +
                 'or caller ID reputation');
  }
  process.exitCode = elevated ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not execute main() and fail on the missing credentials.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from twilio_call_failure_rate_audit import dial_prefix, summarise, verdict


def calls(n, status, to="+15005550006", direction="outbound-api"):
    return [{"status": status, "to": to, "direction": direction} for _ in range(n)]


def test_prefix_uses_leading_digits_only():
    assert dial_prefix("+15005550006") == "+150"
    assert dial_prefix("+44 20 7946 0000", digits=2) == "+44"


def test_sip_and_client_destinations_are_their_own_buckets():
    assert dial_prefix("sip:pbx@example.com") == "sip"
    assert dial_prefix("client:alice") == "client"
    assert dial_prefix("") == "unknown"


def test_calls_still_in_flight_are_not_an_outcome():
    # Counting ringing calls would move the rate with the clock rather than
    # with anything that happened.
    assert summarise(calls(5, "ringing")) == {}


def test_inbound_calls_are_not_in_the_outbound_rate():
    assert summarise(calls(5, "failed", direction="inbound")) == {}


def test_buckets_split_on_direction_and_prefix():
    rows = (calls(3, "failed") + calls(2, "completed")
            + calls(4, "failed", direction="outbound-dial"))
    buckets = summarise(rows)
    assert set(buckets) == {("outbound-api", "+150"), ("outbound-dial", "+150")}
    assert buckets[("outbound-api", "+150")]["total"] == 5
    assert buckets[("outbound-dial", "+150")]["failed"] == 4


def test_a_small_bucket_is_never_elevated():
    state, detail = verdict({"total": 4, "failed": 3}, floor=20)
    assert state == "low-volume"
    assert "too few" in detail


def test_exactly_on_the_threshold_is_elevated():
    state, _ = verdict({"total": 100, "failed": 10}, floor=20, threshold=0.10)
    assert state == "elevated"


def test_just_below_the_threshold_is_ok():
    assert verdict({"total": 100, "failed": 9}, floor=20, threshold=0.10)[0] == "ok"


def test_everything_failing_is_not_reported_as_a_rate():
    state, detail = verdict({"total": 40, "failed": 40}, floor=20)
    assert state == "total-failure"
    assert "permission" in detail


def test_a_bucket_with_no_calls_does_not_divide_by_zero():
    assert verdict({"total": 0, "failed": 0})[0] == "low-volume"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { dialPrefix, summarise, verdict } from './twilio-call-failure-rate-audit.mjs';

function calls(n, status, to = '+15005550006', direction = 'outbound-api') {
  return Array.from({ length: n }, () => ({ status, to, direction }));
}

test('prefix uses leading digits only', () => {
  assert.equal(dialPrefix('+15005550006'), '+150');
  assert.equal(dialPrefix('+44 20 7946 0000', 2), '+44');
});

test('sip and client destinations are their own buckets', () => {
  assert.equal(dialPrefix('sip:pbx@example.com'), 'sip');
  assert.equal(dialPrefix('client:alice'), 'client');
  assert.equal(dialPrefix(''), 'unknown');
});

test('calls still in flight are not an outcome', () => {
  assert.equal(summarise(calls(5, 'ringing')).size, 0);
});

test('inbound calls are not in the outbound rate', () => {
  assert.equal(summarise(calls(5, 'failed', '+15005550006', 'inbound')).size, 0);
});

test('buckets split on direction and prefix', () => {
  const rows = [...calls(3, 'failed'), ...calls(2, 'completed'),
                ...calls(4, 'failed', '+15005550006', 'outbound-dial')];
  const buckets = summarise(rows);
  assert.deepEqual([...buckets.keys()].sort(),
                   ['outbound-api|+150', 'outbound-dial|+150']);
  assert.equal(buckets.get('outbound-api|+150').total, 5);
  assert.equal(buckets.get('outbound-dial|+150').failed, 4);
});

test('a small bucket is never elevated', () => {
  const [state, detail] = verdict({ total: 4, failed: 3 }, 20);
  assert.equal(state, 'low-volume');
  assert.match(detail, /too few/);
});

test('exactly on the threshold is elevated', () => {
  assert.equal(verdict({ total: 100, failed: 10 }, 20, 0.10)[0], 'elevated');
});

test('just below the threshold is ok', () => {
  assert.equal(verdict({ total: 100, failed: 9 }, 20, 0.10)[0], 'ok');
});

test('everything failing is not reported as a rate', () => {
  const [state, detail] = verdict({ total: 40, failed: 40 }, 20);
  assert.equal(state, 'total-failure');
  assert.match(detail, /permission/);
});

test('a bucket with no calls does not divide by zero', () => {
  assert.equal(verdict({ total: 0, failed: 0 })[0], 'low-volume');
});