AssemblyAI Transcription Error Sentinel for Kestra

id: assemblyai-transcription-error-sentinel
namespace: company.team
description: |
  Every hour, list recent transcripts from the AssemblyAI API (read-only: no
  audio is submitted and no transcript is deleted) and classify speech-to-text
  as HEALTHY, WARNING, CRITICAL, LOW_VOLUME or UNREACHABLE. Log quiet runs and
  alert Slack when transcripts start ending in error, with the most common
  error, or when jobs sit queued or processing for too long.

inputs:
  - id: lookback_hours
    type: INT
    defaults: 6
    displayName: "Lookback Window (hours)"
    description: "Only transcripts created in the last this many hours are judged."
  - id: warning_error_rate
    type: FLOAT
    defaults: 3.0
    displayName: "Warning Error Rate (%)"
    description: "At or above this share of transcripts in error among finished
      ones, the check is WARNING."
  - id: critical_error_rate
    type: FLOAT
    defaults: 10.0
    displayName: "Critical Error Rate (%)"
    description: "At or above this share of transcripts in error among finished
      ones, the check is CRITICAL."
  - id: min_transcripts
    type: INT
    defaults: 20
    displayName: "Minimum Transcripts"
    description: "Below this many finished transcripts, the rate is too noisy to
      judge and the run is LOW_VOLUME (stuck jobs still warn)."
  - id: stuck_minutes
    type: INT
    defaults: 60
    displayName: "Stuck Threshold (minutes)"
    description: "A transcript still queued or processing this long after it was
      created is counted as stuck."

triggers:
  - id: hourly_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check transcription errors every hour. Shipped disabled; enable it
      once the AssemblyAI API key secret is set and a manual run has been
      verified.
    cron: "25 * * * *"
    disabled: true

