← Skewcheck / API
Tokens

Drive Skewcheck from your own code

The page reads one pasted PySpark job in the browser for free — one or more files joined into one paste, each introduced by a # file: <path> line: the Python, and optionally its spark-submit script, spark-defaults.conf and requirements.txt (which gives the pyspark version) — with its line numbers. It lexes strings and comments out first, rebuilds every block from indentation, follows every DataFrame from its read to its write, and inventories every read (format, target, schema, inferSchema, JDBC partitioning), every write (format, mode, partitionBy, target), every action (collect, toPandas, count, show, take, write) and whether it sits in a Python loop, every join and whether it is broadcast, every window, Python UDF, RDD round trip and cache, and every Spark setting from the session builder, spark.conf.set, a settings dict, spark-submit flags or spark-defaults.conf. Then it runs 20 deterministic checks drawn from Spark tuning practice (catalogue ids SP-01..SP-20, plus SP-21 for context): an unbounded collect() or toPandas(), Python UDFs and rdd.map round trips, count() used to test emptiness, actions and plans grown inside Python loops, a window with no partitionBy, a join with no condition, adaptive query execution or broadcast joins switched off, repartition(1) before a write, a write partitioned by an id, schema inference, caches that buy nothing, groupByKey, an unpartitioned JDBC read, a credential literal, a global sort before a write, debug actions, executor sizing outside the usual range, and an overwrite that deletes every other partition. Nothing is imported or run: every flag is a reading of the text. Its two paid steps are available over HTTP. You send one JSON object with task set to "audit" or "fix", the paste as code and the page’s check as a facts string, and the model replies with one JSON object: an audit that rates the driver, shuffle, compute and io, answers every flag and gives exact before/after edits per finding, or the whole fixed file(s) with a list of every change tied to the page’s flags.

Base URL: https://api.skillsafe.ai/v1/app-api. Every call sends Content-Type: application/json and, once you have a token, Authorization: Bearer <token>. The paths below (/guest, /me, /estimate, /run, /jobs/{job_id}, /run-stream) are relative to it, exactly as sdk.js calls them. A plain-text summary of the app for agents is at /llms.txt. Both lanes run on the gpt-terra model with a 10% publisher markup. A run needs a signed-in (user) token and is metered: the platform places a hold on your credits when the run starts and then charges only what the run used. /estimate is free.

The task field: two lanes

task is the first field of every body. It routes the run, and it takes exactly two values: "audit" and "fix". Each lane has its own reply contract, and a reply never mixes them. The skill’s instructions tell the model to do the audit lane when task is missing or anything else (and the page’s own Sparkkit.buildInput turns any lane that is not "fix" into "audit"), so a typo silently gets you an audit: always send one of the two values.

taskWhat it doesverdictUse it when
"audit"Reviews the job the way a senior data engineer reviews a Spark job before it runs on the full data: follows each DataFrame from its read to its write, says what the job reads, computes and writes (job), rates the four areas (driver, shuffle, compute, io) as sound, fragile or broken, confirms or dismisses every page flag, adds what text rules cannot see, and gives the verbatim evidence and exact before/after edits for each finding, listed in the order to apply them — what fails at scale first (driver collects, unkeyed windows and joins), then what wastes the most work, then what loses data or leaks, then hygiene — with a plan that ends in a run on one day or a sample, df.explain("formatted") and the Spark UI’s stages.blocked | needs_work | readyBefore a job runs at scale: will it kill the driver, run one job per loop pass, shuffle everything through one task or delete yesterday’s output, and what should change first.
"fix"Writes the fixed version of the pasted job: the whole of every pasted file it changes (Python, the spark-submit script or spark-defaults.conf; never a new file), with every function, read, write, output path, format, mode and partition column kept — plus a changes list tied to the page flags, the verify steps (a run on a small slice compared with the old job’s output, plus the check that would have caught the worst problem), and what it left open for a decision or a fact only the engineer has (a schema’s columns, a JDBC split column, which key a window is per).fixed | partialWhen you want the job changed, not a list of findings: a job you already know is broken, or what is left after an audit.

The page chains the two: after an audit, “Apply every edit and fix the rest” puts every audit edit into the paste, switches to the fix lane on that edited paste and adds an audit_handoff string (below) with the audit’s verdict, headline and findings. Over the API that is a second run with task: "fix", the edited paste as code, facts recomputed from the edited paste, and that string. The page drops the handoff as soon as the paste is edited again.

