Driving DAG Desk over HTTP

Everything the page does, you can do from a script. One endpoint does the work; the rest is authentication and polling. The app takes one Airflow DAG file and a task field naming which of five jobs to run over it, and returns a single JSON envelope.

Base URL

https://api.skillsafe.ai/v1/app-api

Every request carries Authorization: Bearer <token>. The token is scoped to this app when it is minted, so no slug header is needed on later calls - POST /guest is the one call that names the slug, in its JSON body. Get a token from the token page without opening a developer console.

The response envelope

Every endpoint returns the same wrapper. Success carries data; failure carries error. Nothing returns a bare value, so a client can branch on the presence of error alone.

{"ok": true,  "data":  { ... }}
{"ok": false, "error": {"code": "VALIDATION_ERROR", "message": "…", "details": { ... }}}

Error codes

Code	HTTP	What it means	What to do
UNAUTHORIZED	401	Missing, malformed or expired token.	Mint a new one. Guest tokens expire; personal tokens outlive them.
FORBIDDEN	403	The token is valid but not for this app.	Mint the token against this app: `POST /guest` with `{"slug":"dag-desk"}`.
VALIDATION_ERROR	400	The input did not match the app's shape.	Read `error.details`; it names the offending field.
PAYMENT_REQUIRED	402	Balance below the run's minimum.	Call `/estimate` first and compare against `/me`.
RATE_LIMITED	429	Too many requests.	Back off and retry; never tight-loop.
JOB_FAILED	200	The job reached a terminal failed state.	Returned inside a job payload, not as an HTTP error. Check `status`.

The `task` field comes first

DAG Desk is a five-lane app. Every run must name its lane in task, because the lanes share one system prompt and one model and are routed by that field. If it is missing or unrecognised the model picks the closest lane and says so by setting lane_inferred to true — which is a fallback, not a feature to rely on.

`task`	What it does	Extra input	`artifact.kind`
audit	Reviews the DAG against Airflow authoring practice across twelve named checks.	—	none
migrate	Reports what breaks on Airflow 3 and rewrites the file for it.	—	python
lineage	Infers the data assets the DAG reads and writes, and annotates it.	—	python
tests	Writes the pytest suite that guards the DAG.	—	python
diagnose	Root-causes a failing task run.	log	python

Input fields

Field	Type	Required	Notes
task	string	yes	One of the five lane ids above.
dag	string	yes	The full text of the DAG `.py` file. Several files may be concatenated with `# file: path` marker lines.
log	string	diagnose only	The Airflow task log of the failing run. Send the tail if it is long — the diagnosis is in the last lines.
airflow_source	string	no	`2.x`, `3.x` or `unknown`. The page infers this from the imports.
airflow_target	string	no	Defaults to `3.x`.
prescan	object	no	`{resources: [], flags: []}` — deterministic facts the model must reconcile. See below.
clip_note	string	no	Tells the model the input was truncated so it writes around the gap instead of inventing it. The page clips from the middle and keeps the head and the tail.
retry_note	string	no	Sent only on an automatic re-ask after a malformed reply. It quotes the parse error and restates the contract; the model answers the same lane on the same input and returns only the JSON object. The page reuses an idempotency key derived from the same input with the attempt number appended, so a retry cannot double-bill.

The prescan contract

The browser app runs a deterministic scanner before every run and passes its findings in prescan.flags, each with a stable id. The model must return exactly one coverage_check entry per flag id sent, and none for ids that were not sent. That is what makes the free lane able to hold the paid lane accountable — anything unaccounted for is a defect you can detect programmatically:

sent    = {f["id"] for f in payload["prescan"]["flags"]}
covered = {c["flag_id"] for c in result["coverage_check"]}
assert sent == covered, f"unreconciled: {sent - covered}"

Each flag is {id, label, severity, line, occurrences}. Flags are deduplicated by rule id before they are sent, so line is the first line the rule fired on and occurrences is how many times it fired across the whole file. An occurrences of 7 means seven sites need the same change — one coverage_check entry still covers them all.

prescan.resources is a separate array of {id, label} entries naming the connections, hooks, variables, pools and datasets the scanner saw the DAG reference. They are context, not findings: they need no coverage_check entry and the model is instructed not to manufacture a finding just to mention one.

