What if a dltHub pipeline fails without errors?

TL/DR: dltHub's @job decorator can (add of a couple of weeks ago) ​notify on any Python exception. But what if a pipeline doesn't raise one? If it died before its own trampoline body ever started, about 1 minute in and without any logs? No exception to be raised means no alert to be triggered and nobody notices until someone stumbles over it. So I needed a small watchdog to check every job's latest run, retries once if failed and alerts if the retry fails too.

The blind spot in my own alert

A few days ago I wired all of my dltHub jobs to alert Teams on failure. This uses a decorator that wraps each job's trampoline in a try/except: raise inside the job, the decorator catches it and posts to Teams. That works really well for actual Python errors (bad credentials, timeouts, etc.).

dltHub notifies Slack and email. But my org runs on Teams 😩

dltHub ships Slack and email alerts out of the box. My org runs on Teams, so I built the third one myself and wired it into all ~70 pipelines with one decorator shadow.

sfrt.ioMartin Seifert

Before this I had to rely on a daily routine (usually triggered manually when I sat down at my desk in the morning). I checked if any job failed and started investigating.

Now I ran into a new issue my alerts didn't catch: A pipeline job ran for just over 1 minute and produced no log output at all, then showed up in the dltHub UI as failed. Whatever killed it happened even before the trampoline ran: a platform or container-level crash or something like that 🤷 The decorator can only alert on exceptions it can see, and this one never reached Python at all.

Before I had alerts, I had an (over time) ​evolving daily routine. I started this evolution with simply looking into the dashboard every morning. If something failed, I'd look into it. Next, I created skills for my agent and told it to look into the job logs for me. And then let it fix eventual failures. The skill was nice but consumed quite many tokens as it had to reason over what to do every morning. However, this routine (at least parts of it) is fairly deterministic, so the next step was to script a watchdog...

The watchdog is a really just a Python script. Every time it runs, it scans __deployment__.py for @job-decorated functions (via AST, because a regex would trip on multi-line callys), then asks the dltHub CLI for each job's most recent run.

Here's the decision per job, the state is persisted in a JSON locally:

def process_job(job_name: str, state: dict) -> None:
    latest = get_latest_run(job_name)
    if latest is None:
        return
    run_number, status = latest
    job_state = state.setdefault(
        job_name, {"last_processed_run": -1, "retry_pending": False}
    )

    if run_number == job_state["last_processed_run"]:
        return  # already handled this run

    if status in ("running", "pending"):
        return  # still in flight, re-check next tick

    if status in ("completed", "cancelled"):
        job_state["last_processed_run"] = run_number
        job_state["retry_pending"] = False
        return

    if status == "failed":
        if not job_state["retry_pending"]:
            trigger_retry(job_name)
            job_state["last_processed_run"] = run_number
            job_state["retry_pending"] = True
        else:
            run_url = get_run_url(job_name, run_number)
            alert_teams(job_name, run_number, run_url)
            job_state["last_processed_run"] = run_number
            job_state["retry_pending"] = False
        return

First failure on a job means: retry it, mark retry_pending, nothing to notify yet. The retry's own outcome becomes that job's new "latest run" on the next tick, so if it fails again, retry_pending is already true and this time it alerts instead of retrying again. One retry, never an infinite loop.

Why local, not on dltHub Runtime?

Checking status and triggering a retry both go through the dlthub CLI, which needs an authenticated session through a device-code login, cached locally. This (to my knowledge) would not quite work through the dltHub API, but I might very well be wrong on this... Please tell me if I am! The API path would enable a fully autonomous watchdog, as an OpenCode server (or something similar) could then also immediately modify, deploy and test the pipeline. That would be pretty neat 🤩

Running the watchdog on dltHub runtime itself would mean bootstrapping that auth inside a fresh ephemeral container on every run, with nowhere to persist or refresh it. Running it on my own machine reuses the CLI session that's already sitting there and lets me just click through the device flow again when it expires.

Also, if I'm not on my desk, I won't look into any failures anyway. Larger teams (than single persons) might use a different approach here 😅

The watchdog

Before checking any jobs, the watchdog checks whether the cached CLI session is still valid. If not, it pops up a visible console running dlthub login to get a new token.

def prompt_login() -> None:
    subprocess.Popen(
        [
            "cmd", "/c", "start", "dltHub Login Required",
            str(DLTHUB_EXE), "login",
        ],
        cwd=DLTHUB_DIR,
        creationflags=subprocess.CREATE_NEW_CONSOLE,
    )