tasks:
  - id: list_transcripts
    type: io.kestra.plugin.core.http.Request
    description: List the 200 most recent transcripts with their status, creation
      time and error from the AssemblyAI transcript list. This is a GET; no
      audio is submitted and nothing is deleted.
    uri: "https://api.assemblyai.com/v2/transcript?limit=200"
    method: GET
    headers:
      Authorization: "{{ secret('ASSEMBLYAI_API_KEY') }}"
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3
    allowFailure: true

  - id: summarize_transcripts
    type: io.kestra.plugin.core.debug.Return
    description: Keep transcripts created in the lookback window, compute the share
      that ended in error among finished ones, find the most common error, count
      jobs stuck in queued or processing past the threshold, and classify in one
      jq query. A missing or non-200 response becomes UNREACHABLE.
    format: >-
      {{ (outputs.list_transcripts.code is not defined or
      outputs.list_transcripts.code != 200) ?
      '{"status":"UNREACHABLE","finished":0,"completed":0,"errors":0,"rate":0,"in_flight":0,"stuck":0,"top_error":"none"}'
      : (fromJson(outputs.list_transcripts.body) | jq(' "' ~ (now() |
      dateAdd(-inputs.lookback_hours, 'HOURS') | date("yyyy-MM-dd'T'HH:mm:ss",
      timeZone="UTC")) ~ '" as $since | "' ~ (now() |
      dateAdd(-inputs.stuck_minutes, 'MINUTES') | date("yyyy-MM-dd'T'HH:mm:ss",
      timeZone="UTC")) ~ '" as $stuck_before | [(.transcripts // [])[] |
      select((.created // "") >= $since)] as $t | [$t[] | select(.status ==
      "completed" or .status == "error")] as $f | [$f[] | select(.status ==
      "error")] as $bad | [$t[] | select(.status == "queued" or .status ==
      "processing")] as $open | [$open[] | select((.created // "") <
      $stuck_before)] as $stuck | ($f | length) as $n | (if $n == 0 then 0 else
      ((($bad | length) * 1000 / $n) | floor) / 10 end) as $r | {finished: $n,
      completed: ($n - ($bad | length)), errors: ($bad | length), rate: $r,
      in_flight: ($open | length), stuck: ($stuck | length), top_error: (if
      ($bad | length) == 0 then "none" else ([$bad[] | ((.error // "no error
      message") | tostring | split("\"") | join("") | split("\n") | join(" ") |
      .[0:100])] | group_by(.) | map({e: .[0], c: length}) | max_by(.c) | (.e +
      " (x" + (.c | tostring) + ")")) end), status: (if $n >= ' ~
      inputs.min_transcripts ~ ' and $r >= ' ~ inputs.critical_error_rate ~ '
      then "CRITICAL" elif ($n >= ' ~ inputs.min_transcripts ~ ' and $r >= ' ~
      inputs.warning_error_rate ~ ') or ($stuck | length) > 0 then "WARNING"
      elif $n < ' ~ inputs.min_transcripts ~ ' then "LOW_VOLUME" else "HEALTHY"
      end)} ') | first | toJson) }}

  - id: route_by_errors
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log when transcription is healthy or volume is too
      low to judge, a Slack message when errors cross the warning line or jobs
      are stuck, or a Slack page when errors cross the critical line or
      AssemblyAI could not be read.
    value: "{{ fromJson(outputs.summarize_transcripts.value).status }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          description: Log a quiet line when the check is healthy; no Slack message is sent.
          message: "AssemblyAI transcription healthy: {{
            fromJson(outputs.summarize_transcripts.value).errors }} of {{
            fromJson(outputs.summarize_transcripts.value).finished }} finished
            transcripts in error ({{
            fromJson(outputs.summarize_transcripts.value).rate }}%), {{
            fromJson(outputs.summarize_transcripts.value).in_flight }} in
            flight."
      "LOW_VOLUME":
        - id: log_low_volume
          type: io.kestra.plugin.core.log.Log
          description: Log a quiet line when there is too little activity in the window to
            judge; no Slack message is sent.
          message: "AssemblyAI transcription: only {{
            fromJson(outputs.summarize_transcripts.value).finished }} finished
            transcript(s) in the last {{ inputs.lookback_hours }} hour(s), below
            the minimum of {{ inputs.min_transcripts }}; rate not judged this
            run."
      "WARNING":
        - id: alert_warning_errors
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post a Slack message when the transcript error rate is at or above
            the warning line or jobs are stuck in the queue.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "AssemblyAI transcription WARNING: {{ fromJson(outputs.summarize_transcripts.value).errors }} of {{ fromJson(outputs.summarize_transcripts.value).finished }} finished transcripts in error ({{ fromJson(outputs.summarize_transcripts.value).rate }}%); {{ fromJson(outputs.summarize_transcripts.value).stuck }} stuck over {{ inputs.stuck_minutes }} min. Most common error: {{ fromJson(outputs.summarize_transcripts.value).top_error }}. Execution {{ execution.id }}."
            }
      "CRITICAL":
        - id: alert_critical_errors
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the transcript error rate is at or above the
            critical line, which usually means audio URLs, formats or the
            account are broken.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "AssemblyAI transcription CRITICAL: {{ fromJson(outputs.summarize_transcripts.value).errors }} of {{ fromJson(outputs.summarize_transcripts.value).finished }} finished transcripts in error ({{ fromJson(outputs.summarize_transcripts.value).rate }}%). Most common error: {{ fromJson(outputs.summarize_transcripts.value).top_error }}. Calls and uploads are not being transcribed; check audio URLs and the account now. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_api
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the transcript list could not be read, indicating
            an outage or an invalid API key.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "AssemblyAI transcription sentinel UNREACHABLE: the transcript list did not return a usable response (HTTP {{ outputs.list_transcripts.code ?? 'no response' }}). Check the API key. Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        description: Fallback for any classification the cases above do not cover, so an
          unexpected value is visible in the logs.
        message: "AssemblyAI transcription sentinel completed with an unexpected
          classification: {{ outputs.summarize_transcripts.value }}."

  - id: log_audit_result
    type: io.kestra.plugin.core.log.Log
    description: Record one audit line for every run, whichever branch fired, so the
      error history stays queryable even if Slack delivery fails.
    message: "AssemblyAI transcription sentinel audit: lookback_hours={{
      inputs.lookback_hours }} http={{ outputs.list_transcripts.code ?? 'n/a' }}
      summary={{ outputs.summarize_transcripts.value }} execution={{
      execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the sentinel flow itself fails, so a broken check
      is never mistaken for healthy transcription.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "AssemblyAI transcription sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). Transcription health is now unknown; check the task logs and the AssemblyAI dashboard."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ fromJson(outputs.summarize_transcripts.value).status }}"

Call recordings, meeting notes, podcast pipelines and voice-agent logs that go through AssemblyAI fail one transcript at a time: a storage bucket starts returning 403 for the audio URL, a recorder switches to a codec that cannot be decoded, a file has no audio track, or the account runs out of balance. Each job ends with status: error and an error string, the webhook that should have delivered the text never fires the way the app expects, and the gap shows up days later as empty call summaries. This blueprint turns a scheduled, read-only call to the AssemblyAI API into a transcription health gate: io.kestra.plugin.core.http.Request lists recent transcripts, one jq query computes the error rate, the most common error and the number of stuck jobs, and the result routes to a quiet log or a Slack alert.

How it works