You may send an empty prescan — the lane still works, it simply has fewer facts to ground itself in and nothing to reconcile.

1. A tiny client

Six lines of setup that every later step reuses: the base URL, the app slug header, the bearer token, and a JSON post that raises on the error branch of the envelope. Put your token in an environment variable rather than a literal in the file.

# Every call in this document uses these three values.
BASE="https://api.skillsafe.ai/v1/app-api"
SLUG="dag-desk"
TOKEN="YOUR_TOKEN"        # from https://dag-desk.skillsafe.ai/tokens.html

post() {  # post <path> <json>
  curl -sS -X POST "$BASE$1" \
    -H "Authorization: Bearer $TOKEN" \
    \
    -H "Content-Type: application/json" \
    -d "$2"
}

import json
import urllib.request

BASE = "https://api.skillsafe.ai/v1/app-api"
SLUG = "dag-desk"
TOKEN = "YOUR_TOKEN"   # from https://dag-desk.skillsafe.ai/tokens.html


def call(path, payload=None, method="POST"):
    body = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(BASE + path, data=body, method=method)
    req.add_header("Authorization", "Bearer " + TOKEN)
    req.add_header("Content-Type", "application/json")
    with urllib.request.urlopen(req) as resp:
        envelope = json.loads(resp.read())
    if not envelope.get("ok"):
        raise RuntimeError(envelope["error"]["code"] + ": " + envelope["error"]["message"])
    return envelope["data"]

const BASE = "https://api.skillsafe.ai/v1/app-api";
const SLUG = "dag-desk";
const TOKEN = "YOUR_TOKEN"; // from https://dag-desk.skillsafe.ai/tokens.html

async function call(path, payload, method = "POST") {
  const res = await fetch(BASE + path, {
    method,
    headers: {
      "Authorization": `Bearer ${TOKEN}`,
      "Content-Type": "application/json"
    },
    body: payload === undefined ? undefined : JSON.stringify(payload)
  });
  const envelope = await res.json();
  if (!envelope.ok) {
    throw new Error(`${envelope.error.code}: ${envelope.error.message}`);
  }
  return envelope.data;
}

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
)

const (
	base = "https://api.skillsafe.ai/v1/app-api"
	slug = "dag-desk"
	token = "YOUR_TOKEN" // from https://dag-desk.skillsafe.ai/tokens.html
)

type envelope struct {
	OK    bool            `json:"ok"`
	Data  json.RawMessage `json:"data"`
	Error *struct {
		Code    string `json:"code"`
		Message string `json:"message"`
	} `json:"error"`
}

func call(path string, payload any) (json.RawMessage, error) {
	var body io.Reader
	if payload != nil {
		b, _ := json.Marshal(payload)
		body = bytes.NewReader(b)
	}
	req, _ := http.NewRequest("POST", base+path, body)
	req.Header.Set("Authorization", "Bearer "+token)
	req.Header.Set("Content-Type", "application/json")

	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	var env envelope
	if err := json.NewDecoder(res.Body).Decode(&env); err != nil {
		return nil, err
	}
	if !env.OK {
		return nil, fmt.Errorf("%s: %s", env.Error.Code, env.Error.Message)
	}
	return env.Data, nil
}

import java.net.URI;
import java.net.http.*;

public class DagDesk {
    static final String BASE = "https://api.skillsafe.ai/v1/app-api";
    static final String SLUG = "dag-desk";
    static final String TOKEN = "YOUR_TOKEN"; // from https://dag-desk.skillsafe.ai/tokens.html

    static final HttpClient CLIENT = HttpClient.newHttpClient();

    static String call(String path, String jsonBody) throws Exception {
        HttpRequest req = HttpRequest.newBuilder()
            .uri(URI.create(BASE + path))
            .header("Authorization", "Bearer " + TOKEN)
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(jsonBody))
            .build();

        HttpResponse<String> res = CLIENT.send(req, HttpResponse.BodyHandlers.ofString());
        if (res.statusCode() >= 400) {
            throw new RuntimeException("HTTP " + res.statusCode() + ": " + res.body());
        }
        return res.body();
    }
}

require "json"
require "net/http"
require "uri"