Afterwards, the actual checks in process_job() happen. That's really all there is... I registered this as Windows scheduled job to run twice on a normal workday: immediately after login and at 10:00 AM, when it's fairly likely I'm actually sitting on my desk. Fairly 😅

$atLogon = New-ScheduledTaskTrigger -AtLogOn
$daily10 = New-ScheduledTaskTrigger -Daily -At "10:00"

$principal = New-ScheduledTaskPrincipal -UserId $env:USERNAME -LogonType Interactive -RunLevel Limited

Register-ScheduledTask -TaskName "dltHub Job Watchdog" `
    -Action $action -Trigger @($atLogon, $daily10) -Principal $principal `
    -Settings $settings -Force | Out-Null

The full script:

"""dltHub job watchdog: detects jobs that failed with no exception raised
(a platform/container-level crash the `@job` decorator's try/except in
__deployment__.py cannot see), retries each such failure once, and alerts
via Teams only if the retry also fails.

Watches every `@job`-decorated trampoline in __deployment__.py, and for
each one only ever inspects the single latest run — never a history scan.
See docs/superpowers/specs/2026-08-27-dlthub-job-watchdog-design.md.

Run via Windows Task Scheduler (install_task.ps1), not as a long-running
process: two triggers, at log-on and daily 10:00.
"""

from __future__ import annotations

import ast
import json
import os
import re
import subprocess
import sys
from pathlib import Path

WATCHDOG_DIR = Path(__file__).resolve().parent
DLTHUB_DIR = WATCHDOG_DIR.parent / "dltHub"
STATE_PATH = WATCHDOG_DIR / "state.json"
DEPLOYMENT_FILE = DLTHUB_DIR / "__deployment__.py"


def load_state() -> dict:
    """Read state.json; a missing or corrupt file is treated as empty
    state (fresh start), never a crash."""
    if not STATE_PATH.exists():
        return {}
    try:
        return json.loads(STATE_PATH.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}


def save_state(state: dict) -> None:
    STATE_PATH.write_text(json.dumps(state, indent=2), encoding="utf-8")


def discover_jobs(deployment_path: Path = DEPLOYMENT_FILE) -> list[str]:
    """Every function decorated with `@job(...)` in __deployment__.py, via
    AST (handles multi-line decorator calls a line-based regex would
    miss). Re-scanned every tick, so adding/removing a job there is picked
    up automatically."""
    tree = ast.parse(deployment_path.read_text(encoding="utf-8"))
    names = []
    for node in tree.body:
        if not isinstance(node, ast.FunctionDef):
            continue
        for dec in node.decorator_list:
            target = dec.func if isinstance(dec, ast.Call) else dec
            if isinstance(target, ast.Name) and target.id == "job":
                names.append(node.name)
                break
    return names


DLTHUB_EXE = DLTHUB_DIR / ".venv" / "Scripts" / "dlthub.exe"

_RUN_NUMBER_RE = re.compile(r"^Run #\s+(\d+)", re.MULTILINE)
_STATUS_RE = re.compile(r"^Status\s+(\S+)", re.MULTILINE)
_RUN_URL_RE = re.compile(r"available at (\S+)")


def run_dlthub(*args: str, timeout: float | None = None) -> subprocess.CompletedProcess:
    """Run the dlthub CLI with cwd=dlt/dltHub/ (secrets/config only
    resolve there) and --non-interactive, capturing text stdout/stderr."""
    return subprocess.run(
        [str(DLTHUB_EXE), *args, "--non-interactive"],
        cwd=DLTHUB_DIR,
        capture_output=True,
        text=True,
        timeout=timeout,
    )


def get_latest_run(job_name: str) -> tuple[int, str] | None:
    """Latest run number + status for one job, or None if it can't be
    parsed (e.g. the job has never run)."""
    result = run_dlthub(
        "job", "runs", "info", f"jobs.__deployment__.{job_name}", timeout=30
    )
    output = result.stdout or ""
    run_match = _RUN_NUMBER_RE.search(output)
    status_match = _STATUS_RE.search(output)
    if not run_match or not status_match:
        return None
    return int(run_match.group(1)), status_match.group(1)


def trigger_retry(job_name: str) -> None:
    run_dlthub("job", "run", f"jobs.__deployment__.{job_name}", timeout=30)


