tao.chen/opencode-build
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

feat: spark mcp skill

Browse Source
This commit is contained in:
tao.chen
2026-06-30 16:51:00 +08:00
parent 43e5bbbf03
commit 0fc1c702d9
1 changed files with 557 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/spark-executor-mcp-operate/SKILL.md
+557
View File
@@ -0,0 +1,557 @@
---
name: spark-executor-mcp-operate
description: Operation guide for the spark-executor-mcp service — 16 MCP tools that submit, monitor, fetch logs from, and kill PySpark jobs on YARN, plus connection-profile and script-file management. USE THIS SKILL whenever the user wants to do anything with the spark-executor-mcp MCP service: submit a Spark/PySpark job to YARN, check job status or result, fetch container logs, kill a running job, save/list/get/delete a YARN connection, or write/read/update a PySpark script file. Also use it when the user asks generally about Spark, YARN, pyspark, cluster jobs, or distributed compute via this service. Do NOT guess tool names or invent a submit flow — read this skill first to learn the two-step prepare/confirm flow, the dual-ID contract (job_id vs application_id), and the SQL guard / path restrictions. The MCP service does NOT generate PySpark code — the calling LLM writes the code, then uses `write_job_file` to persist it on disk before submitting.
---
# spark-executor-mcp 操作指南
Spark-executor-mcp is an MCP (Model Context Protocol) service that exposes
**16 tools** for managing PySpark jobs on YARN. It runs as a FastAPI app
mounted at `/spark-executor-mcp`; clients (typically LLM agents) call the
tools via the standard MCP `tools/call` flow.
This skill is the canonical reference for an LLM agent using the service.
It does **not** describe implementation details — for that, see the source
in `spark_executor/`.
---
## 0. Mindset before you start
Three things you MUST internalize before calling any tool:
1. **The submit flow is two-step: `prepare_submit_job` then `confirm_submit_job`.**
Prepare is cheap and reversible (just snapshots params + writes JSON to
disk). Confirm is irreversible — it actually invokes `spark-submit` and
a YARN application starts running. Never skip the prepare step, even
when the user is in a hurry.
2. **Every Job has two IDs that are NOT interchangeable everywhere.**
- `job_id` — local, 12-char hex (`a1b2c3d4e5f6`), generated by the
service. Used for: looking up the Job record, all MCP tool calls.
- `application_id` — YARN's own ID (`application_17400000001_0001`).
Used for: YARN UI, `tracking_url`, the YARN REST API.
The service's 4 job-lifecycle tools accept **either** ID (after the
2026-06-29 fix), but other systems (YARN UI, kubectl-style scripts)
only know `application_id`. **Always store both** from every
`confirm_submit_job` response.
3. **The LLM writes the PySpark code. The MCP service writes the file.**
`write_job_file` is a file writer with SQL-safety checks, not a
code generator. You compose the code in your context, then call
`write_job_file(code=...)` to drop it on the container's
filesystem where `spark-submit` can see it.
4. **Three decisions MUST be approved by the human user before you act on
them — never pick defaults silently, never auto-confirm.** See §0.1
below. This is non-negotiable: confirm_submit_job starts a real YARN
job that costs cluster time and may write to production data paths,
and the user is the one who has to live with the result. Get their
explicit "yes, confirm" before pulling the trigger.
### 0.1. Mandatory user confirmations (the three checkpoints)
For every Spark job the user asks you to run, you must stop and ask the
human user to approve these three decisions in order. Do not pick
defaults on their behalf. Do not assume "they'd obviously want prod /
default queue / 2G / 2 cores / 2 executors" — those are YOUR defaults,
not theirs.
**Checkpoint 1 — when the user first asks to run a job:**
ASK the user for:
- Which **connection** to submit to. Show the existing connections
(from `list_connections`) as options; if the right one doesn't exist,
ask for the new connection's full profile (name, master, deploy_mode,
yarn_rm_url, spark_conf, auth). See §5 for the field reference.
- The **app_name** (human-readable label for the YARN app). The
service has no default for this; without it `prepare_submit_job`
returns 400.
Do not skip past this even if the user is in a hurry or the request
seems obvious. "Run a daily job on prod" still needs you to confirm
which "daily job", which "prod", and what app_name to label it with.
**Checkpoint 2 — before `prepare_submit_job`:**
ASK the user for the values of these submit parameters, even if they
have server-side defaults:
- `app_name` (if you didn't get it at Checkpoint 1)
- `queue`
- `executor_memory`
- `executor_cores`
- `num_executors`
- any `extra_args` (jars, py-files, …) needed for this script
Tell the user the server-side defaults you WOULD use if they don't care
(say "I'll use default queue, 2G memory, 2 cores, 2 executors unless
you say otherwise") so they can either approve or override. The
service itself rejects implicit defaults (returns 400 listing the
assumed values) — this is by design, and you should respect it by
being the one to surface the assumed values, not by silently letting
the service 400 and then re-asking.
**Checkpoint 3 — before `confirm_submit_job` (the irreversible one):**
Present a one-screen summary of exactly what will happen, and wait for
the user to say "confirm" / "yes" / "do it" (or any unambiguous
affirmative). The summary must include:
- The connection name and its master / deploy_mode / yarn_rm_url
- The app_name and the script path
- queue, executor_memory, executor_cores, num_executors
- The YARN tracking URL that will be returned (you don't know it yet,
but you can say "you'll get a tracking URL once it starts")
The user's exact words don't need to be "confirm" — any clear
affirmative works ("go", "ship it", "y", "🚀", "do it"). A non-answer
or a "wait" / "hold on" / "let me think" is NOT consent; wait.
After they say yes, you may call `confirm_submit_job`. **Never call
`confirm_submit_job` without that explicit go-ahead.**
---
## 1. Tool index
| Operation ID | HTTP | Group | Purpose |
|---|---|---|---|
| `save_connection` | POST | Connection | Upsert a named YARN connection (master, deploy_mode, spark_conf, …) |
| `list_connections` | POST | Connection | List all saved connections |
| `get_connection` | POST | Connection | Fetch one connection by name |
| `delete_connection` | POST | Connection | Delete a connection by name |
| `write_job_file` | POST | Job file | Write LLM-written PySpark code to disk |
| `read_job_file` | POST | Job file | Read a script file back (capped at 1 MB) |
| `update_job_file` | POST | Job file | Overwrite a script file (capped at 1 MB) |
| `prepare_submit_job` | POST | Submit | Snapshot a job to `PendingSubmission`; **does NOT submit** |
| `confirm_submit_job` | POST | Submit | Actually invoke `spark-submit` for a `pending_id` |
| `update_pending_job` | POST | Submit | Edit a `PENDING` submission (script_path, queue, …) |
| `cancel_pending_job` | POST | Submit | Cancel a `PENDING` submission |
| `list_pending_jobs` | POST | Submit | List every PendingSubmission regardless of status |
| `get_pending_job` | POST | Submit | Fetch one PendingSubmission by id |
| `get_job_status` | POST | Job lifecycle | Query YARN for a job's live state (RUNNING / SUCCEEDED / …) |
| `get_job_result` | POST | Job lifecycle | Terminal-oriented view: final_status, diagnostics, timing |
| `get_job_logs` | POST | Job lifecycle | Aggregated container logs (default last 5000 chars) |
| `kill_job` | POST | Job lifecycle | PUT state=KILLED to YARN REST for the app |
(Total: 16 MCP tools + `/health` GET, which is not exposed as a tool.)
---
## 2. The canonical happy path: submit a PySpark job
```
-- CHECKPOINT 1: ask user for connection + app_name --
1. list_connections / save_connection # pick or create the right profile
2. write_job_file # write the LLM's PySpark to disk
3. read_job_file # (optional) sanity-check what was written
-- CHECKPOINT 2: ask user for queue / memory / cores / num_executors --
4. prepare_submit_job # snapshot + persist; no YARN call yet
-- CHECKPOINT 3: present summary, wait for explicit "yes, confirm" --
5. confirm_submit_job # NOW spark-submit runs (irreversible)
6. get_job_status / logs # poll while running
7. get_job_result # terminal view when done
```
The three `CHECKPOINT` lines above are mandatory pauses. Do NOT
collapse them or skip past them. See §0.1 for the full rules; the
short version is: ask the user before each tool call that affects
cluster state, and never call `confirm_submit_job` without their
explicit go-ahead.
### 2.1 Save a connection (one-time per cluster)
```python
save_connection(
name="prod",
master="yarn",
deploy_mode="cluster",
yarn_rm_url="http://rm.example.com:8088",
spark_conf={"spark.sql.shuffle.partitions": "200"},
# optional: auth_type="kerberos", ssl_verify=False, ...
)
```
**`name`** is the lookup key — keep it short and stable. Re-calling
`save_connection` with the same `name` overwrites; the original `yarn_rm_url`
and `spark_conf` are **snapshot into the PendingSubmission at prepare time**
and survive later edits to the connection (this is intentional — the user
can re-point the connection at a different cluster without retargeting
in-flight submissions).
### 2.2 Write the script to disk
The LLM writes the PySpark code. Then:
```python
script_path = write_job_file(code=PYSPARK_SOURCE)["script_path"]
# Returns {"script_path": "/absolute/path/in/container.py", "bytes": N}
```
The path lives inside the container's `SPARK_EXECUTOR_JOBS_DIR` (default
`./data/jobs/`). For pre-existing scripts on the host, mount the host
directory into the container and skip this step — pass the in-container
path directly to `prepare_submit_job`.
### 2.3 Prepare (snapshot, no YARN call)
```python
prepare_submit_job(
connection="prod",
script_path=script_path,
app_name="daily-aggregation", # REQUIRED, no default
queue="default", # these 4 must be explicit
executor_memory="2G",
executor_cores=2,
num_executors=2,
# extra_args={"jars": "hdfs://...jar"}, # optional
)
# Returns {"pending_id": "p_a1b2c3", "status": "PENDING", "parameters": {...}}
```
**You MUST pass `queue`, `executor_memory`, `executor_cores`, `num_executors`
explicitly every time** — even though defaults exist server-side, the
service rejects implicit defaults with a 400 listing the assumed values.
This forces the agent (or the human reviewer) to see and approve what
will actually be submitted.
The response is the only chance to capture the `pending_id` you need for
`confirm_submit_job`. Store it.
### 2.4 (Optional) Edit before submit
If something needs to change between prepare and confirm:
```python
update_pending_job(
pending_id="p_a1b2c3",
queue="research",
# script_path can also be changed — but the new file MUST exist
# and pass the SQL guard
)
```
Refuses to edit anything that is no longer `PENDING`.
### 2.5 Confirm (the irreversible step)
```python
result = confirm_submit_job(pending_id="p_a1b2c3")
# Returns {"job_id": "a1b2c3d4e5f6",
# "application_id": "application_17400000001_0001",
# "tracking_url": "http://rm:8088/proxy/application_17400000001_0001/"}
```
**STORE BOTH IDS NOW.** Future calls to `get_job_status` / `get_job_logs`
/ `kill_job` accept either, but `application_id` is the one YARN itself
recognizes.
`confirm_submit_job` is **idempotent** on `SUBMITTED` (re-calling returns
the same record) and **retries** on transient `SparkSubmitError` (3
attempts with 5 s delay by default). It transitions to `FAILED` only
after all retries are exhausted.
### 2.6 Monitor / fetch / kill
```python
# While running
state = get_job_status(job_id="a1b2c3d4e5f6").state
logs = get_job_logs(job_id="a1b2c3d4e5f6", tail_chars=10000)
# When done
result = get_job_result(job_id="a1b2c3d4e5f6")
# result.state -> "SUCCEEDED" / "FAILED" / "KILLED" / ...
# result.final_status -> "SUCCEEDED" / "FAILED" / "KILLED" / "UNDEFINED"
# result.diagnostics -> YARN's "why" string (often empty on success)
# result.tracking_url -> the same proxy URL confirm_submit_job returned
# To stop it
kill_job(job_id="a1b2c3d4e5f6")
```
All four accept **either** `job_id` or `application_id`. Pass whichever
you have on hand; the service tries `job_id` first, falls back to
`application_id`.
---
## 3. The PendingSubmission state machine
```
prepare_submit_job
|
v
+-------+ cancel_pending_job
|PENDING+----------> CANCELLED (terminal)
+---+---+
| confirm_submit_job
v
+----------+
| SUBMITTED| (terminal — kill via kill_job, not cancel)
+----------+
(or)
+--------+
| FAILED | <-- confirm_submit_job exhausted retries
+--------+
| confirm_submit_job (resets to PENDING, retries)
v
PENDING
```
- `cancel_pending_job` **refuses** to cancel `SUBMITTED` or `FAILED`
records — those are terminal; use `kill_job` instead.
- `update_pending_job` **refuses** to edit anything that is not
`PENDING`.
- `confirm_submit_job` on a `SUBMITTED` record is a no-op (returns the
same `job_id` / `application_id`). On a `FAILED` record, it resets
to `PENDING` and retries. On a `CANCELLED` record, it raises 400.
---
## 4. The dual-ID contract
| ID | Where it comes from | What it's good for |
|---|---|---|
| `job_id` | service-generated, 12-char hex (`a1b2c3d4e5f6`) | Looking up the Job in this service's store; all 4 lifecycle tools |
| `application_id` | parsed from `spark-submit` stderr (`application_<ts>_<n>`) | YARN UI, YARN REST, `tracking_url`, talking to ops scripts |
**Both IDs are returned by `confirm_submit_job`.** Both are accepted by
`get_job_status` / `get_job_result` / `get_job_logs` / `kill_job`. **You
don't need to remember which to pass — but DO remember to store both
from the confirm response** because:
- If the Job store has been wiped (rare; only on data dir loss), only
`application_id` is still meaningful to YARN — you can't poll a YARN
app you can't identify.
- If you're going to paste a link to a teammate, `tracking_url` (derived
from `application_id`) is the human-friendly form.
If a lifecycle tool returns `KeyError("No Job found for id='X' …")`,
check that `X` came from a successful `confirm_submit_job` response.
The error message intentionally mentions both ID forms and shows an
example of each so you can self-diagnose.
---
## 5. Connection profiles
`Connection` is the cluster profile — it's the "where to submit" record.
Fields:
| Field | Required | Default | Notes |
|---|---|---|---|
| `name` | yes | — | The lookup key |
| `master` | yes | `"yarn"` | `"yarn"` / `"spark://…"` / `"k8s://…"` / `"local[N]"` |
| `deploy_mode` | yes | `"cluster"` | `"cluster"` or `"client"` |
| `yarn_rm_url` | no | env `YARN_RESOURCE_MANAGER_URL` | Required if you'll call `get_job_status` / `get_job_logs` / `kill_job` (the lifecycle tools need a YARN endpoint) |
| `spark_conf` | no | `{}` | Merged into `spark-submit --conf` at submit time |
| `ssl_verify` | no | env `SPARK_EXECUTOR_SSL_VERIFY_DEFAULT` | bool |
| `ssl_ca_bundle` | no | env `SPARK_EXECUTOR_SSL_CA_BUNDLE_DEFAULT` | Path to CA file |
| `auth_type` | no | `"none"` | `"none"` / `"simple"` / `"basic"` / `"kerberos"` |
| `auth_user` / `auth_password` | for `basic` | — | |
| `auth_principal` / `auth_keytab` | for `kerberos` (display only) | — | Real auth uses the system cache |
**Snapshot semantics:** when `prepare_submit_job` runs, it copies
`master` / `deploy_mode` / `spark_conf` / `yarn_rm_url` from the
**current** Connection into the `PendingSubmission`. Editing the
Connection later does **not** retarget an in-flight submission.
**Lifecycle tools use the Job's snapshotted `yarn_rm_url`** (set at
confirm time) — not the Connection's current value. If the RM moves
between confirm and a follow-up `get_job_logs`, you have a problem;
re-pointing the Connection does not help.
---
## 6. Job file workflow (`write_job_file` / `read_job_file` / `update_job_file`)
These are **filesystem I/O tools, not code generators**. The LLM in
context writes the PySpark code; these tools just persist it where
`spark-submit` can find it, with safety guards.
**Typical lifecycle:**
```
LLM composes code --> write_job_file(code=...) # write
LLM reviews code --> read_job_file(script_path=...) # verify
LLM revises code --> update_job_file(script_path=..., # overwrite
content=...)
LLM submits --> prepare_submit_job(script_path=...)
```
**Guards (these are why the tools exist instead of just `cat > file.sh`):**
- **SQL guard** — `write_job_file` and `update_job_file` both reject
code containing forbidden SQL (DROP, DELETE, UPDATE, TRUNCATE, ALTER,
GRANT, REVOKE, …). The check is conservative; if it false-positives,
rewrite to use only SELECT/INSERT or restructure.
- **Path restriction** — `update_job_file` only writes under
`SPARK_EXECUTOR_JOBS_DIR`. This blocks accidentally overwriting
host-mounted configs or the service's own files.
- **Size cap** — 1 MB on both `read_job_file` and `update_job_file`.
Use `read_job_file` for small scripts; for huge ones, mount the host
directory and skip these tools.
**For pre-existing scripts on the host:** mount the host dir into the
container (e.g. `-v /host/scripts:/app/scripts:ro` in `docker run`) and
pass the in-container path directly to `prepare_submit_job` — no need
to round-trip through the file tools.
---
## 7. Error reference
| Status | When | What to do |
|---|---|---|
| 400 | `prepare_submit_job` without explicit defaults; `cancel_pending_job` on terminal state; `update_pending_job` on non-PENDING; `prepare_submit_job` with missing/non-file `script_path`; SQL policy violation | Read the message — it almost always names the fix. Re-submit with corrected args. |
| 404 | `KeyError` from tool layer: unknown `connection`, `pending_id`, `job_id`, or `application_id` | The ID you sent doesn't match anything in the store. For lifecycle tools, the error message explicitly suggests both ID forms. |
| 422 | Pydantic validation: missing field, wrong type | Usually a typo or missing required arg. The response body lists every field error. |
| 500 | spark-submit subprocess failed and was not a `SparkSubmitError` (e.g. binary missing); YARN REST call failed | The tool logs the full traceback to `data/logs/error/...` — check there. |
The service translates exceptions at the FastAPI layer:
- `KeyError` → 404 (with the KeyError's string as `detail`)
- `ValueError` → 400
- Pydantic validation → 422
So when the server returns 404, the body `detail` field is the same
"Unknown job_id: …" / "Unknown connection: …" string the tool layer
raised.
---
## 8. Common pitfalls
| Pitfall | Why it bites | What to do |
|---|---|---|
| Skipping the prepare step | `confirm_submit_job` requires a `pending_id`; calling it without one returns 400 | Always prepare first |
| Passing implicit defaults | Server returns 400 listing what you didn't confirm | Pass `queue`, `executor_memory`, `executor_cores`, `num_executors` every time |
| `script_path` doesn't exist in the container | `prepare_submit_job` returns 400; the agent generated the path in its own context but the container's filesystem is different | Either call `write_job_file` first, or mount a host dir |
| SQL injection in the script | `write_job_file` / `update_job_file` / `prepare_submit_job` all 400 | Only use SELECT/INSERT; never DROP/DELETE/UPDATE |
| `cancel_pending_job` on a running YARN app | Returns 400 — submissions are terminal once submitted | Use `kill_job(job_id=...)` instead, which talks to YARN REST |
| Re-pointing a Connection and expecting in-flight jobs to follow | They don't — the Job snapshots `yarn_rm_url` at confirm time | Re-issue with a new `prepare_submit_job` if you actually need a different RM |
| Calling lifecycle tools with the wrong ID type | Used to fail; **fixed in the 2026-06-29 update** — both IDs are accepted | Just pass whichever you have; the service tries `job_id` first |
| Auto-confirming without the user's explicit "yes" | Triggers an irreversible YARN job the user may not have wanted — wastes cluster time, may write to wrong data paths | Per §0.1, **always** present a one-screen summary and wait for the user to say "confirm" / "yes" / equivalent before calling `confirm_submit_job`. Silence, "hmm", or "let me think" are NOT consent. |
| Picking connection / queue / executor defaults on the user's behalf | The user knows their cluster's cost / SLA / data layout better than you do; your defaults may not match their intent | Per §0.1, ASK at Checkpoint 1 (connection + app_name) and Checkpoint 2 (queue / memory / cores / num_executors). Surface the defaults you'd use so they can approve or override, don't just pick. |
| "Session not found" at the MCP layer | Multi-worker gunicorn + in-process session state | Out of scope for this skill; the service warns on startup if `GUNICORN_WORKERS>1`; set `GUNICORN_WORKERS=1` or front with a sticky LB |
---
## 9. End-to-end example (with the three checkpoints)
User: "Run a word count on `s3a://my-bucket/books/*.txt` and write the
result back to `s3a://my-bucket/wc/`."
```
# ============================================================
# CHECKPOINT 1 — ask the user for connection + app_name FIRST
# ============================================================
#
# Agent: "I see you have these saved connections:
# - prod (yarn RM http://rm.prod.example.com:8088, cluster, kerberos)
# - dev (yarn RM http://rm.dev.example.com:8088, client, none)
# Which one should I submit to? And what app_name should I label
# the YARN app with?"
# User: "prod, call it wordcount-books"
#
# (If neither connection is right, ask for the full profile and call
# save_connection before continuing — see §2.1 for the fields.)
# 1. Compose the PySpark code IN YOUR CONTEXT
pyspark_code = '''
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("wordcount").getOrCreate()
sc = spark.sparkContext
files = sc.wholeTextFiles("s3a://my-bucket/books/")
words = files.flatMap(lambda f: f[1].lower().split()) \\
.map(lambda w: (w, 1)) \\
.reduceByKey(lambda a, b: a + b)
df = words.toDF(["word", "count"])
df.write.mode("overwrite").parquet("s3a://my-bucket/wc/")
'''
# 2. Persist to disk
script_path = write_job_file(code=pyspark_code)["script_path"]
# 3. Sanity-check (optional but recommended)
contents = read_job_file(script_path=script_path)["content"]
# ... review in context ...
# ============================================================
# CHECKPOINT 2 — ask the user for submit parameters
# ============================================================
#
# Agent: "I'll use these unless you say otherwise:
# queue=default, executor_memory=2G, executor_cores=2,
# num_executors=2. Any overrides (different queue, more memory,
# jars/py-files to add)?"
# User: "use research queue, 4G memory, 2 executors"
#
# (If the user says "defaults are fine", that's still an explicit
# approval — you may then call prepare_submit_job with those values.)
# 4. Prepare (snapshot — no YARN call yet)
pending = prepare_submit_job(
connection="prod",
script_path=script_path,
app_name="wordcount-books",
queue="research",
executor_memory="4G",
executor_cores=2,
num_executors=2,
)
pending_id = pending["pending_id"]
# ============================================================
# CHECKPOINT 3 — present summary, wait for explicit "confirm"
# ============================================================
#
# Agent: "About to run on YARN:
# • connection: prod (yarn, cluster, http://rm.prod.example.com:8088)
# • app_name: wordcount-books
# • script: <script_path>
# • queue: research
# • resources: 2G memory × 2 cores × 2 executors
# This will invoke spark-submit and start a real YARN application.
# Confirm? (y / n)"
# User: "yes, go"
# ─── only NOW may you call confirm_submit_job ───
# 5. Confirm (NOW spark-submit runs)
result = confirm_submit_job(pending_id=pending_id)
job_id = result["job_id"] # STORE
application_id = result["application_id"] # STORE
tracking_url = result["tracking_url"] # STORE
# 6. Poll
import time
while True:
st = get_job_status(job_id=job_id).state
if st in ("SUCCEEDED", "FAILED", "KILLED", "FINISHED"):
break
time.sleep(30)
# 7. Final result
res = get_job_result(job_id=job_id)
print(res.final_status, res.diagnostics)
# If something went wrong and you want to see why:
# logs = get_job_logs(job_id=job_id, tail_chars=20000)
# (If you need to kill the job: ASK THE USER FIRST, then call
# kill_job(job_id=job_id). Don't kill running jobs without an explicit
# "kill it" — the user may have wanted to let it finish.)
```
---
## 10. Reference paths (for the curious)
This skill is the **operation** guide. The architectural / implementation
details live in:
- `CLAUDE.md` — project conventions, persistence layout, two-step submit
rationale, MCP arg pattern, MCP session affinity caveat.
- `docs/superpowers/plans/2026-06-24-spark-executor-mcp.md` — the
original implementation plan with Stage 1 / 2 / 3 breakdown.
- `spark_executor/server.py` — the route + exception-handler layer.
- `spark_executor/tools/requests.py` — the Pydantic body models; useful
to read when the MCP 422 response confuses you.
- `spark_executor/core/job_store.py` — Job persistence; explains the
dual-ID contract and the file-backed JSON layout.