BASE  = "https://api.skillsafe.ai/v1/app-api"
SLUG  = "dag-desk"
TOKEN = "YOUR_TOKEN" # from https://dag-desk.skillsafe.ai/tokens.html

def call(path, payload = nil)
  uri = URI(BASE + path)
  req = Net::HTTP::Post.new(uri)
  req["Authorization"] = "Bearer #{TOKEN}"
  req["Content-Type"] = "application/json"
  req.body = JSON.dump(payload) unless payload.nil?

  res = Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) { |http| http.request(req) }
  envelope = JSON.parse(res.body)
  raise "#{envelope['error']['code']}: #{envelope['error']['message']}" unless envelope["ok"]

  envelope["data"]
end

<?php
const BASE  = "https://api.skillsafe.ai/v1/app-api";
const SLUG  = "dag-desk";
const TOKEN = "YOUR_TOKEN"; // from https://dag-desk.skillsafe.ai/tokens.html

function call(string $path, ?array $payload = null): array {
    $ch = curl_init(BASE . $path);
    curl_setopt_array($ch, [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_POST => true,
        CURLOPT_HTTPHEADER => [
            "Authorization: Bearer " . TOKEN,
            "Content-Type: application/json",
        ],
        CURLOPT_POSTFIELDS => $payload === null ? "" : json_encode($payload),
    ]);
    $envelope = json_decode(curl_exec($ch), true);
    curl_close($ch);

    if (empty($envelope["ok"])) {
        throw new RuntimeException($envelope["error"]["code"] . ": " . $envelope["error"]["message"]);
    }
    return $envelope["data"];
}

using System.Net.Http;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;

public static class DagDesk
{
    const string Base = "https://api.skillsafe.ai/v1/app-api";
    const string Slug = "dag-desk";
    const string Token = "YOUR_TOKEN"; // from https://dag-desk.skillsafe.ai/tokens.html

    static readonly HttpClient Client = new HttpClient();

    public static async Task<JsonElement> CallAsync(string path, object payload = null)
    {
        var req = new HttpRequestMessage(HttpMethod.Post, Base + path);
        req.Headers.Authorization = new AuthenticationHeaderValue("Bearer", Token);
        req.Content = new StringContent(
            payload is null ? "" : JsonSerializer.Serialize(payload),
            Encoding.UTF8, "application/json");

        var res = await Client.SendAsync(req);
        var envelope = JsonDocument.Parse(await res.Content.ReadAsStringAsync()).RootElement;

        if (!envelope.GetProperty("ok").GetBoolean())
        {
            var err = envelope.GetProperty("error");
            throw new Exception($"{err.GetProperty("code")}: {err.GetProperty("message")}");
        }
        return envelope.GetProperty("data");
    }
}

2. Get a token

A guest token is minted on demand and is enough for /me and /estimate. Running a lane is metered, so it needs a personal token — sign in at the token page and copy it from there. Note that every POST /guest mints a new guest identity, so reuse one token across a session rather than minting per request.

curl -sS -X POST "$BASE/guest" \
  -H "Content-Type: application/json" \
  -d "{\"slug\":\"$SLUG\"}" | tee guest.json
# {"ok":true,"data":{"token":"aut_...","subject_type":"guest"}}
TOKEN=$(python3 -c "import json;print(json.load(open('guest.json'))['data']['token'])")

import json, urllib.request

body = json.dumps({"slug": SLUG}).encode()
req = urllib.request.Request(BASE + "/guest", data=body, method="POST")
req.add_header("Content-Type", "application/json")
with urllib.request.urlopen(req) as resp:
    guest = json.loads(resp.read())["data"]

TOKEN = guest["token"]          # reuse this for the whole session
print(guest["subject_type"])    # "guest"

const res = await fetch(`${BASE}/guest`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ slug: SLUG })
});
const guest = (await res.json()).data;

const token = guest.token;      // reuse for the whole session
console.log(guest.subject_type); // "guest"

guestBody := bytes.NewReader([]byte(`{"slug":"dag-desk"}`))
req, _ := http.NewRequest("POST", base+"/guest", guestBody)
req.Header.Set("Content-Type", "application/json")

res, err := http.DefaultClient.Do(req)
if err != nil {
	panic(err)
}
defer res.Body.Close()