def get_run_url(job_name: str, run_number: int) -> str | None:
    result = run_dlthub(
        "job", "runs", "show", f"jobs.__deployment__.{job_name}", str(run_number),
        timeout=30,
    )
    match = _RUN_URL_RE.search(result.stdout or "")
    return match.group(1) if match else None


def check_login() -> bool:
    """True if the cached dltHub CLI session is still valid. Forces the
    device flow (`--device`) so this never tries to pop a browser loopback
    server; a short timeout guards against the device flow sitting there
    polling for a code that will never come in an unattended run."""
    try:
        result = run_dlthub("login", "--device", timeout=8)
    except subprocess.TimeoutExpired as e:
        return "Already logged in" in (e.stdout or "")
    return "Already logged in" in (result.stdout or "")


def prompt_login() -> None:
    """Not logged in: pop a visible console running the interactive
    (browser-loopback) login flow — the watchdog only ever runs while
    someone is at the machine (at-log-on trigger, or the daily check with
    the user present), so this is always reachable. No Teams alert here
    (unlike alert_teams) — a session expiring is routine, not something
    worth a notification every time it happens."""
    subprocess.Popen(
        [
            "cmd", "/c", "start", "dltHub Login Required",
            str(DLTHUB_EXE), "login",
        ],
        cwd=DLTHUB_DIR,
        creationflags=subprocess.CREATE_NEW_CONSOLE,
    )


def alert_teams(job_name: str, run_number: int, run_url: str | None) -> None:
    sys.path.insert(0, str(DLTHUB_DIR))
    from _teams_notify_common import send_teams_message

    send_teams_message(
        f"Watchdog: {job_name} failed twice (retry did not help)",
        f"Run #{run_number} failed. An automatic retry was already "
        "attempted and also failed... this needs a look.",
        url=run_url,
    )


def process_job(job_name: str, state: dict) -> None:
    """Apply the retry-once-then-alert decision for one job's latest run,
    mutating `state` in place. Only ever looks at the single latest run —
    never a history scan."""
    latest = get_latest_run(job_name)
    if latest is None:
        return
    run_number, status = latest
    job_state = state.setdefault(
        job_name, {"last_processed_run": -1, "retry_pending": False}
    )

    if run_number == job_state["last_processed_run"]:
        return  # already handled this run

    if status in ("running", "pending"):
        return  # still in flight, re-check next tick

    if status in ("completed", "cancelled"):
        job_state["last_processed_run"] = run_number
        job_state["retry_pending"] = False
        return

    if status == "failed":
        if not job_state["retry_pending"]:
            trigger_retry(job_name)
            job_state["last_processed_run"] = run_number
            job_state["retry_pending"] = True
        else:
            run_url = get_run_url(job_name, run_number)
            alert_teams(job_name, run_number, run_url)
            job_state["last_processed_run"] = run_number
            job_state["retry_pending"] = False
        return

    # Unknown status: record it so we don't loop on it forever, but don't
    # trigger a retry or alert for something we don't recognize.
    job_state["last_processed_run"] = run_number


def main() -> None:
    os.chdir(DLTHUB_DIR)

    if not check_login():
        prompt_login()
        return

    state = load_state()
    for job_name in discover_jobs():
        try:
            process_job(job_name, state)
        except Exception as e:
            print(f"watchdog: error processing {job_name}: {e}")
        save_state(state)


if __name__ == "__main__":
    main()

Alternatively or additionally, It's also possible to add a downstream job to every job on the "failed" path: 👉 https://dlthub.com/docs/hub/pipeline-operations/triggers#basic-triggers

This doesn't have access to an exception raised in the job, but it covers container runtime issues, too... For notifications in Teams that generic failure notification would look something like this:

@_dlthub_job(
    trigger=[job_fail(f"__deployment__.{_name}") for _name in __all__ if _name != "job_failure_watchdog"],
    execute={"timeout": "5m"},
)
def job_failure_watchdog(run_context: TJobRunContext = None):
    """Fires on any job's platform-recorded failure, exception or not."""
    trigger = run_context["trigger"] if run_context else ""
    failed_job_ref = trigger.split(":", 1)[-1] if ":" in trigger else trigger
    try:
        from _teams_notify_common import send_teams_message
        send_teams_message(
            f"Job run failed (platform-detected): {failed_job_ref}",
            "No exception was necessarily raised in-process — this can be a "
            "timeout, OOM kill, or container crash. Check dltHub run logs.",
            url=_job_run_url(),
        )
    except Exception as notify_err:
        print(f"Failed to send Teams notification for {failed_job_ref}: {notify_err}")