Cortex 代理升级：可自动重试、修复并重新部署失败的 dltHub 任务

Last time the agent only read logs and wrote a diagnosis. Now it pushes commits to master 🤖

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.

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 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.

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, 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:

{ "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 😎

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:

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):

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.

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:

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:

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:

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() retur