var env struct {
	Data struct {
		Token       string `json:"token"`
		SubjectType string `json:"subject_type"`
	} `json:"data"`
}
json.NewDecoder(res.Body).Decode(&env)
fmt.Println(env.Data.SubjectType) // "guest"

HttpRequest req = HttpRequest.newBuilder()
    .uri(URI.create(BASE + "/guest"))
    .header("Content-Type", "application/json")
    .POST(HttpRequest.BodyPublishers.ofString("{\"slug\":\"dag-desk\"}"))
    .build();

HttpResponse<String> res = CLIENT.send(req, HttpResponse.BodyHandlers.ofString());
System.out.println(res.body());
// {"ok":true,"data":{"token":"aut_...","subject_type":"guest"}}

uri = URI(BASE + "/guest")
req = Net::HTTP::Post.new(uri)
req["Content-Type"] = "application/json"
req.body = JSON.dump({ "slug" => SLUG })

res = Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) { |http| http.request(req) }
guest = JSON.parse(res.body)["data"]

token = guest["token"]           # reuse for the whole session
puts guest["subject_type"]       # "guest"

<?php
$ch = curl_init(BASE . "/guest");
curl_setopt_array($ch, [
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_POST => true,
    CURLOPT_HTTPHEADER => ["Content-Type: application/json"],
    CURLOPT_POSTFIELDS => json_encode(["slug" => SLUG]),
]);
$guest = json_decode(curl_exec($ch), true)["data"];
curl_close($ch);

$token = $guest["token"];          // reuse for the whole session
echo $guest["subject_type"];       // "guest"

var req = new HttpRequestMessage(HttpMethod.Post, Base + "/guest");
req.Content = new StringContent("{\"slug\":\"dag-desk\"}", Encoding.UTF8, "application/json");

var res = await Client.SendAsync(req);
var guest = JsonDocument.Parse(await res.Content.ReadAsStringAsync())
    .RootElement.GetProperty("data");

var token = guest.GetProperty("token").GetString();   // reuse for the session
Console.WriteLine(guest.GetProperty("subject_type")); // "guest"

3. Check who you are and what you can spend

GET /me is free and tells you the subject type and the credit balance. Compare that balance against the estimate in the next step before you run anything — a 402 after submitting is a failure of the client, not of the user.

curl -sS "$BASE/me" \
  -H "Authorization: Bearer $TOKEN" \
# {"ok":true,"data":{"subject_type":"user","credits":48210, ...}}

me = call("/me", method="GET")
print(me["subject_type"], me["credits"])

const me = await call("/me", undefined, "GET");
console.log(me.subject_type, me.credits);

req, _ := http.NewRequest("GET", base+"/me", nil)
req.Header.Set("Authorization", "Bearer "+token)

res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()

var env struct {
	Data struct {
		SubjectType string `json:"subject_type"`
		Credits     int    `json:"credits"`
	} `json:"data"`
}
json.NewDecoder(res.Body).Decode(&env)
fmt.Println(env.Data.SubjectType, env.Data.Credits)

HttpRequest req = HttpRequest.newBuilder()
    .uri(URI.create(BASE + "/me"))
    .header("Authorization", "Bearer " + TOKEN)
    .GET()
    .build();

System.out.println(CLIENT.send(req, HttpResponse.BodyHandlers.ofString()).body());

uri = URI(BASE + "/me")
req = Net::HTTP::Get.new(uri)
req["Authorization"] = "Bearer #{TOKEN}"

res = Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) { |http| http.request(req) }
me = JSON.parse(res.body)["data"]
puts "#{me['subject_type']} #{me['credits']}"

<?php
$ch = curl_init(BASE . "/me");
curl_setopt_array($ch, [
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_HTTPHEADER => [
        "Authorization: Bearer " . TOKEN,
    ],
]);
$me = json_decode(curl_exec($ch), true)["data"];
curl_close($ch);

echo $me["subject_type"] . " " . $me["credits"];

var req = new HttpRequestMessage(HttpMethod.Get, Base + "/me");
req.Headers.Authorization = new AuthenticationHeaderValue("Bearer", Token);