Guards the page applies before a run and the API does not: it refuses to price or run a paste it cannot read as a job (“Nothing was pasted.”, “No Python file was found. Paste the PySpark job (.py), with a # file: <path>.py line above each file.”, or “This does not read as Python source: no def, class, import or call was found.”), and it warns before a run on a paste with no Spark in it (“No SparkSession, pyspark import or Spark read was found: there may be little Spark here to audit.”, or “... to fix.”).

One worked example per lane follows, both examples the page ships (the example buttons on the app page; the full pastes are in /example.js, ids orders and churn), built exactly as the page builds them: Sparkkit.buildInput(lane, Sparkkit.analyze(text), note, null), with the saved model reply for each. The request bodies cut the long code and facts strings with a visible ..., and the replies are pretty-printed excerpts, with ... where entries or text were left out. Every value shown is copied from the input or the reply. The companies, buckets and hosts are invented.

Worked example, task: "audit": a nightly orders job that collects every active customer id to the driver and ranks orders in one global window

Three files: requirements.txt (pyspark==3.5.1), submit.sh (a spark-submit call that sets spark.sql.adaptive.enabled=false) and jobs/orders_nightly.py. main reads a day of order CSVs with inferSchema, tests emptiness with orders.count() == 0, collects every active customer id into a Python list for an isin() filter, cleans the SKU in a Python UDF, numbers orders with Window.orderBy(F.col("order_ts")) (no partitionBy) and writes through repartition(1). The page’s free check raises 7 flags (2 high, 5 medium) (plus the SP-21 context flag); its own reading of the verdict is blocked. The engineer’s note: “Nightly job over a day of orders; the driver died last night once customers passed 40 million rows. order_seq is meant to number each customer's orders by time. Downstream reads the whole output directory with Spark, so one file is not needed. The CSV header is order_id,customer_id,sku,quantity,unit_price,order_ts - order_id, customer_id and sku are strings, quantity is an integer, unit_price a decimal(10,2), order_ts an ISO-8601 timestamp. What do we fix first?”

The code string, whole: 1,764 characters, 66 lines ending in a newline. Save it as orders-paste.txt and run node build-check.js orders-paste.txt > orders-check.json (build-check.js): 

# file: requirements.txt
pyspark==3.5.1
---
# file: submit.sh
#!/bin/sh
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 16g \
  --executor-cores 4 \
  --num-executors 20 \
  --conf spark.sql.adaptive.enabled=false \
  --conf spark.sql.shuffle.partitions=2000 \
  jobs/orders_nightly.py "$1"
---
# file: jobs/orders_nightly.py
import sys

from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F
from pyspark.sql.types import StringType

RAW_ORDERS = "s3a://quellmor-lake/raw/orders/"
CUSTOMERS = "s3a://quellmor-lake/curated/customers/"
OUT = "s3a://quellmor-lake/curated/orders_enriched/"


def clean_sku(sku):
    if sku is None:
        return None
    return sku.strip().upper().replace(" ", "-")


clean_sku_udf = F.udf(clean_sku, StringType())


def main(run_date):
    spark = SparkSession.builder.appName("orders-nightly").getOrCreate()

    orders = (spark.read
              .option("header", "true")
              .option("inferSchema", "true")
              .csv(RAW_ORDERS + run_date))

    if orders.count() == 0:
        print("no orders for", run_date)
        return

    customers = spark.read.parquet(CUSTOMERS)
    active_ids = [r.customer_id for r in customers.filter(F.col("status") == "active").select("customer_id").collect()]

    enriched = (orders
                .filter(F.col("customer_id").isin(active_ids))
                .withColumn("sku", clean_sku_udf(F.col("sku")))
                .join(customers, "customer_id", "left"))

    w = Window.orderBy(F.col("order_ts"))
    ranked = enriched.withColumn("order_seq", F.row_number().over(w))

    (ranked.repartition(1)
        .write.mode("overwrite")
        .parquet(OUT + run_date))


if __name__ == "__main__":
    main(sys.argv[1])

Its facts, decoded (the request carries this object as one compact JSON string):

{
  "job": {
    "spark_version": "3.5.1",
    "python_files": 1,
    "functions": 2,
    "dataframes": 4,
    "reads": 2,
    "writes": 1,
    "actions": 2,
    "joins": 1,
    "settings": 5
  },
  "files": [
    {
      "path": "requirements.txt",
      "kind": "deps",
      "line": 2,
      "end_line": 2
    },
    {
      "path": "submit.sh",
      "kind": "submit",
      "line": 5,
      "end_line": 14
    },
    {
      "path": "jobs/orders_nightly.py",
      "kind": "python",
      "line": 17,
      "end_line": 67
    }
  ],
  "functions": [
    {
      "name": "clean_sku",
      "file": "jobs/orders_nightly.py",
      "line": 28
    },
    {
      "name": "main",
      "file": "jobs/orders_nightly.py",
      "line": 37
    }
  ],
  "dataframes": [
    "customers",
    "enriched",
    "orders",
    "ranked"
  ],
  "reads": [
    {
      "file": "jobs/orders_nightly.py",
      "line": 40,
      "func": "main",
      "format": "csv",
      "target": "RAW_ORDERS + run_date",
      "schema": false,
      "infer": true,
      "jdbc_partitioned": false,
      "stream": false
    },
    {
      "file": "jobs/orders_nightly.py",
      "line": 49,
      "func": "main",
      "format": "parquet",
      "target": "CUSTOMERS",
      "schema": false,
      "infer": false,
      "jdbc_partitioned": false,
      "stream": false
    }
  ],
  "writes": [
    {
      "file": "jobs/orders_nightly.py",
      "line": 60,
      "func": "main",
      "receiver": "ranked.repartition(1)",
      "format": "parquet",
      "mode": "overwrite",
      "partition_by": [],
      "target": "OUT + run_date",
      "stream": false,
      "in_loop": false
    }
  ],
  "actions": [
    {
      "file": "jobs/orders_nightly.py",
      "line": 45,
      "func": "main",
      "kind": "count",
      "receiver": "orders",
      "in_loop": false
    },
    {
      "file": "jobs/orders_nightly.py",
      "line": 50,
      "func": "main",
      "kind": "collect",
      "receiver": "customers.filter(F.col(\"status\") == \"active\").select(\"customer_id\")",
      "in_loop": true
    }
  ],
  "joins": [
    {
      "file": "jobs/orders_nightly.py",
      "line": 52,
      "func": "main",
      "kind": "join",
      "receiver": "orders.filter(F.col(\"customer_id\").isin(active_ids)).withColumn(\"sku\", clean_sku_udf(F.col(\"sku\")))",
      "other": "customers",
      "on": "\"customer_id\"",
      "how": "left",
      "broadcast": false
    }
  ],
  "windows": [
    {
      "file": "jobs/orders_nightly.py",
      "line": 57,
      "func": "main",
      "name": "w",
      "spec": "Window.orderBy(F.col(\"order_ts\"))",
      "partitioned": false
    }
  ],
  "udfs": [
    {
      "file": "jobs/orders_nightly.py",
      "line": 34,
      "name": "clean_sku_udf",
      "kind": "udf",
      "how": "call"
    }
  ],
  "caches": [],
  "settings": [
    {
      "key": "spark.executor.memory",
      "value": "16g",
      "file": "submit.sh",
      "line": 9,
      "source": "spark-submit"
    },
    {
      "key": "spark.executor.cores",
      "value": "4",
      "file": "submit.sh",
      "line": 10,
      "source": "spark-submit"
    },
    {
      "key": "spark.executor.instances",
      "value": "20",
      "file": "submit.sh",
      "line": 11,
      "source": "spark-submit"
    },
    {
      "key": "spark.sql.adaptive.enabled",
      "value": "false",
      "file": "submit.sh",
      "line": 12,
      "source": "spark-submit"
    },
    {
      "key": "spark.sql.shuffle.partitions",
      "value": "2000",
      "file": "submit.sh",
      "line": 13,
      "source": "spark-submit"
    }
  ],
  "flags": [
    {
      "uid": "F1",
      "id": "SP-01",
      "sev": "high",
      "file": "jobs/orders_nightly.py",
      "line": 50,
      "text": "collect() pulls every row of customers into the driver",
      "detail": "Nothing in the lineage of customers.filter(F.col(\"status\") == \"active\").select(\"custom limits or aggregates it: every row is shipped to the driver process and held in its memory, so the job fails with an out-of-memory error (or spark.driver.maxResultSize) as soon as the data grows. Keep the work distributed (a join, a filter, a write) or bound it first with limit(n), take(n) or an aggregate."
    },
    {
      "uid": "F2",
      "id": "SP-06",
      "sev": "high",
      "file": "jobs/orders_nightly.py",
      "line": 57,
      "text": "window w has no partitionBy",
      "detail": "A window with only an ordering moves every row into a single partition on one executor (Spark logs \"No Partition Defined for Window operation! Moving all data to a single partition\"); on real data that task runs for hours or dies. Partition the window by the key the ranking or running total is per (Window.partitionBy(\"account_id\").orderBy(...)); for a global top-n use orderBy(...).limit(n)."
    },
    {
      "uid": "F3",
      "id": "SP-08",
      "sev": "medium",
      "file": "submit.sh",
      "line": 12,
      "text": "spark.sql.adaptive.enabled is set to false",
      "detail": "Adaptive query execution re-plans each stage from real sizes: it coalesces tiny shuffle partitions, splits skewed ones and turns sort-merge joins into broadcast joins when one side turns out small. It is on by default since Spark 3.2; switching it off means spark.sql.shuffle.partitions and every skewed key are yours to tune by hand."
    },
    {
      "uid": "F4",
      "id": "SP-02",
      "sev": "medium",
      "file": "jobs/orders_nightly.py",
      "line": 34,
      "text": "Python UDF clean_sku_udf runs row by row in a Python worker",
      "detail": "Every row is serialised from the JVM to a Python process and back, and the optimizer cannot see inside the function (no predicate pushdown, no code generation). Most string, date, arithmetic and conditional logic has a built-in in pyspark.sql.functions (when/otherwise, regexp_replace, coalesce, date_format, transform ...); keep a UDF only for logic no built-in covers, and then prefer a pandas_udf."
    },
    {
      "uid": "F5",
      "id": "SP-12",
      "sev": "medium",
      "file": "jobs/orders_nightly.py",
      "line": 40,
      "text": "inferSchema reads RAW_ORDERS + run_date twice",
      "detail": "Schema inference makes Spark scan the input once just to guess the types, then read it again; the guess also changes when the data changes (a column of numbers that once contains a letter becomes a string). Pass an explicit schema with .schema(StructType(...)) or a DDL string."
    },
    {
      "uid": "F6",
      "id": "SP-03",
      "sev": "medium",
      "file": "jobs/orders_nightly.py",
      "line": 45,
      "text": "count() used to test whether orders is empty",
      "detail": "Counting runs a full job over every partition just to learn whether one row exists. Use df.isEmpty() (Spark 3.3+) or len(df.take(1)) == 0, which stop at the first row."
    },
    {
      "uid": "F7",
      "id": "SP-10",
      "sev": "medium",
      "file": "jobs/orders_nightly.py",
      "line": 60,
      "text": "repartition(1) before the write to OUT + run_date",
      "detail": "The whole output is written by one task on one executor: the job's last stage has no parallelism, and with coalesce(1) Spark may also pull the upstream work into that single task. Let the write use its partitions (tune spark.sql.files.maxRecordsPerFile or repartition by the output partition columns); if a single file is a hard requirement of the consumer, say so and keep it only for small outputs."
    },
    {
      "uid": "F8",
      "id": "SP-21",
      "sev": "info",
      "file": "jobs/orders_nightly.py",
      "line": 17,
      "text": "context: 1 Python file, 4 DataFrame name(s), 2 read(s), 1 write(s), 2 action(s), 1 join(s), 5 setting(s), pyspark 3.5.1",
      "detail": "Also read: requirements.txt (deps), submit.sh (submit). Only what is pasted is checked: data sizes, key skew and the cluster are not visible in code, so anything that depends on them is a question, not a finding. The Spark UI (stage durations, shuffle read, spill, task time skew) is the ground truth."
    }
  ],
  "verdict_candidate": "blocked",
  "notes": []
}

The request body (POST /estimate, then POST /run or /run-stream with Idempotency-Key: skewcheck:audit:d6d7adf6-251b:a1 as the page sends it): 

{
  "task": "audit",
  "code": "# file: requirements.txt\npyspark==3.5.1\n---\n# file: submit.sh\n#!/bin/sh\nspark-submit \\\n  --master yarn \\\n  --deploy-mode cluster \\\n  --executor-memory 16g \\\n  --executor-cores 4 \\\n  --num-executors 20 \\\n  --conf spark.sql.adaptive.enabled=false \\\n  --conf spar...",
  "facts": "{\"job\":{\"spark_version\":\"3.5.1\",\"python_files\":1,\"functions\":2,\"dataframes\":4,\"reads\":2,\"writes\":1,\"actions\":2,\"joins\":1,\"settings\":5},\"files\":[{\"path\":\"requirements.txt\",\"kind\":\"deps\",\"line\":2,\"end_l...",
  "note": "Nightly job over a day of orders; the driver died last night once customers passed 40 million rows. order_seq is meant to number each customer's orders by time. Downstream reads the whole output directory with Spark, so one file is not needed. The CSV header is order_id,customer_id,sku,quantity,unit_price,order_ts - order_id, customer_id and sku are strings, quantity is an integer, unit_price a decimal(10,2), order_ts an ISO-8601 timestamp. What do we fix first?"
}

The reply (output.output), an excerpt: the verdict, the headline, each area’s status, and the finding that answers “the driver died last night”, with its edit and its flag answer: 

{
  "task": "audit",
  "verdict": "blocked",
  "headline": "Two high-severity issues - an unbounded collect() of active customer ids and a ranking window with no partitionBy - are why the driver died last night and will keep failing or stalling as customers and orders grow, and adaptive execution is disabled so nothing self-corrects.",
  "job": "Reads a day of raw order CSVs and the customers parquet table, drops orders whose customer isn't active, tags each order with a cleaned sku, joins in customer attributes, numbers each order per customer by timestamp, and overwrites that day's directory under orders_enriched.",
  "areas": [
    {
      "area": "driver",
      "status": "broken"
    },
    {
      "area": "shuffle",
      "status": "broken"
    },
    {
      "area": "compute",
      "status": "fragile"
    },
    {
      "area": "io",
      "status": "fragile"
    }
  ],
  "findings": [
    {
      "id": "SP-01",
      "flags": [
        "F1"
      ],
      "file": "jobs/orders_nightly.py",
      "line": 50,
      "severity": "high",
      "area": "driver",
      "title": "Collected active customer ids feed the driver",
      "evidence": "    active_ids = [r.customer_id for r in customers.filter(F.col(\"status\") == \"active\").select(\"customer_id\").collect()]",
      "why": "customers.filter(...).select(\"customer_id\").collect() ships one row per active customer to the driver as a Python list before the isin() filter even runs; with customers at 40 million rows that list alone can be hundreds of MB held in driver memory - this is what killed last night's run, and it fails the same way (OOM or spark.driver.maxResultSize) every time customers grows.",
      "edits": [
        {
          "before": "    customers = spark.read.parquet(CUSTOMERS)\n    active_ids = [r.customer_id for r in customers.filter(F.col(\"status\") == \"active\").select(\"customer_id\").collect()]\n\n    enriched = (orders\n                .filter(F.col(\"customer_id\").isin(active_ids))\n                .withColumn(\"sku\", clean_sku_udf(F.col(\"sku\")))\n                .join(customers, \"customer_id\", \"left\"))",
          "after": "    customers = spark.read.parquet(CUSTOMERS)\n    active_customers = customers.filter(F.col(\"status\") == \"active\").select(\"customer_id\")\n\n    enriched = (orders\n                .join(active_customers, \"customer_id\", \"left_semi\")\n                .withColumn(\"sku\", clean_sku_udf(F.col(\"sku\")))\n                .join(customers, \"customer_id\", \"left\"))"
        }
      ]
    },
    ...
  ],
  "flag_answers": [
    {
      "flag": "F1",
      "status": "confirmed",
      "reason": "collect() on the active customer ids is unbounded and is what died last night."
    },
    ...
  ],
  "plan": [
    ...,
    "Run the job on one day's data, read df.explain(\"formatted\") for a keyed window and a plain join, and check the Spark UI stages for a single long task or spill."
  ],
  "questions": [
    ...
  ],
  "summary": ...
}

The reply has 7 findings, one per flag, and confirms all 7 flags. Besides the left_semi join above, its edits partition the window by customer_id (the note says order_seq numbers each customer’s orders), replace orders.count() == 0 with orders.isEmpty() (pyspark 3.5.1 has it), replace the UDF with F.regexp_replace and F.upper, remove the spark.sql.adaptive.enabled=false line from submit.sh (that finding names submit.sh, the file it edits), drop repartition(1) (the note says downstream reads the whole directory), and pin an explicit ORDERS_SCHEMA built from the column list the note gives — 9 edits in all.

Worked example, task: "fix": a churn feature job with a password in its JDBC options, a Python UDF and a count per plan in a loop

One file, jobs/churn_features.py, with no requirements.txt, so the Spark version is unknown. load_customers reads public.customers over JDBC with no partitioning and a literal password in its .option("password", ...); build_features caches the read, builds tenure_bucket with a Python UDF and runs one count() per plan in a for loop; the session builder sets spark.executor.cores to 8. The free check raises 6 flags (4 medium, 2 low); its reading of the verdict is needs_work. This example has no audit before it, so it carries no audit_handoff. The engineer’s note: “Fix what you can. Leave the JDBC read's partitioning exactly as it is - the DBA has to approve a split column first. The password must come from the CRM_DB_PASSWORD environment variable the scheduler sets. Keep printing the per-plan counts.”

The code string: 1,229 characters, 47 lines ending in a newline, shown whole except the string literal on line 33, which holds the password and is cut to "..." here (the page flags it as F3; the example in /example.js carries the full line). Save the full paste as churn-paste.txt and run node build-check.js churn-paste.txt > churn-check.json: 

# file: jobs/churn_features.py
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType

JDBC_URL = "jdbc:postgresql://crm-db.internal:5432/crm"
OUT = "s3a://quellmor-lake/features/churn/"

spark = (SparkSession.builder
         .appName("churn-features")
         .config("spark.executor.cores", "8")
         .getOrCreate())


def tenure_bucket(days):
    if days is None:
        return 0.0
    if days < 90:
        return 1.0
    if days < 365:
        return 2.0
    return 3.0


tenure_udf = F.udf(tenure_bucket, DoubleType())


def load_customers():
    return (spark.read.format("jdbc")
            .option("url", JDBC_URL)
            .option("dbtable", "public.customers")
            .option("user", "etl_reader")
            .option("password", "...")
            .load())


def build_features():
    customers = load_customers().cache()
    features = customers.withColumn("tenure_bucket", tenure_udf(F.col("tenure_days")))
    for plan in ["basic", "plus", "pro"]:
        n = features.filter(F.col("plan") == plan).count()
        print(plan, n)
    features.write.mode("overwrite").parquet(OUT)


if __name__ == "__main__":
    build_features()

Its facts, decoded, cut to job, files, reads, settings, flags and verdict_candidate (the full object also lists the 3 functions, the 3 DataFrame names, the write to OUT, the count() in the loop, the UDF and the cache). The facts never carry the credential: a setting whose key names one shows as ***, and the flag says only where it is: 

{
  "job": {
    "spark_version": null,
    "python_files": 1,
    "functions": 3,
    "dataframes": 3,
    "reads": 1,
    "writes": 1,
    "actions": 1,
    "joins": 0,
    "settings": 1
  },
  "files": [
    {
      "path": "jobs/churn_features.py",
      "kind": "python",
      "line": 2,
      "end_line": 48
    }
  ],
  "reads": [
    {
      "file": "jobs/churn_features.py",
      "line": 29,
      "func": "load_customers",
      "format": "jdbc",
      "target": "public.customers",
      "schema": false,
      "infer": false,
      "jdbc_partitioned": false,
      "stream": false
    }
  ],
  "settings": [
    {
      "key": "spark.executor.cores",
      "value": "8",
      "file": "jobs/churn_features.py",
      "line": 11,
      "source": "builder"
    }
  ],
  "flags": [
    {
      "uid": "F1",
      "id": "SP-02",
      "sev": "medium",
      "file": "jobs/churn_features.py",
      "line": 25,
      "text": "Python UDF tenure_udf runs row by row in a Python worker",
      "detail": "Every row is serialised from the JVM to a Python process and back, and the optimizer cannot see inside the function (no predicate pushdown, no code generation). Most string, date, arithmetic and conditional logic has a built-in in pyspark.sql.functions (when/otherwise, regexp_replace, coalesce, date_format, transform ...); keep a UDF only for logic no built-in covers, and then prefer a pandas_udf."
    },
    {
      "uid": "F2",
      "id": "SP-15",
      "sev": "medium",
      "file": "jobs/churn_features.py",
      "line": 29,
      "text": "JDBC read of public.customers through one connection",
      "detail": "Without partitionColumn, lowerBound, upperBound and numPartitions (or predicates) Spark reads the whole table in a single task over one connection: no parallelism, and one executor holds the result. Give it a numeric or date column to split on, and set fetchsize."
    },
    {
      "uid": "F3",
      "id": "SP-16",
      "sev": "medium",
      "file": "jobs/churn_features.py",
      "line": 33,
      "text": "credential literal in read option password",
      "detail": "A password or key written into the job or its settings ends up in version control, in the Spark UI's environment tab and in event logs. Read it at run time from the platform's secret store (a Databricks secret scope, a Kubernetes secret, the cluster's credential provider) and rotate the one that was committed."
    },
    {
      "uid": "F4",
      "id": "SP-04",
      "sev": "medium",
      "file": "jobs/churn_features.py",
      "line": 40,
      "text": "Spark action (count()) inside a Python loop",
      "detail": "Each pass of the loop launches its own Spark job (loop at line 40), re-reading and re-computing the input every time. Do it in one job: a groupBy over the key the loop walks, a join against a small DataFrame of the loop values, or one write partitioned by that key."
    },
    {
      "uid": "F5",
      "id": "SP-19",
      "sev": "low",
      "file": "jobs/churn_features.py",
      "line": 11,
      "text": "spark.executor.cores is 8",
      "detail": "More than about five concurrent tasks per executor tends to hurt HDFS/S3 throughput and share one heap between many tasks; four or five cores per executor, with more executors, is the usual shape."
    },
    {
      "uid": "F6",
      "id": "SP-13",
      "sev": "low",
      "file": "jobs/churn_features.py",
      "line": 38,
      "text": "customers is cached but used once afterwards",
      "detail": "A cache pays off only when two or more actions reuse the same DataFrame; used once it costs executor memory (and a materialisation) for nothing. Drop the cache(), or unpersist it after its last use."
    },
    {
      "uid": "F7",
      "id": "SP-21",
      "sev": "info",
      "file": "jobs/churn_features.py",
      "line": 2,
      "text": "context: 1 Python file, 3 DataFrame name(s), 1 read(s), 1 write(s), 1 action(s), 0 join(s), 1 setting(s), Spark version unknown",
      "detail": "Only what is pasted is checked: data sizes, key skew and the cluster are not visible in code, so anything that depends on them is a question, not a finding. The Spark UI (stage durations, shuffle read, spill, task time skew) is the ground truth."
    }
  ],
  "verdict_candidate": "needs_work"
}

The request body (Idempotency-Key: skewcheck:fix:51b16355-1945:a1): 

{
  "task": "fix",
  "code": "# file: jobs/churn_features.py\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\nfrom pyspark.sql.types import DoubleType\n\nJDBC_URL = \"jdbc:postgresql://crm-db.internal:5432/crm\"\nOUT = \"s3a://quellmor-lake/features/churn/\"\n\nspark = (SparkSession.builder\n         .appName(\"churn-features\")\n         .config(\"spark.executor.cores\", \"8\")\n         .getOrCreate())\n\n\n...",
  "facts": "{\"job\":{\"spark_version\":null,\"python_files\":1,\"functions\":3,\"dataframes\":3,\"reads\":1,\"writes\":1,\"actions\":1,\"joins\":0,\"settings\":1},\"files\":[{\"path\":\"jobs/churn_features.py\",\"kind\":\"python\",\"line\":2,\"...",
  "note": "Fix what you can. Leave the JDBC read's partitioning exactly as it is - the DBA has to approve a split column first. The password must come from the CRM_DB_PASSWORD environment variable the scheduler sets. Keep printing the per-plan counts."
}

The reply (output.output), an excerpt: the verdict, the headline, part of the one returned file (the password now read from the environment), the change that does it, the first verification step and the open item: 

{
  "task": "fix",
  "verdict": "partial",
  "headline": "Removed the row-by-row tenure_bucket UDF, collapsed the per-plan count loop into one job, and read the CRM DB password from an environment variable; left the JDBC read's partitioning untouched as asked, pending DBA approval of a split column.",
  "files": [
    {
      "path": "jobs/churn_features.py",
      "content": "import os\n\nfrom pyspark.sql import SparkSession\n...\ndef load_customers():\n    return (spark.read.format(\"jdbc\")\n            .option(\"url\", JDBC_URL)\n            .option(\"dbtable\", \"public.customers\")\n            .option(\"user\", \"etl_reader\")\n            .option(\"password\", os.environ[\"CRM_DB_PASSWORD\"])\n            .load())\n..."
    }
  ],
  "changes": [
    ...,
    {
      "file": "jobs/churn_features.py",
      "function": "load_customers",
      "flags": [
        "F3"
      ],
      "change": "Read the JDBC password from the CRM_DB_PASSWORD environment variable instead of the literal, adding an os import.",
      "why": "Keeps the credential out of the code, the Spark UI environment tab and event logs; the password that was committed must still be rotated."
    },
    ...
  ],
  "verify": [
    "Run build_features against a small slice of public.customers (a filtered sample or a dev copy of the table) and compare the output against the current job's output for that slice: same row count and a checksum over the key columns (customer id, plan, tenure_bucket).",
    ...
  ],
  "open": [
    {
      "item": "The JDBC read of public.customers still has no partition column, bounds or partition count (or predicates), so it still runs through a single connection and task.",
      "flags": [
        "F2"
      ],
      "reason": "The note says to leave this read's partitioning exactly as it is until the DBA approves a split column; the split column and bounds are a decision for whoever owns the database, not something the code can infer."
    }
  ],
  "questions": [
    "Confirm the scheduler sets CRM_DB_PASSWORD in this job's environment before it runs, since the fix now requires it and will raise a KeyError if it is unset.",
    ...
  ],
  "summary": ...
}

The other two changes, both in build_features, replace the tenure_udf call with a F.when(...).otherwise(...) chain that keeps the null and 90/365-day thresholds (F1), and replace the loop of three count() jobs with one groupBy("plan").count(), still printing basic, plus and pro (F4). The verdict is partial because the note asked to leave the JDBC read’s partitioning as it is, so the medium F2 is named in open rather than fixed. The two low flags (F5, spark.executor.cores of 8, and F6, the cache used once) are left alone, which the contract allows; the reply asks about the executor cores in questions.

The response envelope

Responses are JSON of the form {"ok": ..., "data": ..., "error": {...}}. The page’s SDK (sdk.js) decides success from the HTTP status, not from ok: on a 2xx it returns data; otherwise it raises an error whose message is error.message (or the HTTP status text when the body carries none), with status, code (error.code) and details (error.details) attached. A body that is not JSON is treated as {}. 

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

Error codes

Branch on the HTTP status and error.code, show error.message. The page handles 401 and 402 by name; the rest reach it as a generic failure (“The run failed: ...”).

HTTPerror.codeWhat it means / what to do
400VALIDATION_ERRORThe body is not a JSON object, a required field (task, code, facts) is missing, or a declared field has the wrong type. Every field is a string; sending facts or audit_handoff as an object is the usual cause.
401UNAUTHORIZEDNo token, or one that expired or was revoked. The page forgets the signed-in user and says “Your session expired - sign in again.” Get a new token.
402INSUFFICIENT_CREDITSThe balance cannot cover the run. The page says “Not enough credits for this run.” with a top-up link, and tries never to submit into a 402: it disables the run button when the /me balance is below the estimate’s min_credits.
404NOT_FOUNDA wrong path, or a job id that does not belong to this token.
409CONFLICTAn Idempotency-Key replayed with a different body. Send the original body, or use a new key (the page bumps its :a<attempt> suffix for a retry).
429RATE_LIMITEDToo many requests. Back off with a delay; do not tight-loop, and never resend a metered run without its original Idempotency-Key.

Failures that do not arrive as an HTTP status:

WhatWhat happens
event: error on /run-streamThe SDK raises an error carrying the event’s message, code and job_id.
Job status: "failed"Polling stops; the job’s error field says why.
Polling past 180 sThe SDK’s own client-side limit (waitForJob, 1 s interval, 180 000 ms default): it throws “job timed out”. The job itself is not cancelled.

The input fields

This is exactly what the page sends (sparkkit.js buildInput(lane, A, note, handoff), checked by mustBeObject; app.js currentInput and run), and what input-schema.json declares. The platform’s input schema is scalar-only, so every value is a string. buildInput emits them in the order task, code, facts, note, audit_handoff:

FieldTypeRequiredMeaning
taskstringyes"audit" or "fix" (the lanes above).
codestringyesThe pasted job, each file introduced by a # file: <path> line; line 1 is its first line. Every line in an audit reply is counted in this text. A large paste is cut on whole files, with a # [skewcheck: not sent: ...] line at the end (below).
factsstringyesThe page’s free check, Sparkkit.facts(A), JSON-encoded into a string: JSON.stringify(facts) in JavaScript, json.dumps(facts) in Python, and so on. Sending the object itself is a 400. The model is told to parse it.
notestringnoWhat the engineer says: the goal, a constraint (“the consumer needs one file”, “we are on Spark 3.1”, “do not touch the JDBC read”), facts about the data the code cannot show (“customers has 40 million rows”, the CSV header and its types), a question. The page trims it, cuts it at 2000 characters and sends it only when it is not blank. It is context, not instructions: the model tests every claim about the code against code, may use facts about the data and the cluster as given, and answers the note in summary.
audit_handoffstringnofix lane only: a JSON string built from an audit reply on this code (recon.js fixHandoff, below). buildInput drops it for the audit lane.
retry_notestringnoSent only by the page’s automatic reformat retry: when a reply does not parse as JSON (even after the page tries to close a cut-off object), the page runs once more with “Your previous reply was not the single valid JSON object the instructions require (...). Reply again with ONLY the JSON object for task '<lane>' - no prose, no code fences; every array present (empty arrays where there is nothing to say).” A normal call never sends it.

mustBeObject only insists on a plain object that is not wrapped in {"input": ...}; it is buildInput that leaves a blank note out, which is why an empty note is simply absent.

The code string

code is the paste itself. A line # file: <path> (a path with a . or / in it) starts each file, and a --- line just before it only separates files. Files dropped on the page (or a dropped job folder) are joined that way by Sparkkit.joinFiles: each file under its # file: line, trailing whitespace removed, joined by newlines, ending in a newline. A paste with no # file: line is read as one file, job.py. The page normalises line endings to \n. Each file gets a kind from its path: .py (or .pyi) is python; requirements*.txt, pyproject.toml, setup.cfg and Pipfile are deps (the pinned pyspark version gives facts.job.spark_version — isEmpty() needs 3.3, adaptive execution is on by default from 3.2); spark-defaults.conf, .conf and .properties are conf; .sh, .bash or any file that calls spark-submit is submit (its --conf pairs and --executor-memory, --executor-cores, --driver-memory, --driver-cores and --num-executors flags become settings); anything else is other. Send the text you want the reply to quote: every file in an audit reply must be a path from facts.files, every line must fall inside that file, every evidence must be copied from it, and every edit’s before must occur in it exactly once.

The page reads up to 400 000 characters (MAX_TEXT, cut at a line) and 40 files (MAX_FILES), and sends at most 60 000 characters (MAX_SEND): past that buildInput drops whole files from the end — the last non-Python file first, then the last file — until the rest fits, cuts a single remaining file at a line (“the end of <path>”) if it still does not, and ends the string with one line naming what it left out: 

# [skewcheck: not sent: conf/spark-defaults.conf, jobs/helpers.py]

The model is told never to quote or rewrite what was not sent, and facts.notes gets the sentence “Not sent (the paste is over 60,000 characters): ... The page checked them; the run did not see them.” facts is still computed from the whole paste, and is capped too: past about 40,000 characters (or 150 flags) the page keeps the most severe flags first (flags in files that were sent before flags in files that were not), clips long entries with ... and shortens the longest lists, and facts.notes names every cut (“Facts cut to fit the run: sent 47 of 247 flags ...”). The worked examples above are small, so their code is the paste unchanged and their facts are complete.

The facts string

facts is optional to compute but required to send. It is produced by the page’s free check: Sparkkit.analyze splits the paste into files, lexes the Python, rebuilds its blocks, inventories it and runs the rules, and Sparkkit.facts(A) turns that analysis into the object below (it does not depend on the lane: an audit and a fix run of the same paste carry the same facts). It is how the model knows which flags to answer, which file paths and line ranges to name, and which functions, reads and writes must survive. Its top-level keys, in the order built:

KeyContents
jobspark_version (the pyspark version in a pasted requirements.txt, pyproject.toml or setup.cfg, or null when none was pasted), and counts: python_files, functions, dataframes, reads, writes, actions, joins, settings.
filesEvery file: path (use it verbatim), kind (python, deps, submit, conf, other), line and end_line (the paste lines of the file’s content, without its # file: line).
functionsEvery def (up to 200): name (Class.method for methods), file, line.
dataframesThe names the page reads as DataFrames, sorted (up to 200).
readsEvery read (up to 80): file, line, func, format, target (the path, table or query as written), schema, infer, jdbc_partitioned, stream.
writesEvery write (up to 80): file, line, func, receiver, format, mode, partition_by, target, stream, in_loop.
actionsEvery action (up to 120): file, line, func, kind (collect, toPandas, count, show, take ...), receiver, in_loop (a Python loop or a comprehension).
joinsEvery join (up to 80): file, line, func, kind (join, crossJoin), receiver, other, on, how, broadcast.
windows, udfs, cachesWhere each is (up to 40 each): windows with name, spec and partitioned; UDFs with name, kind (udf, pandas_udf) and how (call, decorator, register); caches with name, kind and in_loop.
settingsEvery Spark setting (up to 120): key, value (*** for a credential key), file, line, source (builder, conf.set, dict, spark-submit, spark-defaults).
flagsEvery flag the page raised: uid (F1, F2, ... in severity, file, then line order), id (SP-01..SP-21), sev (high | medium | low | info), file, line, text, detail, and lines (every line an aggregated flag covers, only when it covers more than one). An audit must answer every non-info flag by its uid; a fix must close every high and medium flag or name it in open.
verdict_candidateThe audit verdict the flags point at (the page’s own reading): blocked with any high flag, needs_work with any medium, otherwise ready.
notesPlain sentences about what was not read, not sent, or cut from these facts (empty for a paste that fits).

The simplest source is sparkkit.js itself: the page loads it from /sparkkit.js, and it also runs in Node (it sets module.exports). This script prints what the page would send as code and facts (with the 60 000-character cut applied), with the facts decoded so you can read them; the step 4 samples compact them back into the page’s string byte for byte. Save the paste as orders-paste.txt and redirect the output into orders-check.json: 

// build-check.js -- run: node build-check.js orders-paste.txt > orders-check.json   (sparkkit.js saved next to it)
const fs = require("fs");
const Sparkkit = require("./sparkkit.js");
const A = Sparkkit.analyze(fs.readFileSync(process.argv[2], "utf8"));
if (!A.ok) { console.error(A.errors.join("\n")); process.exit(1); }
const inp = Sparkkit.buildInput("audit", A, "", null);       // code and facts do not depend on the lane
console.log(JSON.stringify({ code: inp.code, facts: JSON.parse(inp.facts) }, null, 2));
const real = A.flags.filter((f) => f.sev !== "info");
console.error(A.files.length + " file(s), " + A.functions.length + " function(s), " + A.dataframes.length + " DataFrame(s), " + A.reads.length + " read(s), " +
  A.writes.length + " write(s), " + A.actions.length + " action(s), " + A.joins.length + " join(s), " + A.conf.length + " setting(s); " +
  real.length + " flags (" + A.counts.high + " high, " + A.counts.medium + " medium, " + A.counts.low + " low); verdict_candidate " + A.candidate);

For the orders example it reports 3 file(s), 2 function(s), 4 DataFrame(s), 2 read(s), 1 write(s), 2 action(s), 1 join(s), 5 setting(s); 7 flags (2 high, 5 medium, 0 low); verdict_candidate blocked; for the churn example, 1 file(s), 3 function(s), 3 DataFrame(s), 1 read(s), 1 write(s), 1 action(s), 0 join(s), 1 setting(s); 6 flags (0 high, 4 medium, 2 low); verdict_candidate needs_work. For both worked examples its code and facts are exactly what the page sends.

Without the page’s check. The platform only checks that facts is a string, so a minimal hand-written object works — "{}", or one such as {"job": {"spark_version": "3.5.1"}, "files": [{"path": "jobs/orders_nightly.py", "kind": "python", "line": 2, "end_line": 52}], "flags": [], "notes": ["No page check was run; read the code directly."]} — and the model still reads code. But you lose what the check adds: with no flags there is nothing to confirm, dismiss or close, with no files there are no paths and line ranges to name, with no spark_version the model has to hedge every version-dependent edit, and with no functions, reads and writes nothing says what must survive a fix. Build it with sparkkit.js as above when you can.

The audit_handoff string

Only for task: "fix", and only for the code the audit was about with every audit edit applied. The page builds both from the normalised audit reply res (recon.js normalize, below) and the analysis A of the audited paste, in recon.js fixHandoff: 

const run = Recon.applyFindings(A, res.findings, Sparkkit);   // every finding's edits, in order, each on top of the last
JSON.stringify({ verdict: res.verdict, headline: res.headline,
  findings: res.findings.map(f => ({ id: f.id, severity: f.severity, file: f.file, line: f.line, title: f.title, why: f.why, edits: f.edits.length })),
  applied: /* edits that applied */, failed: /* edits that did not */ })
// run.A.text is the edited paste: it becomes code, and facts is recomputed from it

That is {"verdict", "headline", "findings": [{id, severity, file, line, title, why, edits}], "applied", "failed"}, where each finding’s edits is the number of edits it carried. applyFindings applies every finding’s edits with Sparkkit.applyEdits, in the order listed, to the finding’s file: each before must occur exactly once in that file as the earlier edits left it (leading and trailing blank lines ignored), or that edit is reported as failed and skipped — it is never guessed. The handoff is a previous reply, not evidence: the model checks it against code and fixes what is left. This script does the whole step — the edited paste, its recomputed facts and the handoff string, in the same shape as build-check.js prints, so the samples read it the same way: 

// handoff.js -- run: node handoff.js orders-paste.txt orders-reply.txt > orders-fix-check.json   (sparkkit.js, recon.js next to it)
const fs = require("fs");
const Sparkkit = require("./sparkkit.js");
const Recon = require("./recon.js");
const A = Sparkkit.analyze(fs.readFileSync(process.argv[2], "utf8"));          // the audited paste
const res = Recon.normalize(Recon.parseResult(fs.readFileSync(process.argv[3], "utf8")), "audit");
const ho = Recon.fixHandoff(A, res);                                            // every edit, in order
const B = Sparkkit.analyze(ho.text);                                            // facts recomputed from the edited paste
const inp = Sparkkit.buildInput("fix", B, "", ho.handoff);
console.log(JSON.stringify({ code: inp.code, facts: JSON.parse(inp.facts), audit_handoff: inp.audit_handoff }, null, 2));
console.error(ho.applied + " edit(s) applied, " + ho.failed + " did not apply");

Save the audit’s output.output as orders-reply.txt; for the orders audit above it reports 9 edit(s) applied, 0 did not apply, the edited paste raises only the SP-21 context flag, and the page sends that fix run with the audit’s note and the key skewcheck:fix:54ffeb44-260b:a1.

The output contracts

The model is instructed (SKILL.md) to reply with one JSON object and nothing else — no prose, no code fences — with every array of its lane present ([] when empty). The run’s output text is that object. In both lanes it must ground everything in code: data sizes, row counts, key skew, file counts and the cluster are not in the code, so they are never stated as facts unless the note gives them (“if dim_pages is under a few hundred MB ...”, and a question); what lives outside the paste is asked about, not assumed. It must never write a secret value: a line that holds a literal credential is never copied into the reply (not in evidence, not in an edit, not in a file) — the value is read at run time instead, through the mechanism the paste or the note names or an environment variable with a clear name such as os.environ["CRM_DB_PASSWORD"] — and no credential, hostname or bucket is invented. It must keep the job’s results: which rows, columns and values it produces, where it writes them, in which format and under which partition columns, unless that is the finding and the note says what readers need. It never invents a column, a type or a table, never picks a JDBC split column or a window key the note or the code does not give, and applies version-dependent advice from facts.job.spark_version. Every page flag whose sev is not info is answered exactly once (audit lane: in flag_answers), and an info flag is never reported as a finding.

task: "audit"

task: "fix"

How the page reads a reply (recon.js parseResult and normalize): it strips a code fence and parses from the first { to the last }. A reply is read as the lane that was asked for, unless it names the other lane and carries that lane’s body (a files array for fix, a findings array for audit). An unknown verdict shows as needs_work or partial and counts as a disagreement; an unknown severity becomes medium; ids like SP7 or sp 7 become SP-07; flags keeps only F<n> values; findings with no file, evidence and title are dropped; job may also arrive as service, file as path, a change’s function as func and its change as what, and an open item as path; questions is capped at 10, plan at 30 and verify at 20. An audit reply carries 9 sections (verdict, headline, job, areas, findings, flag_answers, plan, questions, summary) and a fix reply 8 (verdict, headline, files, changes, verify, open, questions, summary), which is what the page counts when it recovers a partial reply.

The page checks the reply against the code; the API does not

What comes back from the API is the model’s raw text in output.output. The page holds it to account in the browser before showing it (recon.js reconcile, with the facts, the code and the handoff it sent): it applies every edit or returned file to the paste and re-runs its own checks on the result. None of that happens in the API. A direct API caller gets the reply unchecked and must do the applying and re-checking itself.

recon.js also runs in Node, so you can hold a reply to exactly the page’s standard. Save /sparkkit.js and /recon.js next to this script, pass the file you built the body from (build-check.js or handoff.js output), the reply text and the lane; it prints every statement, the count, and the paste with every edit (or file) put back: 

// recheck.js -- run: node recheck.js orders-check.json orders-reply.txt audit > orders-edited.txt   (sparkkit.js, recon.js next to it)
const fs = require("fs");
const Recon = require("./recon.js");                                            // loads ./sparkkit.js itself
const sent = JSON.parse(fs.readFileSync(process.argv[2], "utf8"));             // code, facts (and audit_handoff) you sent
const lane = process.argv[4] || "audit";
const res = Recon.normalize(Recon.parseResult(fs.readFileSync(process.argv[3], "utf8")), lane);  // output.output
const r = Recon.reconcile(res, { facts: sent.facts, code: sent.code, handoff: sent.audit_handoff || null });
r.notes.forEach((n) => console.error(n.kind + ": " + n.text));
console.error(r.disagreements + " disagreement(s) in " + r.statements + " statements checked");
if (r.after) (r.after.flags || []).forEach((f) => console.error("still raised: " + f.id + " " + f.sev + " " + f.file + ":" + f.line + " " + f.text));
if (r.after && r.after.text) console.log(r.after.text.replace(/\n$/, ""));      // the paste with every edit (or file) put back

For the orders audit reply it reports:

ok: The areas table rates the driver, shuffle, compute and io once each.
ok: Every one of the 7 findings names a pasted file, a line inside it and evidence copied from it.
ok: The verdict (blocked) fits the worst finding.
ok: Every one of the page's 7 flags is answered once.
ok: All 9 edits applied in order, each before text found exactly once.
ok: Every confirmed flag a finding carries is gone once all edits are applied.
ok: Every function of the paste is still there after the edits.
ok: Every read and write of the paste is still there after the edits (2 reads, 1 write).
ok: The reply repeats no credential from the paste.
0 disagreement(s) in 9 statements checked

and for the churn fix reply (node recheck.js churn-check.json churn-reply.txt fix):

ok: 1 pasted file returned whole.
ok: With the fixed files in place the free check raises no high or medium flag beyond the ones left open (2 low).
info: F2 left open by the reply.
ok: Every one of the 3 functions is still there.
ok: Every read and write of the paste is still there (1 read, 1 write).
ok: Every one of the 3 changes names a returned file, a function in the fixed code and real page flags.
ok: The verdict (partial) fits the open list.
ok: The reply repeats no credential from the paste (1 found in it).
0 disagreement(s) in 7 statements checked
still raised: SP-15 medium jobs/churn_features.py:27 JDBC read of public.customers through one connection
still raised: SP-19 low jobs/churn_features.py:12 spark.executor.cores is 8
still raised: SP-13 low jobs/churn_features.py:36 customers is cached but used once afterwards

The page does not run the job. After that, run the old and the new job on one day or a sample, compare their outputs (row count and a checksum of the key columns), read the new plan with df.explain("formatted") and the Spark UI’s stages (duration, shuffle read, spill, the slowest task against the median), plus whatever the reply’s plan or verify names, before you run it on the full data.

Step 1 — a tiny client helper

One function per language that does what sdk.js does for every call: JSON in, bearer token if you have one, data out on a 2xx, an error carrying status, error.code and error.message otherwise. The later steps assume this helper is in the same file (for Go, Java and C#, the later snippets go inside main / the top-level statements; Java’s makeInput from step 4 goes in the class). 

BASE="https://api.skillsafe.ai/v1/app-api"
TOKEN="${SKILLSAFE_TOKEN:-YOUR_TOKEN}"   # step 2

# api METHOD PATH [extra curl args...] -- prints the whole envelope; pipe to jq
api() {
  method="$1"; path="$2"; shift 2
  curl -sS -X "$method" "$BASE$path" \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer $TOKEN" "$@"
}
# success: jq '.data'      failure: jq '.error | {code, message}'

Step 2 — a token

Two kinds work:

# personal token (needed for a run): copy it from /tokens.html, then
#   export SKILLSAFE_TOKEN=...        and skip the rest of this block
# guest token (enough for /me and /estimate): no Authorization header on this one call
TOKEN=$(curl -sS -X POST "$BASE/guest" \
  -H "Content-Type: application/json" \
  -d '{"slug":"skewcheck"}' | jq -r '.data.token')
echo "guest token: ${TOKEN:0:12}..."

Step 3 — GET /me

Who the token belongs to and what it can spend. sdk.js documents the shape as {subject_type, subject_id, credits, profile?}. The page treats the caller as signed in only when subject_type is "user", shows credits as the balance (1 credit = $0.0001, so it divides by 10 000 for the dollar figure), and compares it with the estimate in step 4 before it lets you run. 

api GET /me | jq '.data // .error'

Step 4 — build the input, then POST /estimate

First a small makeInput that does what buildInput does: task, then code as the paste the page would send, facts as a compact JSON string, the note when it is not blank, and — for the fix lane only — the audit_handoff string when the file carries one. It reads the file build-check.js (or handoff.js) printed. The samples build the audit input of the orders example from orders-check.json; step 6 reuses makeInput for the churn (fix) example. Then price it. The body of /estimate is the same input you will send to /run; no job is created and nothing is charged — /estimate is free. From the response the page reads:

FieldMeaning
hold_creditsA reservation, not a price: the worst case the platform holds while the run is in flight. The page shows it as “reserves N credits ($...) for the audit run” and adds that you are charged only for what the run actually uses, usually far less. What you pay is the job’s charged_credits after settlement (step 5). A balance below the hold is allowed, but the page warns that the reply may be cut short (truncated: true).
min_creditsThe smallest balance that can start a run. Below it the page disables the run button and names the shortfall, because the call would be a 402.
sponsor_enabledWhen true, a caller who is not signed in may run; otherwise the page asks them to sign in.
warningsInput warnings. They do not refuse the call; the page shows them next to the price (“input warnings: ...”), so check them before you run.

The page re-prices whenever the paste, the note or the lane change. The amounts grow with the size of code and facts, so read them from the call rather than hard-coding them. 

# orders-check.json: printed by build-check.js (see "The facts string").
# (.facts | tojson) turns the facts into ONE compact string, as JSON.stringify does.
BODY=$(jq -c --arg note "Nightly job over a day of orders; the driver died last night once customers passed 40 million rows. order_seq is meant to number each customer's orders by time. Downstream reads the whole output directory with Spark, so one file is not needed. The CSV header is order_id,customer_id,sku,quantity,unit_price,order_ts - order_id, customer_id and sku are strings, quantity is an integer, unit_price a decimal(10,2), order_ts an ISO-8601 timestamp. What do we fix first?" \
  '{task: "audit", code, facts: (.facts | tojson), note: $note}' orders-check.json)
# No page check at hand? facts: "{}" is accepted as a string, but the reply then has no page flags to answer.
# Fix lane after an audit: read handoff.js's output instead and add its string:
#   '{task: "fix", code, facts: (.facts | tojson), note: $note} + (if .audit_handoff then {audit_handoff} else {} end)'

api POST /estimate -d "$BODY" \
  | jq '.error // (.data | {hold_credits, min_credits, sponsor_enabled, warnings})'

Step 5 — POST /run, then poll the job (audit)

Send the same input to /run with an Idempotency-Key header, using a personal token. Its data carries a job_id. Poll GET /jobs/{job_id} the way sdk.js waitForJob does: every second until status is succeeded or failed, giving up on the client side after 180 seconds. On the finished job, read:

Idempotency key. Resending a request with the same key must not start a second paid run, so derive the key from the input, not from the clock — and include the lane, so an audit and a fix run of the same paste never share a key. The page uses skewcheck:<lane>:<inputHash>:a<attempt> (app.js run), where <lane> is audit or fix, <inputHash> is a 32-bit djb2 hash of JSON.stringify(input) in hex, a dash, and the string’s length in hex (app.js inputHash, over the input without retry_note), and <attempt> is 1 for a normal run and 2 for the automatic reformat retry (which also adds retry_note). For the two worked examples that is skewcheck:audit:d6d7adf6-251b:a1 (orders audit) and skewcheck:fix:51b16355-1945:a1 (churn fix); the fix run of the orders audit handed on (below) is skewcheck:fix:54ffeb44-260b:a1. The format is the page’s convention, not a requirement: any stable, unique string does. The samples below use the lane and a SHA-256 prefix.

The page itself uses /run-stream (step 6), not /run; both take the same body and header. 

KEY="skewcheck:audit:$(printf '%s' "$BODY" | shasum -a 256 | cut -c1-16):a1"

JOB=$(api POST /run -H "Idempotency-Key: $KEY" -d "$BODY" | jq -r '.data.job_id')

for i in $(seq 180); do                       # 1 s interval, 180 s client-side limit
  J=$(api GET "/jobs/$JOB")
  S=$(printf '%s' "$J" | jq -r '.data.status')
  if [ "$S" = "succeeded" ] || [ "$S" = "failed" ]; then break; fi
  sleep 1
done

printf '%s' "$J" | jq '.data | {status, truncated, charged_credits, error}'
printf '%s' "$J" | jq -r '.data.output.output' \
  | jq '{verdict, headline, areas: [(.areas // [])[] | {area, status}],
         findings: [(.findings // [])[] | {id, file, line, severity, edits: ((.edits // []) | length)}],
         flag_answers: [(.flag_answers // [])[] | {flag, status}]}'

Step 6 — POST /run-stream (Server-Sent Events, fix)

The alternative to step 5, and what the page uses: same kind of body, same Idempotency-Key header, one request. Run one or the other for a given input, not both. To show the second lane, the samples build the fix input of the churn example (with its own key) and stream it; to stream the audit input instead, pass the step 4 input and key. The response is text/event-stream. sdk.js splits it into frames on a blank line, takes the event name from an event: line (default message) and concatenates the data: lines, skips frames with no data or with data that is not JSON, and handles these events:

EventDataWhat the SDK does
delta{text}Passes text to onDelta.
jobjob infoPasses it to onJob (the page uses it to advance its progress display).
donethe final payload: {job_id, status, charged_credits, output}Keeps it as the result.
pendinga payloadTreated exactly like done. If it carries a job_id and no output, poll that job as in step 5.
error{message, code, job_id}After the stream ends, raises an error with those fields.

If the response is not text/event-stream, it is an ordinary JSON envelope: the SDK returns its data (an idempotent replay returns this way) or raises the error on a non-2xx status. Handle both shapes.

From the final payload the page reads output.output (falling back to the concatenated delta text), truncated and charged_credits. Browsers receive progress ticks rather than deltas from this endpoint, so the page relies on the final payload; if the stream fails part-way, it tries to recover a partial reply from whatever delta text arrived. The samples print the verdict and headline, each returned file’s path and size, each change (function, flags and change), each verification step, each open item and each question. To put the files into the paste and re-check them the way the page’s “Put every file into the paste” does, keep the reply text and run recheck.js with the lane fix. To fix code you audited, read handoff.js’s output instead of churn-check.json: makeInput then carries its audit_handoff. 

# the fix worked example: churn-check.json from build-check.js
# (for audited code, read handoff.js's output instead; the + (...) adds its audit_handoff)
FBODY=$(jq -c --arg note "Fix what you can. Leave the JDBC read's partitioning exactly as it is - the DBA has to approve a split column first. The password must come from the CRM_DB_PASSWORD environment variable the scheduler sets. Keep printing the per-plan counts." \
  '{task: "fix", code, facts: (.facts | tojson), note: $note} + (if .audit_handoff then {audit_handoff} else {} end)' \
  churn-check.json)
FKEY="skewcheck:fix:$(printf '%s' "$FBODY" | shasum -a 256 | cut -c1-16):a1"

# -N turns off buffering; the whole stream lands in $STREAM
STREAM=$(api POST /run-stream -N -H "Idempotency-Key: $FKEY" -d "$FBODY")

# frames look like:
#   event: job
#   data: {...}
#
#   event: done
#   data: {"job_id":"...","status":"...","charged_credits":...,"output":{"output":"{\"task\":\"fix\",...}"}}

# the final reply text, from the done (or pending) frame:
REPLY=$(printf '%s\n' "$STREAM" \
  | awk '/^event:/{ev=$2} /^data:/ && (ev=="done" || ev=="pending"){sub(/^data: */,""); print}' \
  | jq -r '.output.output')
# an idempotent replay answers with a plain JSON envelope instead: REPLY=$(printf '%s' "$STREAM" | jq -r '.data.output.output')
printf '%s' "$REPLY" | jq '{verdict, headline,
  files: [(.files // [])[] | {path, characters: (.content | length)}],
  changes: [(.changes // [])[] | {function, flags, change}], verify, open, questions}'

What the page does that you now do

Or skip all of this and use Skewcheck in the browser: the check, the flags, the read, write, action, join and settings inventories, putting an edit or the fixed files back into the paste, and the Markdown, CSV and JSON exports are free.