list_transcripts (io.kestra.plugin.core.http.Request) calls GET /v2/transcript?limit=200 with the API key in the Authorization header, which returns the most recent transcripts, newest first. It is a read: no audio is submitted and nothing is deleted. Retries cover short network blips, and allowFailed plus allowFailure keep an API error from aborting the flow so it can be reported instead.
summarize_transcripts (io.kestra.plugin.core.debug.Return) first guards for a missing or non-200 response (UNREACHABLE). Otherwise one jq query keeps transcripts whose created time falls in the lookback window, computes the share of finished ones (completed or error) that ended in error, groups them by error text, and counts jobs still queued or processing that were created more than stuck_minutes ago. With at least min_transcripts finished, a rate at or above critical_error_rate is CRITICAL and one at or above warning_error_rate is WARNING; any stuck job is also WARNING; fewer finished transcripts is LOW_VOLUME; otherwise HEALTHY.
route_by_errors (io.kestra.plugin.core.flow.Switch) branches on that classification: HEALTHY and LOW_VOLUME log quietly, WARNING posts a Slack note, CRITICAL and UNREACHABLE page Slack. A defaults fallback logs anything unexpected.
log_audit_result (io.kestra.plugin.core.log.Log) writes one audit line per run with the HTTP status and the full summary.
The flow-level errors block pages Slack if the flow itself fails, and the hourly_check trigger runs at 25 minutes past every hour, shipped disabled: true.

What you get

An hourly, read-only check of speech-to-text with no custom scripting and no extra plugin to install.
A status output (HEALTHY / WARNING / CRITICAL / LOW_VOLUME / UNREACHABLE) other flows can gate on, for example to hold call-summary jobs or resubmit failed audio once the cause is fixed.
Slack messages with the error rate, the number of stuck jobs and the most common error text, which usually names the cause (unreachable URL, unsupported file, no speech).
An audit line on every run, separate from Slack.

Who it's for

Contact centres, sales-call and meeting tools that transcribe recordings with AssemblyAI.
Media and podcast teams running batch transcription in Kestra.
Engineers who own a voice pipeline and want to hear about broken audio before customers do.

Why orchestrate this with Kestra

AssemblyAI reports each transcript's outcome on its own, through polling or a webhook. Kestra turns a plain, read-only list call into a scheduled, typed classification with retries, a failure path and an execution history, so a rising error rate or a stuck queue becomes one clear signal, and the status output lets your transcription flows react automatically.

Prerequisites

An AssemblyAI account with an API key from the dashboard.
A Slack incoming webhook for the WARNING, CRITICAL, UNREACHABLE and flow-failure alerts.

Secrets

ASSEMBLYAI_API_KEY: the API key of the project whose transcripts you want to watch.
SLACK_WEBHOOK: Slack incoming webhook used for every alert branch and the flow-failure alert.

In Kestra OSS, secrets are supplied as environment variables prefixed SECRET_ and base64-encoded, for example SECRET_ASSEMBLYAI_API_KEY=$(echo -n 'xxxxxxxx' | base64), and read back with {{ secret('ASSEMBLYAI_API_KEY') }}. Use the Enterprise secrets backend for encryption at rest and access control.

Quick start

Submit a few transcripts of public audio files, plus one with an unreachable URL to produce an error.
Add ASSEMBLYAI_API_KEY and SLACK_WEBHOOK as secrets.
Run the flow manually with a low min_transcripts and compare the counts and error text in the logs with the transcript history in the dashboard.
Enable hourly_check.

How to extend

Follow page_details.prev_url to read further back when you transcribe more than 200 files in the window.
Use the EU host (api.eu.assemblyai.com) if your project is in the EU region.
Pass status=error and created_on to build a daily list of failed transcript IDs to resubmit.
Store the last status in a KV store and alert only when it changes.

Inputs

lookback_hours: type INT, default 6.
warning_error_rate: type FLOAT, default 3.0 (percent).
critical_error_rate: type FLOAT, default 10.0 (percent).
min_transcripts: type INT, default 20.
stuck_minutes: type INT, default 60.

Outputs

status: one of HEALTHY, WARNING, CRITICAL, LOW_VOLUME, UNREACHABLE.
Example path: {{ fromJson(outputs.summarize_transcripts.value).top_error }} for the most common error.

Pitfalls

Only the 200 most recent transcripts are read; high-volume projects need paging or a shorter window.
The window compares the created timestamp as a UTC string; if your account shows another time zone, widen lookback_hours a little.
Very long files can legitimately take a while; raise stuck_minutes if you transcribe multi-hour recordings.
The list only covers the last 30 days of transcripts, and deleted transcripts are not returned.
This flow only reads; it never submits, deletes or resubmits a transcript.