var res = await Client.SendAsync(req);
var me = JsonDocument.Parse(await res.Content.ReadAsStringAsync())
    .RootElement.GetProperty("data");

Console.WriteLine($"{me.GetProperty("subject_type")} {me.GetProperty("credits")}");

4. Price the run before you make it

POST /estimate is free, creates no job and charges nothing. It returns the model, the markup and — the number that matters — hold_credits, the amount reserved for the run. The hold differs per lane, because the prompts and output caps differ, so re-estimate whenever task changes. Present it as reserved, never as the price: the actual charge is usually far lower.

post /estimate '{
  "task": "audit",
  "dag": "from airflow import DAG\nfrom airflow.operators.python import PythonOperator\n...",
  "airflow_target": "3.x"
}'
# {"ok":true,"data":{"model":"gpt-5.6-terra","model_alias":"gpt-terra",
#                    "markup_bps":1000,"hold_credits":3820,"min_credits":420}}

payload = {
    "task": "audit",
    "dag": open("dags/orders_etl.py").read(),
    "airflow_target": "3.x",
}

est = call("/estimate", payload)
print(est["model"], est["hold_credits"], est["min_credits"])

if me["credits"] < est["min_credits"]:
    raise SystemExit(f"short by {est['min_credits'] - me['credits']} credits")

import { readFileSync } from "node:fs";

const payload = {
  task: "audit",
  dag: readFileSync("dags/orders_etl.py", "utf8"),
  airflow_target: "3.x"
};

const est = await call("/estimate", payload);
console.log(est.model, est.hold_credits, est.min_credits);

if (me.credits < est.min_credits) {
  throw new Error(`short by ${est.min_credits - me.credits} credits`);
}

dag, _ := os.ReadFile("dags/orders_etl.py")

payload := map[string]any{
	"task":           "audit",
	"dag":            string(dag),
	"airflow_target": "3.x",
}

data, err := call("/estimate", payload)
if err != nil {
	panic(err)
}

var est struct {
	Model       string `json:"model"`
	HoldCredits int    `json:"hold_credits"`
	MinCredits  int    `json:"min_credits"`
}
json.Unmarshal(data, &est)
fmt.Println(est.Model, est.HoldCredits, est.MinCredits)

String dag = Files.readString(Path.of("dags/orders_etl.py"));

String payload = """
    {"task": "audit", "dag": %s, "airflow_target": "3.x"}
    """.formatted(JsonUtil.quote(dag));

String estimate = call("/estimate", payload);
System.out.println(estimate);
// {"ok":true,"data":{"model":"gpt-5.6-terra","hold_credits":3820, ...}}

payload = {
  "task" => "audit",
  "dag" => File.read("dags/orders_etl.py"),
  "airflow_target" => "3.x"
}

est = call("/estimate", payload)
puts "#{est['model']} #{est['hold_credits']} #{est['min_credits']}"

abort "short by #{est['min_credits'] - me['credits']}" if me["credits"] < est["min_credits"]

<?php
$payload = [
    "task" => "audit",
    "dag" => file_get_contents("dags/orders_etl.py"),
    "airflow_target" => "3.x",
];

$est = call("/estimate", $payload);
echo "{$est['model']} {$est['hold_credits']} {$est['min_credits']}\n";

if ($me["credits"] < $est["min_credits"]) {
    throw new RuntimeException("short by " . ($est["min_credits"] - $me["credits"]));
}

var payload = new
{
    task = "audit",
    dag = await File.ReadAllTextAsync("dags/orders_etl.py"),
    airflow_target = "3.x"
};

var est = await CallAsync("/estimate", payload);
Console.WriteLine($"{est.GetProperty("model")} {est.GetProperty("hold_credits")}");

5. Run a lane and poll for the result

POST /run returns a job_id immediately; poll GET /jobs/{id} until status is terminal. Always send an Idempotency-Key derived from the lane plus the input — two lanes over the same DAG are two distinct runs and must not collide on one key, and a retried request with the same key will never bill twice.

KEY="dag-desk:audit:$(shasum -a 256 dags/orders_etl.py | cut -c1-16)"

