> ## Content Index
> Fetch the complete content index at: https://www.sfrt.io/llms.txt
> Use this file to discover other available public pages before exploring further.

# My pipeline inspector got hands: retrying and fixing failed dltHub jobs with a Cortex agent
- URL: https://www.sfrt.io/my-pipeline-inspector-got-hands-retrying-and-fixing-failed-dlthub-jobs-with-a-cortex-agent/
- Published: 2026-11-06T07:00:29.000Z
- Updated: 2026-11-06T07:00:29.000Z
- Description: Last time the agent only read logs and wrote a diagnosis. Now it pushes commits to master 🤖
- Author: Martin Seifert
- Tags: Infrastructure

TL/DR: the Cortex agent I built for failed dltHub jobs has grown hands. When something gets flaky it fires a single retry. When the bug sits in code or config or the pipeline simply needs s slight adjustment for whatever reason, it edits the repo, deploys the updated pipeline, and re-runs the job. And undoes the whole thing if that run fails 😅

---

In a previous post I approached self-healing pipelines by creating a job inspector for dlt pipelines: when a dltHub job fails, the `job.fail` trigger spawns a second dltHub job. That second job asks a Cortex agent what went wrong, a card shows up in Teams.

[Towards self-healing data pipelines: dltHub inspector in ❄️ CortexA Snowflake Cortex Agent for dltHub jobs![](https://storage.ghost.io/c/7d/94/7d942fe1-7868-4a1a-b2c9-4eb415b1a546/content/images/icon/headshot_ring-10b7dfcb-6040-440d-af69-87135a30c973.png)sfrt.ioMartin Seifert![](https://storage.ghost.io/c/7d/94/7d942fe1-7868-4a1a-b2c9-4eb415b1a546/content/images/thumbnail/inspector_agents-cc768ca4-21c1-47a6-98a1-84b18fd8edc6.png)](https://www.sfrt.io/towards-self-healing-data-pipelines-dlthub-inspector-in-cortex/)

Handy, because it meant I knew what to look at in the morning in case a notification arrived overnight. But that's obviously just a first step. And the reason why I built this in Cortex instead of using dltHubs own [agent runtime](https://dlthub.com/products/dlthub?ref=sfrt.io) remains: For now the agent looks at dlt pipelines, but it might very well expand to all kinds of transformations (tasks or dbt or what have you) or other downstream parts (for instance a CoWork agent not being able to answer because the user doesn't have read priviledges to a recently replaces table).

Now, I had some time to validate the inspector does its job well and also built some confidence in agentic-engineered fixes of pipelines. So I felt the time is right to allow the inspector to apply such fixes itself. With guardrails, of course 😜 Everything below builds on the read-only tools from last time, so I'll focus on what's new.

Snowflake is working in a [similar direction](https://www.snowflake.com/en/blog/autonomous-data-engineering/?ref=sfrt.io), and I am really curious how they approach this 😎 I presume they will focus on all the fun stuff happening inside Snowflake. Hence, I leave that part out for now - I made some experimental steps here, too, but the dlt-part is way more mature and something Snowflake will not cover themselves immediately. Well, it would be rather surprising if they did, anyway... 🙃

## How it fits together

0:00 

/0:07 

1× 

As before, when a job fails, the trigger provides the run id of the failed run to the agent, and the agent pulls what it needs: the run record, the tail of the log, the runs around it, plus the job's source code from the git repo Snowflake has mounted. 

The challenge here was: How does the agent get information from an external system? Cortex agents have a [code execution sandbox](https://docs.snowflake.com/en/user-guide/snowflake-cortex/cortex-agents-code-execution-tool?ref=sfrt.io), but (currently) no external access integration. So those read-only tools are UDFs/procedures, running on a warehouse.

The result of the inspector is a failure classification such as `transient`, `code`, `config`, `upstream_data`, and a handful of others. Three of those classes can lead to an action: `transient` to a retry, `code` and `config` to a fix. The inspector needed a few new tools. They're procedures, too, registered like the read tools were. Here's the spec for `Apply_Fix` as an example:

```json
{
  "tool_spec": {
    "type": "generic",
    "name": "Apply_Fix",
    "description": "Repairs a failed run caused by code or config: applies exact text edits to the job's pipeline files on master, deploys, re-runs the job and rolls back by itself if the re-run fails. Refuses when the run was already attempted, started by an attempt, is not failed, or the job had 3 attempts in 24 hours. Returns once the fix is started.",
    "input_schema": {
      "type": "object",
      "properties": {
        "failed_run_id": {"description": "dltHub job run id (uuid) of the failed run", "type": "string"},
        "classification": {"description": "your classification, code or config", "type": "string"},
        "confidence": {"description": "your confidence, must be high", "type": "string"},
        "edits": {"description": "JSON array of {\"path\": file path relative to the workspace root, \"old\": exact text to replace, occurring once, \"new\": replacement}", "type": "string"},
        "rationale": {"description": "one sentence: what is wrong and why this edit fixes it", "type": "string"}
      },
      "required": ["failed_run_id", "classification", "confidence", "edits", "rationale"]
    }
  }
}
...
"Apply_Fix": {
  "execution_environment": {"type": "warehouse", "warehouse": "XS_G1"},
  "identifier": "META.LOAD.P_DLTHUB_APPLY_FIX",
  "name": "P_DLTHUB_APPLY_FIX(VARCHAR, VARCHAR, VARCHAR, VARCHAR, VARCHAR)",
  "type": "procedure"
}
```

Why is `edits` a string and not an array? That seemed to be easier that ensuring the agent produces a valid JSON 😅 Theagent's model produces JSON in a string and the procedure parses and validates it before anything happens. The other new tools (`Retry_Job`, `Second_Opinion`, ...) follow a similar pattern.

The agent also writes its own run id and the id of its Cortex trace into the job output. A reviewer can later retrieve exactly what the inspector looked at and what it concluded. But I'll cover that part another time 😎

## One table to remember every attempt

Before the fun part, there's bookkeeping. Every retry and every fix gets a row in a log table, whether it worked or not. That table will be useful for the guardrails:

```sql
create or alter table meta.log.t_dlthub_fix_attempt (
      attempt_id        varchar         not null
    , created_at        timestamp_ltz   not null default current_timestamp()
    , kind              varchar         not null    -- retry, fix
    , job_ref           varchar         not null
    , source_run_id     varchar         not null    -- the failed run that was inspected
    , attempt_run_id    varchar         null        -- the run the attempt started
    , status            varchar         not null    -- starting, running, succeeded, failed, skipped, error, dry_ok, dry_failed, rejected, rolled_back, rollback_failed, duplicate
    , classification    varchar         null
    , confidence        varchar         null
    , finished_at       timestamp_ltz   null
    , details           variant         null
)
comment = 'JOB_INSPECTOR attempts to repair a failed dltHub run: one row per attempt'
;
```

And a second, tiny table: the kill switch (always build in a kill switch! 😎). A missing row counts as enabled, so tools/actions run unless a row for that action explicitly disables it (`enabled = false`):

```sql
create or alter table meta.log.t_dlthub_fix_switch (
      action            varchar         not null    -- retry, fix
    , enabled           boolean         not null
    , changed_at        timestamp_ltz   null
    , changed_by        varchar         null
)
comment = 'Kill switch of the JOB_INSPECTOR actions: one row per action, enabled = false refuses new attempts'
;

insert into meta.log.t_dlthub_fix_switch (action, enabled, changed_at, changed_by)
select s.action, true, current_timestamp(), current_user()
from (select 'retry' as action union all select 'fix') s
where not exists (select 1 from meta.log.t_dlthub_fix_switch w where w.action = s.action)
;
```

The agent sees the attempt history log through its `Check_Attempts` tool, which it calls in the first round alongside `Get_Run`. That call performs an important job: a retry that fails fires `job.fail` again, which wakes the inspector again, which... yeah, exactly. `is_attempt_run = true` tells the agent that an earlier attempt started this run, and the agent instructions (the prompt) say to stop right there with status `skipped`. Additionally, the procedures refuse that case anyway (see below)... the early stop only saves tokens and a pointless Teams card.

Show p\_dlthub\_attempt\_check (the Check\_Attempts tool)▾

```sql
create or replace procedure meta.cortex.p_dlthub_attempt_check(run_id varchar)
copy grants
returns varchar
language sql
comment = 'JOB_INSPECTOR tool: was this run started by an attempt, or already attempted?'
execute as owner
as
$$
declare
    result varchar;
begin
    select
        to_json(object_construct(
              'run_id', :run_id
            , 'is_attempt_run', coalesce(boolor_agg(a.attempt_run_id = :run_id), false)
            , 'already_attempted', coalesce(boolor_agg(a.source_run_id = :run_id), false)
            , 'attempts', array_agg(object_construct(
                  'kind', a.kind
                , 'status', a.status
                , 'source_run_id', a.source_run_id
                , 'attempt_run_id', a.attempt_run_id
              )) within group (order by a.created_at)
        ))
    into :result
    from meta.log.t_dlthub_fix_attempt a
    where   a.attempt_run_id = :run_id
        or  a.source_run_id = :run_id;
    return result;
end;
$$
;

alter procedure if exists meta.cortex.p_dlthub_attempt_check(varchar) set tag meta.tag.triggered_by = 'cortex_agent';
```

## Retry: the easy path

`Retry_Job` is a Snowflake procedure (`meta.cortex.p_dlthub_retry`). It fires the failed job once, using the same dltHub API call many of my Snowflake tasks already make when they need data in Snowflake refreshed (I said it somewhere else already and I'll say it again: Who needs an orchestrator when you have task-chains and cron? 😜). It also spins up a short-lived SPCS container to watch that run and report back to Teams: a one-liner when it passed or an alert when it failed again.

The procedure resolves the failed run in dltHub and reads the ref there, which means a confused model can't accidentally retry the wrong job. The top of `main()` is a list of refusals:

```python
def main(session, failed_run_id, classification, confidence):
    if not switched_on(session, "retry"):
        return reply("refused", reason="retries are switched off (meta.log.t_dlthub_fix_switch)")
    if classification != "transient":
        return reply("refused", reason=f"only transient failures are retried, classification is {classification}")
    if confidence == "low":
        return reply("refused", reason="confidence is low")
    if not second_opinion_agrees(session, failed_run_id, classification):
        return reply("refused", reason="no agreeing second opinion for this run and classification, call Second_Opinion first")

    try:
        key = _snowflake.get_generic_secret_string("api_key")
        workspace = dlthub_sdk.connect(token=key, base_url="https://api.dlthub.com").workspaces.get(id=WORKSPACE_ID)
        run = workspace.job_runs.get(id=failed_run_id).to_dict()
    except Exception as e:
        return reply("error", reason=f"could not read the run from dltHub: {type(e).__name__}: {str(e)[:200]}")

    run_status = str(value(run.get("status")))
    job_ref = run.get("job_ref")
    if run_status != "failed" or not job_ref:
        return reply("refused", reason=f"run status is {run_status}, only failed runs are retried")
```

Then the attempt gets claimed. A duplicate check and a 3-in-24h cap sit inside the very `insert` into the attempt log table `t_dlthub_fix_attempt`. If the insert inserts nothing, the procedure stops:

```python
attempt_id = str(uuid.uuid4())
claimed = session.sql(
    f"""
    insert into {TABLE} (attempt_id, kind, job_ref, source_run_id, status, classification, confidence)
    select ?, 'retry', ?, ?, 'starting', ?, ?
    from (select 1 as one) d
    where   not exists (
                select 1 from {TABLE} a
                where a.source_run_id = ? or a.attempt_run_id = ?
            )
        and (
                select count(*) from {TABLE} a
                where   a.job_ref = ?
                    and a.status <> 'duplicate'
                    and a.created_at > dateadd(hour, -24, current_timestamp())
            ) < ?
    """,
    params=[attempt_id, job_ref, failed_run_id, classification, confidence,
            failed_run_id, failed_run_id, job_ref, ATTEMPTS_PER_JOB_AND_DAY],
).collect()[0][0]

if claimed and not confirm_claim(session, attempt_id, failed_run_id, job_ref):
    return reply("refused", reason="a parallel call for this run or job won the claim")
if claimed == 0:
```

`confirm_claim` settles races afterwards, and `is_latest_failure` keeps the agent from naming an old run:

```python
def is_latest_failure(workspace, job_ref, run_id):
    """Only the job's latest failed run may be acted on, not any run of the workspace."""
    runs = [r.to_dict() for r in workspace.jobs.get_by_ref(ref=job_ref).runs.list(limit=30)]
    failed = [r for r in runs if str(value(r.get("status"))) == "failed" and r.get("started_at")]
    return not failed or str(max(failed, key=lambda r: str(r["started_at"]))["id"]) == run_id

def confirm_claim(session, attempt_id, source_run_id, job_ref):
    """insert ... where not exists is not atomic across sessions: two parallel calls can both pass.
    The earliest row wins, a later one steps back; the same goes for the cap per job."""
    first = session.sql(
        f"select attempt_id from {TABLE} where source_run_id = ? and status <> 'duplicate' order by created_at, attempt_id limit 1",
        params=[source_run_id],
    ).collect()[0][0]
    earlier = session.sql(
        f"""
        select count(*) from {TABLE} a join {TABLE} m on m.attempt_id = ?
        where   a.job_ref = ? and a.status <> 'duplicate' and a.created_at > dateadd(hour, -24, current_timestamp())
            and (a.created_at < m.created_at or (a.created_at = m.created_at and a.attempt_id < m.attempt_id))
        """,
        params=[attempt_id, job_ref],
    ).collect()[0][0]
    if first == attempt_id and earlier < ATTEMPTS_PER_JOB_AND_DAY:
        return True
    session.sql(
        f"update {TABLE} set status = 'duplicate', finished_at = current_timestamp() where attempt_id = ?", params=[attempt_id]
    ).collect()
    return False

```

The `insert ... where not exists` is not atomic across sessions: two parallel calls can both pass the check. So after the insert, `confirm_claim` looks at the table again. The earliest row for that run wins, and the same goes for the 3-in-24h cap. A later row steps back and gets status `duplicate`, which the cap count ignores.

After the claim, the procedure triggers the job, stores the new run id on the row and calls `meta.load.p_dlthub_start_fixer(attempt_id, 'watch')`, which starts the same container image the fixer uses (more on that one in a second) in watch mode: it only waits for the run and reports. Nothing polls, no task runs every five minutes to check on retries.

Show the full p\_dlthub\_retry▾

```sql
create or replace procedure meta.cortex.p_dlthub_retry(
    failed_run_id       varchar
    , classification    varchar
    , confidence        varchar
)
copy grants
returns varchar
language python
runtime_version = '3.13'
artifact_repository = snowflake.snowpark.pypi_shared_repository
packages = ('snowflake-snowpark-python', 'dlthub-client')
handler = 'main'
comment = 'JOB_INSPECTOR tool: retry a transient failed dltHub run once; guarded, logged in meta.log.t_dlthub_fix_attempt'
external_access_integrations = (i_dlthub)
secrets = ('api_key' = meta.integration.se_dlthub_api_key)
execute as owner
as
$$
import json
import uuid
import _snowflake
import dlthub_sdk

WORKSPACE_ID = "4f3da7b4-ae6a-43db-8ce6-d909aba0b47f"
ATTEMPTS_PER_JOB_AND_DAY = 3
MIN_PROBABILITY = 0.6  # a second opinion below this agrees by luck
TABLE = "meta.log.t_dlthub_fix_attempt"

def value(x):
    return getattr(x, "value", x)

def reply(status, **fields):
    return json.dumps({"status": status, **fields}, default=str)

def switched_on(session, action):
    rows = session.sql("select enabled from meta.log.t_dlthub_fix_switch where action = ?", params=[action]).collect()
    return bool(rows[0][0]) if rows else True

def second_opinion_agrees(session, run_id, classification):
    """The latest second opinion of the last 30 minutes must agree and be sure enough."""
    rows = session.sql(
        "select agrees and coalesce(jev_probability, 0) >= ? from meta.log.t_dlthub_second_opinion "
        "where run_id = ? and agent_classification = ? and created_at > dateadd(minute, -30, current_timestamp()) "
        "order by created_at desc limit 1",
        params=[MIN_PROBABILITY, run_id, classification],
    ).collect()
    return bool(rows and rows[0][0])

def is_latest_failure(workspace, job_ref, run_id):
    """The agent passes the run id: only the job's latest failed run may be acted on, not any run of the workspace."""
    runs = [r.to_dict() for r in workspace.jobs.get_by_ref(ref=job_ref).runs.list(limit=30)]
    failed = [r for r in runs if str(value(r.get("status"))) == "failed" and r.get("started_at")]
    return not failed or str(max(failed, key=lambda r: str(r["started_at"]))["id"]) == run_id

def confirm_claim(session, attempt_id, source_run_id, job_ref):
    """insert ... where not exists is not atomic across sessions: two parallel calls can both pass.
    The earliest row wins, a later one steps back; the same goes for the cap per job."""
    first = session.sql(
        f"select attempt_id from {TABLE} where source_run_id = ? and status <> 'duplicate' order by created_at, attempt_id limit 1",
        params=[source_run_id],
    ).collect()[0][0]
    earlier = session.sql(
        f"""
        select count(*) from {TABLE} a join {TABLE} m on m.attempt_id = ?
        where   a.job_ref = ? and a.status <> 'duplicate' and a.created_at > dateadd(hour, -24, current_timestamp())
            and (a.created_at < m.created_at or (a.created_at = m.created_at and a.attempt_id < m.attempt_id))
        """,
        params=[attempt_id, job_ref],
    ).collect()[0][0]
    if first == attempt_id and earlier < ATTEMPTS_PER_JOB_AND_DAY:
        return True
    session.sql(
        f"update {TABLE} set status = 'duplicate', finished_at = current_timestamp() where attempt_id = ?", params=[attempt_id]
    ).collect()
    return False

def alert(session):
    """A failure that no container reports (it never started) still reaches Teams."""
    try:
        session.sql("call meta.load.p_monitor('dlthub_fix_failed')").collect()
    except Exception:
        pass

def main(session, failed_run_id, classification, confidence):
    if not switched_on(session, "retry"):
        return reply("refused", reason="retries are switched off (meta.log.t_dlthub_fix_switch)")
    if classification != "transient":
        return reply("refused", reason=f"only transient failures are retried, classification is {classification}")
    if confidence == "low":
        return reply("refused", reason="confidence is low")
    if not second_opinion_agrees(session, failed_run_id, classification):
        return reply("refused", reason="no agreeing second opinion for this run and classification, call Second_Opinion first")

    try:
        key = _snowflake.get_generic_secret_string("api_key")
        workspace = dlthub_sdk.connect(token=key, base_url="https://api.dlthub.com").workspaces.get(id=WORKSPACE_ID)
        run = workspace.job_runs.get(id=failed_run_id).to_dict()
    except Exception as e:
        return reply("error", reason=f"could not read the run from dltHub: {type(e).__name__}: {str(e)[:200]}")

    run_status = str(value(run.get("status")))
    job_ref = run.get("job_ref")
    if run_status != "failed" or not job_ref:
        return reply("refused", reason=f"run status is {run_status}, only failed runs are retried")
    try:
        if not is_latest_failure(workspace, job_ref, failed_run_id):
            return reply("refused", reason="a newer run of this job failed, only the latest failure is retried")
    except Exception as e:
        return reply("error", reason=f"could not list the runs of the job: {type(e).__name__}: {str(e)[:200]}")

    attempt_id = str(uuid.uuid4())
    claimed = session.sql(
        f"""
        insert into {TABLE} (attempt_id, kind, job_ref, source_run_id, status, classification, confidence)
        select ?, 'retry', ?, ?, 'starting', ?, ?
        from (select 1 as one) d
        where   not exists (
                    select 1 from {TABLE} a
                    where a.source_run_id = ? or a.attempt_run_id = ?
                )
            and (
                    select count(*) from {TABLE} a
                    where   a.job_ref = ?
                        and a.status <> 'duplicate'
                        and a.created_at > dateadd(hour, -24, current_timestamp())
                ) < ?
        """,
        params=[attempt_id, job_ref, failed_run_id, classification, confidence,
                failed_run_id, failed_run_id, job_ref, ATTEMPTS_PER_JOB_AND_DAY],
    ).collect()[0][0]

    if claimed and not confirm_claim(session, attempt_id, failed_run_id, job_ref):
        return reply("refused", reason="a parallel call for this run or job won the claim")
    if claimed == 0:
        started_by_attempt = session.sql(
            f"select count(*) from {TABLE} where attempt_run_id = ?", params=[failed_run_id]
        ).collect()[0][0]
        already = session.sql(
            f"select count(*) from {TABLE} where source_run_id = ?", params=[failed_run_id]
        ).collect()[0][0]
        if started_by_attempt:
            reason = "this run was started by an earlier attempt, it is never retried again"
        elif already:
            reason = "this run was already attempted once"
        else:
            reason = f"{ATTEMPTS_PER_JOB_AND_DAY} attempts for this job in 24 hours already"
        return reply("refused", reason=reason)

    def close(status, **details):
        session.sql(
            f"update {TABLE} set status = ?, finished_at = current_timestamp(), details = parse_json(?) where attempt_id = ?",
            params=[status, json.dumps(details, default=str), attempt_id],
        ).collect()

    try:
        result = workspace.jobs.trigger(refs=[job_ref], dry_run=False)[0]
    except Exception as e:
        close("error", reason=f"trigger failed: {type(e).__name__}: {str(e)[:200]}")
        alert(session)
        return reply("error", attempt_id=attempt_id, reason="trigger of the job failed")

    if not result.started:
        why = str(value(result.status))
        close("error", reason=f"the job was not started: {why}")  # error, so the alert view lists it
        alert(session)
        return reply("skipped", attempt_id=attempt_id, job_ref=job_ref, reason=why)

    session.sql(
        f"update {TABLE} set status = 'running', attempt_run_id = ? where attempt_id = ?",
        params=[result.job_run_id, attempt_id],
    ).collect()
    try:
        session.sql("call meta.load.p_dlthub_start_fixer(?, 'watch')", params=[attempt_id]).collect()
    except Exception as e:
        session.sql(
            f"update {TABLE} set status = 'error', finished_at = current_timestamp(), "
            f"details = object_construct('reason', ?) where attempt_id = ?",
            params=[f"the retry runs but nothing watches it: {str(e)[:200]}", attempt_id],
        ).collect()
        alert(session)
        return reply("started", attempt_id=attempt_id, job_ref=job_ref, attempt_run_id=result.job_run_id,
                     warning=f"the retry runs but nothing watches it: {str(e)[:200]}")
    return reply("started", attempt_id=attempt_id, job_ref=job_ref, attempt_run_id=result.job_run_id)
$$
;

alter procedure if exists meta.cortex.p_dlthub_retry(varchar, varchar, varchar) set tag meta.tag.triggered_by = 'cortex_agent';
```

For a test I stood up a canary job that fetches a URL out of a workspace variable:

```python
"""Canary job: manual trigger only, loads and writes nothing."""
import os

import requests

def main():
    url = os.environ.get("RETRY_CANARY__URL", "https://httpbin.org/status/200")
    response = requests.get(url, timeout=20)
    response.raise_for_status()
    print(f"retry canary ok: HTTP {response.status_code}")
```

Aim the variable at `https://httpbin.org/status/503` and the job dies the way a flaky upstream dies; flip it back to `.../status/200` within a minute and the retry can succeed. The agent called the 503 `transient`, the second opinion (Jev, we'll get to that, too) agreed and the retry ran, a note hit Teams.

![](https://storage.ghost.io/c/7d/94/7d942fe1-7868-4a1a-b2c9-4eb415b1a546/content/images/2026/11/2026-11-05-22_09_57-Chat-_-System-Monitoring-_-Data-Pipelines----------------_-Pro-Juventute-_-martin.seifer.png)

When a Snowflake task triggers such a failing dltHub job and waits on it, the task fails, even though the retry might succeed. The data does land (delayed), but the task's downstream steps sit idle until the next cycle. I haven't found a clean fix for that, so if anyone has an idea...

## Fix: the scary path

`Apply_Fix` is the part where the agent edits code. And where it adds the most value: If something fails in the middle of the night, the only information I need is "it's fixed and running again - if you want to look, here is what happened..."

The agent in this case doesn't write a patch file. It rather hands over a list of exact `old` and `new` strings for each file that needs fixing, and every one of them has to match exactly once. Why not a unified diff? I.m.h.o. models are worse at line numbers than at copying text, and "occurs exactly once" is a check a deterministic script can make.

The procedure (`meta.load.p_dlthub_apply_fix`) runs the same refusals and the same claiming insert as the retry, plus a sanity check on the edits:

```python
def main(session, failed_run_id, classification, confidence, edits, rationale):
    if not switched_on(session, "fix"):
        return reply("refused", reason="fixes are switched off (meta.log.t_dlthub_fix_switch)")
    if classification not in ("code", "config"):
        return reply("refused", reason=f"only code or config failures are fixed, classification is {classification}")
    if confidence != "high":
        return reply("refused", reason=f"fixes need high confidence, confidence is {confidence}")
    if not second_opinion_agrees(session, failed_run_id, classification):
        return reply("refused", reason="no agreeing second opinion for this run and classification, call Second_Opinion first")
    try:
        parsed = json.loads(edits)
        assert isinstance(parsed, list) and parsed
        assert all(isinstance(e, dict) and all(isinstance(e.get(k), str) for k in ("path", "old", "new")) for e in parsed)
        assert all(e["old"] and e["old"] != e["new"] for e in parsed)
    except Exception:
        return reply("refused", reason="edits must be a non-empty json array of {path, old, new} with old non-empty and different from new")

    try:
```

The edits and the rationale go into `details` on the attempt row in `t_dlthub_fix_attempt`. Then `p_dlthub_start_fixer` launches a job service on a SPCS compute pool. The GitHub token and the dltHub key are accessible via Snowflake secrets (like they already were to the read-only tools), neither the agent nor the procedure ever actually see them:

```python
def main(session, attempt_id, mode):
    if mode not in ("dry", "apply", "watch") or not re.fullmatch(r"[0-9a-f-]{36}", attempt_id):
        raise ValueError("bad attempt id or mode")
    job_ref = session.sql(f"select job_ref from {TABLE} where attempt_id = ?", params=[attempt_id]).collect()[0][0]
    if not re.fullmatch(r"jobs\.[A-Za-z0-9_.]+", job_ref):
        raise ValueError("bad job ref")

    tidy_up(session)
    service = f"{SERVICE_SCHEMA}.{'watch' if mode == 'watch' else 'fix'}_{attempt_id[:8]}"
    spec = f"""
spec:
  containers:
  - name: fixer
    image: {IMAGE}
    env:
      ATTEMPT_ID: "{attempt_id}"
      MODE: "{mode}"
      JOB_REF: "{job_ref}"
      WORKSPACE_ID: "{WORKSPACE_ID}"
    secrets:
    - snowflakeSecret: meta.integration.se_github
      secretKeyRef: password
      envVarName: GH_PAT
    - snowflakeSecret: meta.integration.se_dlthub_api_key
      secretKeyRef: secret_string
      envVarName: RUNTIME__API_KEY
"""
    # single-quoted literal: a nested dollar-quote would end this procedure's own body
    spec_sql = spec.replace("\\", "\\\\").replace("'", "\\'")
    session.sql(
        f"execute job service in compute pool cp_dlt name = {service} async = true "
        f"external_access_integrations = (i_github_egress, i_dlthub) from specification '{spec_sql}'"
    ).collect()
    return service
```

The container runs an image I built out of a small `fixer.py`. Dependencies are baked in, because the job service only gets egress to GitHub and dltHub, not to PyPI:

```dockerfile
# Container for meta.cortex.p_dlthub_apply_fix: applies a JOB_INSPECTOR fix to the dlthub repo.
# Build context needs pyproject.toml and uv.lock copied from the dlthub repo (they pin the
# dependencies, which are baked in so the job needs no PyPI egress). Rebuild when uv.lock changes.
FROM python:3.13-slim
RUN apt-get update && apt-get install -y --no-install-recommends git ca-certificates \
    && rm -rf /var/lib/apt/lists/*
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
ENV UV_PROJECT_ENVIRONMENT=/opt/venv UV_LINK_MODE=copy PYTHONUTF8=1
WORKDIR /deps
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-install-project --no-dev
COPY fixer.py /app/fixer.py
CMD ["/opt/venv/bin/python", "/app/fixer.py"]

```

Why a container? Well, Cortex agents have a code-execution sandbox, but that can't connect to the outside world (yet). They can also use a curated set of tools, one of them allows the agent to connect to GitHub. But dltHub is not on that curated list (yet 😜). A SPCS job service combines both: deterministic logic + external access.

Inside that container, in mode `apply`:

1. clone `master` (or `main` or however your branch is named) of the dltHub repo,
2. apply the edits,
3. run the gates: `py_compile` over each touched file, `dlthub deploy --dry-run`, and a push preflight (testing if the token can push at all),
4. commit only the edited files as `JOB_INSPECTOR`, push, run `dlthub deploy`,
5. trigger the failed job again and wait for that run.

A run that completes posts a success note. A run that fails reverts the commit, redeploys the previous state (if the push itself timed out, the rollback first checks whether the commit reached master and only reverts if it did), and posts an alert card. In that case I'm expected to show up 🙄

![](https://storage.ghost.io/c/7d/94/7d942fe1-7868-4a1a-b2c9-4eb415b1a546/content/images/2026/11/2026-11-05-22_08_11-Chat-_-System-Monitoring-_-Data-Pipelines----------------_-Pro-Juventute-_-martin.seifer.png)

```python
def run_fix(cur, attempt_id, mode, result, details, source_run_id):
    result["base_sha"] = clone()
    need("git", "push", "--dry-run", "origin", f"HEAD:{BRANCH}", what="push preflight (no write access?)", git_auth=True)
    result["gates"]["push_access"] = True
    apply_edits(details["edits"], result)
    need(*DLTHUB, "workspace", "connect", os.environ["WORKSPACE_ID"], what="workspace connect")
    need(*DLTHUB, "deploy", "--dry-run", what="deploy --dry-run")
    result["gates"]["deploy_dry_run"] = True
    if mode == "dry":
        save(cur, attempt_id, "dry_ok", result, True)
        return

    job_ref = os.environ["JOB_REF"]
    message = (
        f"Fix {job_ref.rsplit('.', 1)[-1]}: {details.get('rationale', '')[:120]}\n\n"
        f"Attempt {attempt_id} for run {source_run_id}, applied by the Cortex JOB_INSPECTOR."
    )
    # the kill switch is read once more: an attempt started before it was flipped stops here
    cur.execute("select enabled from meta.log.t_dlthub_fix_switch where action = 'fix' order by changed_at desc nulls last limit 1")
    row = cur.fetchone()
    if row and not row[0]:
        raise GateFailed("fixes were switched off before the push")
    # only the edited files are committed, not whatever the gates touched (lock files, config)
    need("git", "add", "--", *sorted({e["path"] for e in details["edits"]}), what="git add")
    need(*GIT, "commit", "-m", message, what="commit")
    result["fix_sha"] = need("git", "rev-parse", "HEAD", what="rev-parse")
    try:
        need("git", "push", "origin", f"HEAD:{BRANCH}", what="push", git_auth=True)
    except Exception:
        # a timed out push may still have reached master: undo checks and reverts only if it did
        result["push_uncertain"] = True
        roll_back(cur, attempt_id, result)
        raise
    result["pushed"] = True
    try:
        deploy()
        result["deployed"] = True
        run_id = trigger(job_ref)
        save(cur, attempt_id, "running", result, False, run_id)
        run_status = wait_for_run(run_id)
    except Exception as e:
        result["error"] = f"after the push: {str(e)[:400]}"
        roll_back(cur, attempt_id, result)
        notify(cur)
        return
    result["run_status"] = run_status
    if run_status == "completed":
        save(cur, attempt_id, "succeeded", result, True)
        notify(cur, "dlthub_fix_succeeded")
    elif run_status == "timeout":
        result["error"] = f"the re-run did not end within {WAIT_SECONDS // 3600} hours, the fix stays deployed"
        save(cur, attempt_id, "error", result, True)
        notify(cur)
    else:
        result["error"] = f"the re-run ended {run_status}"
        roll_back(cur, attempt_id, result)
        notify(cur)
```

The container accepts edits only to the failed job's own file (.py) and my shared `_*_common.py` helpers in the repo root, so a manipulated agent can't patch another job's code. On top of that, a few paths stay off limits whatever the agent asks: `__deployment__.py` (a bad edit there can archive jobs), hidden folders, `_archive/`, the (dltHub) agent prompts, anything that isn't `.py`, and any patch above 60 changed lines. Those limits sit as plain constants at the top of the script. The edit loop checks them before and after:

```python
WORK = "/work"
WAIT_SECONDS = 3 * 3600
MAX_CHANGED_LINES = 60
# never patchable: manifest (a bad edit can archive jobs), hidden and archived folders, agent prompts
```

```python
def check_path(path):
    p = PurePosixPath(path)
    if p.is_absolute() or ".." in p.parts or p.suffix != ".py":
        raise GateFailed(f"path not allowed: {path}")
    if path in DENIED_FILES or path.startswith(DENIED_PREFIXES):
        raise GateFailed(f"path is protected: {path}")
    # only the failed job's own file and the shared _*_common.py helpers: a manipulated agent
    # cannot patch another job's code
    job = os.environ.get("JOB_REF", "").rsplit(".", 1)[-1]
    allowed = len(p.parts) == 1 and (path == f"{job}.py" or (path.startswith("_") and path.endswith("_common.py")))
    if not allowed:
        raise GateFailed(f"path is not part of the failed job ({job}): {path}")

```

```python
def apply_edits(edits, result):
    diff = []
    for e in edits:
        check_path(e["path"])
        target = os.path.join(WORK, e["path"])
        if not os.path.isfile(target):
            raise GateFailed(f"file not found: {e['path']}")
        text = open(target, encoding="utf-8", newline="").read()
        if text.count(e["old"]) != 1:
            raise GateFailed(f"old text occurs {text.count(e['old'])} times in {e['path']}, must be exactly once")
        new_text = text.replace(e["old"], e["new"])
        diff += list(difflib.unified_diff(text.splitlines(), new_text.splitlines(), e["path"], e["path"], lineterm="", n=2))
        open(target, "w", encoding="utf-8", newline="").write(new_text)
    changed = sum(1 for line in diff if line[:1] in "+-" and not line.startswith(("+++", "---")))
    result["diff"] = "\n".join(diff)[:6000]
    result["gates"]["edits_applied"] = True
    if changed > MAX_CHANGED_LINES:
        raise GateFailed(f"{changed} changed lines, limit is {MAX_CHANGED_LINES}")
    result["gates"]["size"] = changed
    for e in edits:
        py_compile.compile(os.path.join(WORK, e["path"]), doraise=True)
    result["gates"]["py_compile"] = True
```

A mode `dry` exists, too: clone, edits, gates, and then stop without pushing anything. It runs through `p_dlthub_start_fixer(id, 'dry')` on an attempt row, which was useful while building and testing that thing, but now is just lingering around.

Show the full fixer.py▾

```python
"""Applies a JOB_INSPECTOR fix to the dlthub repo, or watches a retry, and waits for the result.

MODE (env) selects what runs; the attempt row in meta.log.t_dlthub_fix_attempt carries the input
(details:edits, details:rationale) and receives the outcome (status, details:result).

  dry    clone, push preflight, apply edits, gates. Pushes and deploys nothing.
  apply  dry, then commit, push to master, deploy, trigger the failed job and wait for that run.
         Completed: succeeded. Failed: revert the commit, redeploy, rolled_back. Still running
         after WAIT_SECONDS: error, the fix stays deployed.
  watch  wait for the run a retry started (attempt_run_id) and record the outcome.

Every outcome except dry ends with a call to meta.load.p_monitor: a failure posts the alert card,
a success a short note (what was wrong, what was done).
A failure after the push (deploy or trigger) reverts the commit as well. Prints no secrets.
"""
import base64
import difflib
import json
import os
import py_compile
import subprocess
import time
from pathlib import PurePosixPath

import dlthub_sdk
import snowflake.connector

REPO = "projuventute/dlthub"
BRANCH = "master"
WORK = "/work"
WAIT_SECONDS = 3 * 3600
MAX_CHANGED_LINES = 60
# never patchable: manifest (a bad edit can archive jobs), hidden and archived folders, agent prompts
DENIED_FILES = {"__deployment__.py"}
DENIED_PREFIXES = (".", "_archive/", "agents/")
GIT = ["git", "-c", "user.name=JOB_INSPECTOR", "-c", "user.email=job-inspector@users.noreply.github.com"]
DLTHUB = ["uv", "run", "--frozen", "--no-sync", "dlthub", "--non-interactive"]

class GateFailed(Exception):
    pass

def _git_header():
    """Auth header for git network calls: the token never sits in the remote url or in the environment of the gates."""
    pat = os.environ.get("GH_PAT", "")
    return base64.b64encode(f"x-access-token:{pat}".encode()).decode() if pat else ""

def sh(*cmd, cwd=WORK, timeout=900, git_auth=False):
    secrets = [x for x in (os.environ.get("GH_PAT", ""), _git_header()) if x]
    if git_auth:
        cmd = (cmd[0], "-c", f"http.extraheader=Authorization: Basic {_git_header()}", *cmd[1:])
    # gates run edited code (deploy --dry-run imports the workspace): they get no GitHub token
    env = {k: v for k, v in os.environ.items() if k != "GH_PAT"}
    done = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout, env=env)
    out = done.stdout + done.stderr
    for secret in secrets:
        out = out.replace(secret, "***")
    return done.returncode, out

def need(*cmd, what, cwd=WORK, git_auth=False):
    code, out = sh(*cmd, cwd=cwd, git_auth=git_auth)
    if code:
        raise GateFailed(f"{what} failed: {out[-500:]}")
    return out.strip()

def check_path(path):
    p = PurePosixPath(path)
    if p.is_absolute() or ".." in p.parts or p.suffix != ".py":
        raise GateFailed(f"path not allowed: {path}")
    if path in DENIED_FILES or path.startswith(DENIED_PREFIXES):
        raise GateFailed(f"path is protected: {path}")
    # only the failed job's own file and the shared _*_common.py helpers: a manipulated agent
    # cannot patch another job's code
    job = os.environ.get("JOB_REF", "").rsplit(".", 1)[-1]
    allowed = len(p.parts) == 1 and (path == f"{job}.py" or (path.startswith("_") and path.endswith("_common.py")))
    if not allowed:
        raise GateFailed(f"path is not part of the failed job ({job}): {path}")

def connect():
    return snowflake.connector.connect(
        host=os.environ["SNOWFLAKE_HOST"],
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        authenticator="oauth",
        token=open("/snowflake/session/token").read(),
        warehouse="XS_G1",
        database="META",
        schema="LOG",
    )

def save(cur, attempt_id, status, result, finished, run_id=None):
    cur.execute(
        "update meta.log.t_dlthub_fix_attempt set status = %s, "
        "finished_at = iff(%s, current_timestamp(), null), "
        "attempt_run_id = coalesce(%s, attempt_run_id), "
        "details = object_insert(coalesce(details, object_construct()), 'result', parse_json(%s), true) "
        "where attempt_id = %s",
        (status, finished, run_id, json.dumps(result), attempt_id),
    )

def notify(cur, monitor="dlthub_fix_failed"):
    """Posts the Teams card or note for the attempts that ended (meta.load.p_monitor, watermarked)."""
    try:
        cur.execute("call meta.load.p_monitor(%s)", (monitor,))
        print(f"{monitor}:", cur.fetchone()[0])
    except Exception as e:
        print(f"{monitor} not sent: {type(e).__name__}: {str(e)[:200]}")

def clone():
    url = f"https://github.com/{REPO}.git"
    need("git", "clone", "--depth", "20", "--branch", BRANCH, url, WORK, what="clone", cwd="/", git_auth=True)
    return need("git", "rev-parse", "HEAD", what="rev-parse")

def deploy():
    need(*DLTHUB, "deploy", what="deploy")

def undo(fix_sha):
    """Revert the fix commit on master and redeploy."""
    need(*GIT, "fetch", "origin", BRANCH, what="fetch", git_auth=True)
    need(*GIT, "reset", "--hard", f"origin/{BRANCH}", what="reset")
    code, _ = sh("git", "merge-base", "--is-ancestor", fix_sha, "HEAD")
    if code:
        return False  # the fix never reached master (a push that timed out): nothing to revert
    need(*GIT, "revert", "--no-edit", fix_sha, what="revert")
    need("git", "push", "origin", f"HEAD:{BRANCH}", what="push of the revert", git_auth=True)
    deploy()
    return True

def workspace():
    runtime = dlthub_sdk.connect(token=os.environ["RUNTIME__API_KEY"], base_url="https://api.dlthub.com")
    return runtime.workspaces.get(id=os.environ["WORKSPACE_ID"])

def trigger(job_ref):
    res = workspace().jobs.trigger(refs=[job_ref], dry_run=False)[0]
    if not res.started:
        raise GateFailed(f"the job was not triggered: {getattr(res.status, 'value', res.status)}")
    return res.job_run_id

def wait_for_run(run_id):
    """Final status of the run, or 'timeout'."""
    deadline = time.time() + WAIT_SECONDS
    errors = 0
    while True:
        try:
            run = workspace().job_runs.get(id=run_id)
            return str(getattr(run.wait(timeout=max(1, int(deadline - time.time()))).status, "value", None))
        except dlthub_sdk.errors.WaitTimeout:
            return "timeout"
        except Exception:
            # a network blip is not a failed run: only repeated errors count
            errors += 1
            if errors >= 5 or time.time() > deadline:
                raise
            time.sleep(30)

def apply_edits(edits, result):
    diff = []
    for e in edits:
        check_path(e["path"])
        target = os.path.join(WORK, e["path"])
        if not os.path.isfile(target):
            raise GateFailed(f"file not found: {e['path']}")
        text = open(target, encoding="utf-8", newline="").read()
        if text.count(e["old"]) != 1:
            raise GateFailed(f"old text occurs {text.count(e['old'])} times in {e['path']}, must be exactly once")
        new_text = text.replace(e["old"], e["new"])
        diff += list(difflib.unified_diff(text.splitlines(), new_text.splitlines(), e["path"], e["path"], lineterm="", n=2))
        open(target, "w", encoding="utf-8", newline="").write(new_text)
    changed = sum(1 for line in diff if line[:1] in "+-" and not line.startswith(("+++", "---")))
    result["diff"] = "\n".join(diff)[:6000]
    result["gates"]["edits_applied"] = True
    if changed > MAX_CHANGED_LINES:
        raise GateFailed(f"{changed} changed lines, limit is {MAX_CHANGED_LINES}")
    result["gates"]["size"] = changed
    for e in edits:
        py_compile.compile(os.path.join(WORK, e["path"]), doraise=True)
    result["gates"]["py_compile"] = True

def roll_back(cur, attempt_id, result):
    try:
        reverted = undo(result["fix_sha"])
        result["rolled_back"] = reverted
        save(cur, attempt_id, "rolled_back" if reverted else "rejected", result, True)
    except Exception as e:
        result["rollback_error"] = str(e)[:400]
        save(cur, attempt_id, "rollback_failed", result, True)

def run_fix(cur, attempt_id, mode, result, details, source_run_id):
    result["base_sha"] = clone()
    need("git", "push", "--dry-run", "origin", f"HEAD:{BRANCH}", what="push preflight (no write access?)", git_auth=True)
    result["gates"]["push_access"] = True
    apply_edits(details["edits"], result)
    need(*DLTHUB, "workspace", "connect", os.environ["WORKSPACE_ID"], what="workspace connect")
    need(*DLTHUB, "deploy", "--dry-run", what="deploy --dry-run")
    result["gates"]["deploy_dry_run"] = True
    if mode == "dry":
        save(cur, attempt_id, "dry_ok", result, True)
        return

    job_ref = os.environ["JOB_REF"]
    message = (
        f"Fix {job_ref.rsplit('.', 1)[-1]}: {details.get('rationale', '')[:120]}\n\n"
        f"Attempt {attempt_id} for run {source_run_id}, applied by the Cortex JOB_INSPECTOR."
    )
    # the kill switch is read once more: an attempt started before it was flipped stops here
    cur.execute("select enabled from meta.log.t_dlthub_fix_switch where action = 'fix' order by changed_at desc nulls last limit 1")
    row = cur.fetchone()
    if row and not row[0]:
        raise GateFailed("fixes were switched off before the push")
    # only the edited files are committed, not whatever the gates touched (lock files, config)
    need("git", "add", "--", *sorted({e["path"] for e in details["edits"]}), what="git add")
    need(*GIT, "commit", "-m", message, what="commit")
    result["fix_sha"] = need("git", "rev-parse", "HEAD", what="rev-parse")
    try:
        need("git", "push", "origin", f"HEAD:{BRANCH}", what="push", git_auth=True)
    except Exception:
        # a timed out push may still have reached master: undo checks and reverts only if it did
        result["push_uncertain"] = True
        roll_back(cur, attempt_id, result)
        raise
    result["pushed"] = True
    try:
        deploy()
        result["deployed"] = True
        run_id = trigger(job_ref)
        save(cur, attempt_id, "running", result, False, run_id)
        run_status = wait_for_run(run_id)
    except Exception as e:
        result["error"] = f"after the push: {str(e)[:400]}"
        roll_back(cur, attempt_id, result)
        notify(cur)
        return
    result["run_status"] = run_status
    if run_status == "completed":
        save(cur, attempt_id, "succeeded", result, True)
        notify(cur, "dlthub_fix_succeeded")
    elif run_status == "timeout":
        result["error"] = f"the re-run did not end within {WAIT_SECONDS // 3600} hours, the fix stays deployed"
        save(cur, attempt_id, "error", result, True)
        notify(cur)
    else:
        result["error"] = f"the re-run ended {run_status}"
        roll_back(cur, attempt_id, result)
        notify(cur)

def run_watch(cur, attempt_id):
    cur.execute("select attempt_run_id from meta.log.t_dlthub_fix_attempt where attempt_id = %s", (attempt_id,))
    run_status = wait_for_run(cur.fetchone()[0])
    result = {"mode": "watch", "run_status": run_status}
    if run_status == "completed":
        save(cur, attempt_id, "succeeded", result, True)
        notify(cur, "dlthub_fix_succeeded")
        return
    save(cur, attempt_id, "error" if run_status == "timeout" else "failed", result, True)
    notify(cur)

def main():
    attempt_id = os.environ["ATTEMPT_ID"]
    mode = os.environ["MODE"]
    cur = connect().cursor()
    cur.execute(
        "select details, source_run_id from meta.log.t_dlthub_fix_attempt where attempt_id = %s", (attempt_id,)
    )
    raw, source_run_id = cur.fetchone()
    details = json.loads(raw) if raw else {}
    result = {"mode": mode, "gates": {}}
    try:
        if mode == "watch":
            run_watch(cur, attempt_id)
        else:
            run_fix(cur, attempt_id, mode, result, details, source_run_id)
    except (GateFailed, py_compile.PyCompileError) as e:
        result["error"] = f"py_compile: {str(e)[:400]}" if isinstance(e, py_compile.PyCompileError) else str(e)
        save(cur, attempt_id, "dry_failed" if mode == "dry" else "rejected", result, True)
        if mode != "dry":
            notify(cur)
    except Exception as e:
        result["error"] = f"{type(e).__name__}: {str(e)[:400]}"
        save(cur, attempt_id, "error", result, True)
        if mode != "dry":
            notify(cur)
    print(json.dumps({"mode": mode, "error": result.get("error"), "gates": result["gates"]}))

if __name__ == "__main__":
    main()

```

The first time the inspector picked up a fixable issue in prod was two days ago 🥳 A source REST API updated it's pagination method, so the dlt pipeline required a [modified parameter](https://dlthub.com/docs/dlt-ecosystem/verified-sources/rest%5Fapi/basic?ref=sfrt.io#pagination). The inspector identified the issue, modified the pipeline, deployed and ran it and when it succeeded, it let me know in the morning.

## Guards go in the procedures

Putting the rules into the agent instructions was my first mistake: "retry only once", "never fix without high confidence". Such instructions are suggestions and a model tends to obey what it's told... until it doesn't.

Hence the rules ended up in the tools, where the code above enforces them:

- One attempt per failed run, and a run that an attempt kicked off never gets retried or fixed again.
- Only the job's latest failed run is acted on, not any run id the agent names.
- A ceiling of 3 attempts per job over 24 hours.
- Retries only for `transient`. Fixes only for `code` or `config` when confidence is `high`.
- No action unless the latest second opinion for that run and that class agrees, is no older than 30 minutes, and Jev is at least 60% sure (I'll get to this next).
- The kill switch.

The prompt mentions some of them, so the agent doesn't waste a call on something that would get refused anyway. But the agent's words carry no authority anywhere. The job ref, the run status and the attempt count all come from dltHub and from the table. 

Oh, and pulling the plug looks like this:

```sql
-- stop new attempts (running ones finish)
update meta.log.t_dlthub_fix_switch
set enabled = false, changed_at = current_timestamp(), changed_by = current_user();
-- or one action only: ... where action = 'fix'   (actions: retry, fix)
```

With the switch off, the procedures reply "switched off" and the agent tells whoever reads the card (again: me) that a human needs to step in.

## A second opinion with a veto

After classifying a failure, the agent calls `Second_Opinion`. That procedure sends a trimmed run record, the last 150 log lines (anything credential-shaped masked) and the statuses of the previous few runs to Jev, a small classifier model from TypeSafe. Jev never sees the agent's answer and classifies the same evidence on its own, against the same class list the agent uses:

```python
INSTRUCTIONS = (
    "Classify why this dltHub job run failed, from the run record, the log tail, the pipeline"
    " error and the neighbouring runs. Decide by where the error is raised, not by what its"
    " message claims. An exception raised by the job's own statements (a hard-coded raise, a"
    " NameError, a bad access on a value the code built) is `code`. An error that the log shows"
    " as an upstream response, which the code merely surfaced through an HTTP client or"
    " `raise_for_status`, is not a code bug: 401 or 403 means `credentials`; 429, a 5xx status,"
    " a connection reset or a timeout means `transient` when the neighbouring runs show it does"
    " not persist, otherwise `upstream_data`."
)
CRITERIA = {
    "config": "missing or wrong setting, wrong profile, a trigger or manifest that cannot work as written",
    "credentials": "authentication failure reaching a source or destination",
    "upstream_data": "the job ran correctly; the data it received was wrong, late or absent (an empty or malformed payload), not a service outage",
    "code": "an exception in workspace code; the traceback points into the pipeline or transformation",
    "resources": "out-of-memory kill, timeout, or disk exhaustion",
    "transient": "network blip, rate limit, service unavailable (HTTP 429, 502, 503, 504), timeout or platform-side failure; neighbouring runs show it does not persist",
    "unknown": "the evidence does not establish a cause",
}
# credential-looking values never leave the account
SECRET = re.compile(r"(?i)(bearer\s+)[A-Za-z0-9._~+/=-]+|((?:api[_-]?key|token|secret|password|passwd|sig|signature)\s*[=:]\s*)[^\s&\"',;]+")

```

Why not use other [decision models in Snowflake](https://www.snowflake.com/en/developers/blog/decisions-are-all-you-need-run-jev-class-decision-models-on-snowflake/?ref=sfrt.io)? Or `AI_CLASSIFY`? No real reason, I just had the [API to Jev](https://www.sfrt.io/asking-small-decision-models-jev-laya-about-first-names/) already set up 😅

The call itself is one line with the TypeSafe SDK, and agreement means Jev's top class matches the agent's, with a probability of at least 0.6 (a match at 0.51 is a coin flip). `unknown` never agrees, and an exception counts as disagreement (so a Jev outage means nothing acts):

```python
def main(session, run_id, classification):
    agent = (classification or "").strip().lower()
    try:
        workspace = dlthub_sdk.connect(
            token=_snowflake.get_generic_secret_string("dlthub"), base_url="https://api.dlthub.com"
        ).workspaces.get(id=WORKSPACE_ID)
        state = evidence(workspace, run_id)
        with TypeSafeClient(api_key=_snowflake.get_generic_secret_string("typesafe"), timeout=90) as client:
            answer = client.system_one(state=state, questions={"classification": Choice(instructions=INSTRUCTIONS, criteria=CRITERIA)})
        choice = answer.choices["classification"]
        probabilities = {k: round(float(v), 2) for k, v in (choice.probabilities or {}).items()}
        jev = choice.choice
        probability = probabilities.get(jev, 0.0)
    except Exception as e:
        error = f"{type(e).__name__}: {str(e)[:300]}"
        store(session, run_id, agent, None, None, False, None, error)
        return json.dumps({"status": "unavailable", "agrees": False, "reason": error})
    agrees = bool(agent) and agent != "unknown" and agent == jev
    store(session, run_id, agent, jev, probability, agrees, probabilities, None)
    top = dict(sorted(probabilities.items(), key=lambda kv: -kv[1])[:3])
    return json.dumps({"status": "ok", "agrees": agrees, "agent_classification": agent, "jev_classification": jev,
                       "probability": probability, "top_choices": top})
$$
```

Every answer lands in `meta.log.t_dlthub_second_opinion`, and that row (the latest one for that run and class, agreeing, at least 0.6, under 30 minutes old) is what `second_opinion_agrees()` in the retry and fix procedures looks for. The agent can't talk its way past it.

Show the full p\_dlthub\_second\_opinion▾

```sql
create or alter table meta.log.t_dlthub_second_opinion (
      run_id                varchar         not null
    , created_at            timestamp_ltz   not null default current_timestamp()
    , agent_classification  varchar         null
    , jev_classification    varchar         null
    , jev_probability       number(5,2)     null
    , agrees                boolean         not null
    , probabilities         variant         null
    , error                 varchar         null
)
comment = 'Jev second opinion on the classification of a failed dltHub run, one row per call'
;

create or replace procedure meta.cortex.p_dlthub_second_opinion(run_id varchar, classification varchar)
copy grants
returns varchar
language python
runtime_version = '3.13'
artifact_repository = snowflake.snowpark.pypi_shared_repository
packages = ('snowflake-snowpark-python', 'dlthub-client', 'typesafe-sdk')
handler = 'main'
comment = 'JOB_INSPECTOR tool: asks Jev to classify a failed dltHub run and compares it with the agent classification; stored in meta.log.t_dlthub_second_opinion'
external_access_integrations = (i_dlthub, i_typesafe)
secrets = ('dlthub' = meta.integration.se_dlthub_viewer_api_key, 'typesafe' = meta.integration.se_typesafe)
execute as owner
as
$$
import json
import re

import _snowflake
import dlthub_sdk
from typesafe_sdk import Choice, TypeSafeClient

WORKSPACE_ID = "4f3da7b4-ae6a-43db-8ce6-d909aba0b47f"
LOG_LINES = 150
NEIGHBOURS = 8
TABLE = "meta.log.t_dlthub_second_opinion"

INSTRUCTIONS = (
    "Classify why this dltHub job run failed, from the run record, the log tail, the pipeline"
    " error and the neighbouring runs. Decide by where the error is raised, not by what its"
    " message claims. An exception raised by the job's own statements (a hard-coded raise, a"
    " NameError, a bad access on a value the code built) is `code`. An error that the log shows"
    " as an upstream response, which the code merely surfaced through an HTTP client or"
    " `raise_for_status`, is not a code bug: 401 or 403 means `credentials`; 429, a 5xx status,"
    " a connection reset or a timeout means `transient` when the neighbouring runs show it does"
    " not persist, otherwise `upstream_data`."
)
CRITERIA = {
    "config": "missing or wrong setting, wrong profile, a trigger or manifest that cannot work as written",
    "credentials": "authentication failure reaching a source or destination",
    "upstream_data": "the job ran correctly; the data it received was wrong, late or absent (an empty or malformed payload), not a service outage",
    "code": "an exception in workspace code; the traceback points into the pipeline or transformation",
    "resources": "out-of-memory kill, timeout, or disk exhaustion",
    "transient": "network blip, rate limit, service unavailable (HTTP 429, 502, 503, 504), timeout or platform-side failure; neighbouring runs show it does not persist",
    "unknown": "the evidence does not establish a cause",
}
# credential-looking values never leave the account
SECRET = re.compile(
    r"(?i)((?:bearer|authorization\s*[=:]\s*basic)\s+)[A-Za-z0-9._~+/=-]+"
    r"|((?:api[_-]?key|token|secret|password|passwd|sig|signature)[\"']?\s*[=:]\s*[\"']?)[^\s&\"',;]+"
    r"|(://[^/\s:@]+:)[^@\s/]+(?=@)"
    r"|\b(?:ghp|gho|ghs|github_pat|sk|xox[bp])[_-][A-Za-z0-9_-]{16,}"
    r"|\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]+"
)

def value(x):
    return getattr(x, "value", x)

def mask(text):
    return SECRET.sub(lambda m: (m.group(1) or m.group(2) or m.group(3) or "") + "***", text)

def evidence(workspace, run_id):
    run = workspace.job_runs.get(id=run_id)
    record = dict(run.to_dict())
    job_ref = record["job_ref"]
    lines = list(run.logs())
    program = [l for l in lines if str(value(l.source)) != "setup"] or lines
    neighbours = [
        {"number": n.get("number"), "status": str(value(n.get("status"))), "started_at": str(n.get("started_at"))[:19],
         "duration_s": round(float(n.get("duration_seconds") or 0))}
        for n in (dict(r.to_dict()) for r in workspace.jobs.get_by_ref(ref=job_ref).runs.list(limit=NEIGHBOURS))
    ]
    state = {
        "run": {"job": str(job_ref).rsplit(".", 1)[-1], "number": record.get("number"),
                "status": str(value(record.get("status"))), "trigger": record.get("trigger"), "profile": record.get("profile")},
        "log_tail": [mask(f"{l.line_num} {l.content}") for l in program[-LOG_LINES:]],
        "neighbour_runs": neighbours,
    }
    return state

def store(session, run_id, agent, jev, probability, agrees, probabilities, error):
    session.sql(
        f"insert into {TABLE} (run_id, agent_classification, jev_classification, jev_probability, agrees, probabilities, error) "
        f"select ?, ?, ?, ?, ?, parse_json(?), ?",
        params=[run_id, agent, jev, probability, agrees, json.dumps(probabilities) if probabilities is not None else None, error],
    ).collect()

def main(session, run_id, classification):
    agent = (classification or "").strip().lower()
    try:
        workspace = dlthub_sdk.connect(
            token=_snowflake.get_generic_secret_string("dlthub"), base_url="https://api.dlthub.com"
        ).workspaces.get(id=WORKSPACE_ID)
        state = evidence(workspace, run_id)
        with TypeSafeClient(api_key=_snowflake.get_generic_secret_string("typesafe"), timeout=90) as client:
            answer = client.system_one(state=state, questions={"classification": Choice(instructions=INSTRUCTIONS, criteria=CRITERIA)})
        choice = answer.choices["classification"]
        probabilities = {k: round(float(v), 2) for k, v in (choice.probabilities or {}).items()}
        jev = choice.choice
        probability = probabilities.get(jev, 0.0)
    except Exception as e:
        error = f"{type(e).__name__}: {str(e)[:300]}"
        store(session, run_id, agent, None, None, False, None, error)
        return json.dumps({"status": "unavailable", "agrees": False, "reason": error})
    agrees = bool(agent) and agent != "unknown" and agent == jev
    store(session, run_id, agent, jev, probability, agrees, probabilities, None)
    top = dict(sorted(probabilities.items(), key=lambda kv: -kv[1])[:3])
    return json.dumps({"status": "ok", "agrees": agrees, "agent_classification": agent, "jev_classification": jev,
                       "probability": probability, "top_choices": top})
$$
;

alter procedure if exists meta.cortex.p_dlthub_second_opinion(varchar, varchar) set tag meta.tag.triggered_by = 'cortex_agent';
```

Agreement lifts confidence to `high`. A disagreement (or Jev being unreachable) drops it to `low`, the tools refuse to act, and a person gets a card.

That rule first went into the agent prompt, where the model I used during development and testing (Haiku 5.5) cheerfully ignored it. The dltHub job that calls the agent now overwrites the confidence in code:

```python
output["inspector_run_id"] = own_run_id
output["trace_request_id"] = trace_request_id
if output.get("status") not in ("skipped", "aborted"):
    # the confidence is Jev's verdict, whatever the agent wrote: the second opinion agrees = high,
    # anything else (disagrees, unavailable, not asked) = low and a person has to look
    agrees = (output.get("second_opinion") or {}).get("agrees") is True
    output["confidence"] = "high" if agrees else "low"
    if not agrees:
        output["requires_human"] = True
```

A genuine disagreement turned up in production as well: the agent said `config`, Jev said `upstream_data` at 0.61, nothing happened, a card arrived. That is the system doing its job. But of course, this might get tweaked if I get too annoyed by repeatedly wrong opinions...

## What still worries me

- Every retry and every fix is worth exactly what the diagnosis is worth. Two models can agree and still be wrong together. All the guards and the rollback cap the damage rather than prevent it.
- `dlthub deploy` uploads a working directory, so a container deploy off `master` overwrites whatever sat deployed locally and uncommitted. I guess I need to remember to work on branches now 😅
- Any `uv.lock` change in the dltHub repo costs an image rebuild and a fresh tag. I didn't really think about that part, yet, but it will certainly become annoying soon enough. And then I'll look into it.

To copy this, the order I'd build it in: the attempt table and the switch first, then `Retry_Job` with a canary, then the fixer in mode `dry` only, and the second opinion before anything gets to push.

With an agent writing to my repo, I want to check if it was wrong about its own work, not only about jobs failing. That calls for a second pair of eyes on what the inspector actually did... the next post.

So there it is: boring failures now close their own loop, and the interesting ones still reach me with a diagnosis attached 😎

---

## The relevant parts of the inspector instructions

```
ATTEMPTS: if Check_Attempts shows is_attempt_run true, an earlier retry started this run
and that attempt is already watched: stop, call nothing else and return status skipped.
If it shows already_attempted true, diagnose as usual but call neither Retry_Job nor Apply_Fix.

SECOND OPINION: after you classify, call Second_Opinion once with the failed run id and your
classification. It asks Jev, an independent system, to classify the same run.
agrees true: confidence is high. agrees false or status unavailable: confidence is low,
requires_human is true, you call neither Retry_Job nor Apply_Fix, and you still return your
own classification, naming Jev's class and probability in summary.
Never change your classification to match Jev.

RETRY: when you classify transient (after the neighbour check), call Retry_Job once.
The tool refuses by itself when it must not retry. Never call it for another classification.
A started retry is watched by the system: set requires_human false and proposed_fix to
'none unless the retry fails'. If the tool refuses or errors, report that and set
requires_human true.

FIX: when you classify code or config with confidence high, and Read_Source shows the lines
of the traceback at the same commit, call Apply_Fix once with a JSON array of minimal edits
[{"path", "old", "new"}] and a one-sentence rationale.
- Copy old verbatim from Read_Source so that it occurs exactly once in the file.
- Change as little as possible.
- Never guess a column, field or key name: an edit may only use a name that the log,
  the trace or the source carries.
- Patch only the job's own pipeline files, never __deployment__.py, secrets or config files.
If the source does not match the traceback, the cause is not certain from the evidence,
or the fix needs more than a few lines, do not call it and set requires_human true.
The rationale ends up in the Teams note, so make it say what was wrong and what the edit does.
```

#### License

This is free and unencumbered software released into the public domain. Anyone is free to copy, modify, publish, use, compile, sell, or distribute this software, either in source code form or as a compiled binary, for any purpose, commercial or non-commercial, and by any means.  
In jurisdictions that recognize copyright laws, the author or authors of this software dedicate any and all copyright interest in the software to the public domain. We make this dedication for the benefit of the public at large and to the detriment of our heirs and successors. We intend this dedication to be an overt act of relinquishment in perpetuity of all present and future rights to this software under copyright law.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

For more information, please refer to [https://unlicense.org/](https://unlicense.org/?ref=sfrt.io)