JOB=$(curl -sS -X POST "$BASE/run" \
  -H "Authorization: Bearer $TOKEN" \
  \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $KEY" \
  -d @payload.json | python3 -c "import json,sys;print(json.load(sys.stdin)['data']['job_id'])")

until curl -sS "$BASE/jobs/$JOB" -H "Authorization: Bearer $TOKEN" \
  | tee job.json | grep -q '"status":"succeeded"'; do sleep 2; done

python3 -c "import json;print(json.load(open('job.json'))['data']['output']['output'])"

import hashlib, time

key = "dag-desk:%s:%s" % (payload["task"], hashlib.sha256(payload["dag"].encode()).hexdigest()[:16])

job = call("/run", dict(payload, idempotency_key=key))
job_id = job["job_id"]

while True:
    status = call("/jobs/" + job_id, method="GET")
    if status["status"] in ("succeeded", "failed", "cancelled"):
        break
    time.sleep(2)

if status["status"] != "succeeded":
    raise RuntimeError("job " + status["status"])

result = json.loads(status["output"]["output"])
print(result["posture"], "-", result["verdict"])
for f in result["findings"]:
    print(f"  [{f['severity']:8}] {f['id']} {f['title']}")

import { createHash } from "node:crypto";

const key = `dag-desk:${payload.task}:` +
  createHash("sha256").update(payload.dag).digest("hex").slice(0, 16);

const job = await call("/run", { ...payload, idempotency_key: key });

let status;
do {
  await new Promise(r => setTimeout(r, 2000));
  status = await call(`/jobs/${job.job_id}`, undefined, "GET");
} while (!["succeeded", "failed", "cancelled"].includes(status.status));

if (status.status !== "succeeded") throw new Error(`job ${status.status}`);

const result = JSON.parse(status.output.output);
console.log(result.posture, "-", result.verdict);
for (const f of result.findings) {
  console.log(`  [${f.severity}] ${f.id} ${f.title}`);
}

sum := sha256.Sum256([]byte(payload["dag"].(string)))
payload["idempotency_key"] = fmt.Sprintf("dag-desk:%s:%x", payload["task"], sum[:8])

data, err := call("/run", payload)
if err != nil {
	panic(err)
}

var job struct {
	JobID string `json:"job_id"`
}
json.Unmarshal(data, &job)

for {
	time.Sleep(2 * time.Second)

	statusData, _ := call("/jobs/"+job.JobID, nil)

	var st struct {
		Status string `json:"status"`
		Output struct {
			Output string `json:"output"`
		} `json:"output"`
	}
	json.Unmarshal(statusData, &st)

	if st.Status == "succeeded" {
		fmt.Println(st.Output.Output)
		break
	}
	if st.Status == "failed" || st.Status == "cancelled" {
		panic("job " + st.Status)
	}
}

String key = "dag-desk:audit:" + Integer.toHexString(dag.hashCode());

HttpRequest run = HttpRequest.newBuilder()
    .uri(URI.create(BASE + "/run"))
    .header("Authorization", "Bearer " + TOKEN)
    .header("Content-Type", "application/json")
    .header("Idempotency-Key", key)
    .POST(HttpRequest.BodyPublishers.ofString(payload))
    .build();

String jobId = JsonUtil.path(
    CLIENT.send(run, HttpResponse.BodyHandlers.ofString()).body(), "data", "job_id");

String status;
do {
    Thread.sleep(2000);
    status = call("/jobs/" + jobId, "");
} while (!status.contains("\"status\":\"succeeded\"")
      && !status.contains("\"status\":\"failed\""));

System.out.println(status);

require "digest"

key = "dag-desk:#{payload['task']}:#{Digest::SHA256.hexdigest(payload['dag'])[0, 16]}"
job = call("/run", payload.merge("idempotency_key" => key))

status = nil
loop do
  sleep 2
  status = call("/jobs/#{job['job_id']}")
  break if %w[succeeded failed cancelled].include?(status["status"])
end

raise "job #{status['status']}" unless status["status"] == "succeeded"

result = JSON.parse(status["output"]["output"])
puts "#{result['posture']} - #{result['verdict']}"
result["findings"].each { |f| puts "  [#{f['severity']}] #{f['id']} #{f['title']}" }

<?php
$key = "dag-desk:{$payload['task']}:" . substr(hash("sha256", $payload["dag"]), 0, 16);
$payload["idempotency_key"] = $key;

$job = call("/run", $payload);

do {
    sleep(2);
    $status = call("/jobs/" . $job["job_id"]);
} while (!in_array($status["status"], ["succeeded", "failed", "cancelled"], true));

if ($status["status"] !== "succeeded") {
    throw new RuntimeException("job " . $status["status"]);
}

$result = json_decode($status["output"]["output"], true);
echo "{$result['posture']} - {$result['verdict']}\n";
foreach ($result["findings"] as $f) {
    echo "  [{$f['severity']}] {$f['id']} {$f['title']}\n";
}

using System.Security.Cryptography;

var hash = Convert.ToHexString(
    SHA256.HashData(Encoding.UTF8.GetBytes(payload.dag)))[..16].ToLower();
var key = $"dag-desk:{payload.task}:{hash}";

var job = await CallAsync("/run", new { payload.task, payload.dag, idempotency_key = key });
var jobId = job.GetProperty("job_id").GetString();

JsonElement status;
string state;
do
{
    await Task.Delay(2000);
    status = await CallAsync($"/jobs/{jobId}");
    state = status.GetProperty("status").GetString();
} while (state is not ("succeeded" or "failed" or "cancelled"));

if (state != "succeeded") throw new Exception($"job {state}");

var result = JsonDocument.Parse(
    status.GetProperty("output").GetProperty("output").GetString()).RootElement;
Console.WriteLine(result.GetProperty("verdict"));

6. Stream it instead, for anything interactive

POST /run-stream is the same call over Server-Sent Events. Deltas arrive as they are generated, which is what the page uses to advance its progress stages. The same Idempotency-Key rule applies. Accumulate the deltas and parse the JSON once the stream closes — the envelope is only valid complete.

curl -sS -N -X POST "$BASE/run-stream" \
  -H "Authorization: Bearer $TOKEN" \
  \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $KEY" \
  -H "Accept: text/event-stream" \
  -d @payload.json

# event: delta
# data: {"text":"{\"lane\":\"audit\","}
# event: done
# data: {"job_id":"job_...","charged_credits":2711,"truncated":false}

import urllib.request

req = urllib.request.Request(BASE + "/run-stream", data=json.dumps(payload).encode())
req.add_header("Authorization", "Bearer " + TOKEN)
req.add_header("Content-Type", "application/json")
req.add_header("Idempotency-Key", key)
req.add_header("Accept", "text/event-stream")

chunks = []
with urllib.request.urlopen(req) as stream:
    for raw in stream:
        line = raw.decode().strip()
        if not line.startswith("data:"):
            continue
        event = json.loads(line[5:].strip())
        if "text" in event:
            chunks.append(event["text"])
            print(".", end="", flush=True)

result = json.loads("".join(chunks))
print("\n", result["posture"], result["verdict"])

const res = await fetch(`${BASE}/run-stream`, {
  method: "POST",
  headers: {
    "Authorization": `Bearer ${TOKEN}`,
    "Content-Type": "application/json",
    "Idempotency-Key": key,
    "Accept": "text/event-stream"
  },
  body: JSON.stringify(payload)
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "", text = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  const lines = buffer.split("\n");
  buffer = lines.pop();

  for (const line of lines) {
    if (!line.startsWith("data:")) continue;
    const event = JSON.parse(line.slice(5).trim());
    if (event.text) text += event.text;
  }
}

const result = JSON.parse(text);
console.log(result.posture, result.verdict);

req, _ := http.NewRequest("POST", base+"/run-stream", bytes.NewReader(body))
req.Header.Set("Authorization", "Bearer "+token)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Idempotency-Key", key)
req.Header.Set("Accept", "text/event-stream")

res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()

var sb strings.Builder
scanner := bufio.NewScanner(res.Body)
scanner.Buffer(make([]byte, 1024*1024), 1024*1024)

for scanner.Scan() {
	line := scanner.Text()
	if !strings.HasPrefix(line, "data:") {
		continue
	}
	var ev struct {
		Text string `json:"text"`
	}
	if json.Unmarshal([]byte(strings.TrimSpace(line[5:])), &ev) == nil && ev.Text != "" {
		sb.WriteString(ev.Text)
	}
}
fmt.Println(sb.String())

HttpRequest req = HttpRequest.newBuilder()
    .uri(URI.create(BASE + "/run-stream"))
    .header("Authorization", "Bearer " + TOKEN)
    .header("Content-Type", "application/json")
    .header("Idempotency-Key", key)
    .header("Accept", "text/event-stream")
    .POST(HttpRequest.BodyPublishers.ofString(payload))
    .build();

StringBuilder text = new StringBuilder();

CLIENT.send(req, HttpResponse.BodyHandlers.ofLines())
      .body()
      .filter(line -> line.startsWith("data:"))
      .forEach(line -> text.append(JsonUtil.path(line.substring(5).trim(), "text")));

System.out.println(text);

uri = URI(BASE + "/run-stream")
req = Net::HTTP::Post.new(uri)
req["Authorization"] = "Bearer #{TOKEN}"
req["Content-Type"] = "application/json"
req["Idempotency-Key"] = key
req["Accept"] = "text/event-stream"
req.body = JSON.dump(payload)

text = +""
Net::HTTP.start(uri.hostname, uri.port, use_ssl: true) do |http|
  http.request(req) do |res|
    res.read_body do |chunk|
      chunk.each_line do |line|
        next unless line.start_with?("data:")

        event = JSON.parse(line[5..].strip) rescue next
        text << event["text"] if event["text"]
      end
    end
  end
end

result = JSON.parse(text)
puts "#{result['posture']} #{result['verdict']}"

<?php
$text = "";

$ch = curl_init(BASE . "/run-stream");
curl_setopt_array($ch, [
    CURLOPT_POST => true,
    CURLOPT_HTTPHEADER => [
        "Authorization: Bearer " . TOKEN,
        "Content-Type: application/json",
        "Idempotency-Key: " . $key,
        "Accept: text/event-stream",
    ],
    CURLOPT_POSTFIELDS => json_encode($payload),
    CURLOPT_WRITEFUNCTION => function ($ch, $chunk) use (&$text) {
        foreach (explode("\n", $chunk) as $line) {
            if (str_starts_with($line, "data:")) {
                $event = json_decode(trim(substr($line, 5)), true);
                if (isset($event["text"])) $text .= $event["text"];
            }
        }
        return strlen($chunk);
    },
]);
curl_exec($ch);
curl_close($ch);

$result = json_decode($text, true);
echo "{$result['posture']} {$result['verdict']}\n";

var req = new HttpRequestMessage(HttpMethod.Post, Base + "/run-stream");
req.Headers.Authorization = new AuthenticationHeaderValue("Bearer", Token);
req.Headers.Add("Idempotency-Key", key);
req.Headers.Add("Accept", "text/event-stream");
req.Content = new StringContent(JsonSerializer.Serialize(payload), Encoding.UTF8, "application/json");

var res = await Client.SendAsync(req, HttpCompletionOption.ResponseHeadersRead);
using var reader = new StreamReader(await res.Content.ReadAsStreamAsync());

var text = new StringBuilder();
while (await reader.ReadLineAsync() is { } line)
{
    if (!line.StartsWith("data:")) continue;

    var ev = JsonDocument.Parse(line[5..].Trim()).RootElement;
    if (ev.TryGetProperty("text", out var t)) text.Append(t.GetString());
}

var result = JsonDocument.Parse(text.ToString()).RootElement;
Console.WriteLine(result.GetProperty("verdict"));

7. One worked example per lane

Each lane sends the same dag and differs only in task. The envelope is identical across all five; only body and artifact change. Fields shown as ... follow the shapes documented above.

`task: "audit"` — Audit

Twelve named checks, an inventory, findings with fixes and quick wins. This is the only lane that emits no file.

Request

{
  "task": "audit",
  "dag": "<the DAG source>",
  "airflow_target": "3.x",
  "prescan": {
    "resources": [
      {
        "id": "DAG-1",
        "label": "dag_id orders_etl (context style, line 32)"
      }
    ],
    "flags": [
      {
        "id": "DS-TOPLEVEL-VARIABLE",
        "label": "Variable.get() runs on every scheduler parse",
        "severity": "high",
        "line": 12
      }
    ]
  }
}