Distributed Job Safety

Distributed Job Safety

Patterns and anti-patterns for concurrent job management with pueue + mise + systemd-run, learned from production failures in distributed data pipeline orchestration.

Scope: Universal principles for any pueue + mise workflow with concurrent parameterized jobs. Examples use illustrative names but the principles apply to any domain.

Prerequisite skills: devops-tools:pueue-job-orchestration, itp:mise-tasks, itp:mise-configuration

---

Self-Evolving Skill: This skill improves through use. If instructions are wrong, parameters drifted, or a workaround was needed — fix this file immediately, don't defer. Only update for real, reproducible issues.

The Nine Invariants

Non-negotiable rules for concurrent job safety. Violating any one causes silent data corruption or job failure.

Full formal specifications: references/concurrency-invariants.md

1. Filename Uniqueness by ALL Job Parameters

Every file path shared between concurrent jobs MUST include ALL parameters that differentiate those jobs.

WRONG:  {symbol}_{start}_{end}.json                    # Two thresholds collide
RIGHT:  {symbol}_{threshold}_{start}_{end}.json         # Each job gets its own file

Test: If two pueue jobs can run simultaneously with different parameter values, those values MUST appear in every shared filename, temp directory, and lock file.

2. Verify Before Mutate (No Blind Queueing)

Before queueing jobs, check what is already running. Before deleting state, check who owns it.

# WRONG: Blind queue
for item in "${ITEMS[@]}"; do
    pueue add --group mygroup -- run_job "$item" "$param"
done

# RIGHT: Check first
running=$(pueue status --json | jq '[.tasks[] | select(.status | keys[0] == "Running") | .label] | join(",")')
if echo "$running" | grep -q "${item}@${param}"; then
    echo "SKIP: ${item}@${param} already running"
    continue
fi

3. Idempotent File Operations (missing_ok=True)

All file deletion in concurrent contexts MUST tolerate the file already being gone.

# WRONG: TOCTOU race
if path.exists():
    path.unlink()        # Crashes if another job deleted between check and unlink

# RIGHT: Idempotent
path.unlink(missing_ok=True)

4. Atomic Writes for Shared State

Checkpoint files must never be partially written. Use the tempfile-fsync-rename pattern.

fd, temp_path = tempfile.mkstemp(dir=path.parent, prefix=".ckpt_", suffix=".tmp")
with os.fdopen(fd, "w") as f:
    f.write(json.dumps(data))
    f.flush()
    os.fsync(f.fileno())
os.replace(temp_path, path)  # POSIX atomic rename

Bash equivalent (for NDJSON telemetry appends):

# Atomic multi-line append via flock + temp file
TMPOUT=$(mktemp)
# ... write lines to $TMPOUT ...
flock "${LOG_FILE}.lock" bash -c "cat '${TMPOUT}' >> '${LOG_FILE}'"
rm -f "$TMPOUT"

5. Config File Is SSoT

The .mise.toml [env] section is the single source of truth for environment defaults. Per-job env overrides bypass the SSoT and allow arbitrary values with no review gate.

# WRONG: Per-job override bypasses mise SSoT
pueue add -- env MY_APP_MIN_THRESHOLD=50 uv run python script.py

# RIGHT: Set the correct value in .mise.toml, no per-job override needed
pueue add -- uv run python script.py

Controlled exception: pueue env set <id> KEY VALUE is acceptable for one-off overrides on stashed/queued tasks (e.g., hyperparameter sweeps). The key distinction: mise [env] is SSoT for defaults that apply to all runs; pueue env set is for one-time parameterization of a specific task without modifying the config file. See devops-tools:pueue-job-orchestration Per-Task Environment Override section.

6. Maximize Parallelism Within Safe Margins

Always probe host resources and scale parallelism to use available capacity. Conservative defaults waste hours of idle compute.

# Probe host resources
ssh host 'nproc && free -h && uptime'

# Sizing formula (leave 20% margin for OS + DB + overhead)
# max_jobs = min(
#     (available_memory_gb * 0.8) / per_job_memory_gb,
#     (total_cores * 0.8) / per_job_cpu_cores
# )

For ClickHouse workloads: The bottleneck is often ClickHouse's concurrent_threads_soft_limit (default: 2 x nproc), not pueue's parallelism. Each query requests max_threads threads (default: nproc). Right-size --max_threads per query to match the effective thread count (soft_limit / pueue_slots), then increase pueue slots. Pueue parallelism can be adjusted live without restarting running jobs.

Post-bump monitoring (mandatory for 5 minutes after any parallelism change):

• uptime -- load average should stay below 0.9 x nproc
• vmstat 1 5 -- si/so columns must remain 0 (no active swapping)
• ClickHouse errors: SELECT count() FROM system.query_log WHERE event_time > now() - INTERVAL 5 MINUTE AND type = 'ExceptionWhileProcessing' -- must be 0

Cross-reference: See devops-tools:pueue-job-orchestration ClickHouse Parallelism Tuning section for the full decision matrix.

7. Per-Job Memory Caps via systemd-run

On Linux with cgroups v2, wrap each job with systemd-run to enforce hard memory limits.

systemd-run --user --scope -p MemoryMax=8G -p MemorySwapMax=0 \
    uv run python scripts/process.py --symbol BTCUSDT --threshold 250

Critical: MemorySwapMax=0 is mandatory. Without it, the process escapes into swap and the memory limit is effectively meaningless.

8. Monitor by Stable Identifiers, Not Ephemeral IDs (INV-8)

Pueue job IDs are ephemeral -- they shift when jobs are removed, re-queued, or split. Use group names and label patterns for monitoring.

# WRONG: Hardcoded job IDs
if pueue status --json | jq -e ".tasks.\"14\"" >/dev/null; then ...

# RIGHT: Query by group/label
pueue status --json | jq -r '.tasks | to_entries[] | select(.value.group == "mygroup") | .value.id'

Full specification: references/concurrency-invariants.md

9. Derived Artifact Filenames Must Include ALL Category Dimensions (INV-9)

When concurrent or sequential pipeline phases produce derived artifacts (Parquet chunks, JSONL summaries, temp files) that share a directory, every filename must include ALL discriminating dimensions -- not just the job-level parameters (INV-1), but also pipeline-level categories like direction, strategy, or generation.

WRONG:  _chunk_{formation}_{symbol}_{threshold}.parquet     # No direction -- LONG glob eats SHORT files
RIGHT:  _chunk_{direction}_{formation}_{symbol}_{threshold}.parquet  # Direction-scoped

Glob scope rule: Cleanup and merge globs must match the filename pattern exactly:

# WRONG: Unscoped glob -- consumes artifacts from other categories
chunk_files = folds_dir.glob("_chunk_*.parquet")

# RIGHT: Category-scoped glob -- only touches this category's artifacts
chunk_files = folds_dir.glob(f"_chunk_{direction}_*.parquet")

Post-merge validation: After merging artifacts, assert expected values in category columns:

merged_df = pl.concat([pl.read_parquet(p) for p in chunk_files])
assert set(merged_df["strategy"].unique()) == {"standard"}, "Direction contamination!"

Relationship to INV-1: INV-1 ensures checkpoint file uniqueness by job parameters (runtime isolation). INV-9 extends this to derived artifacts that persist across pipeline phases (artifact isolation). Both prevent the same class of bug -- silent cross-contamination from filename collisions.

Full specification: references/concurrency-invariants.md

---

Anti-Patterns (Learned from Production)

17 anti-patterns documented from production failures. Full details with code examples: references/anti-patterns.md

---

The Mise + Pueue + systemd-run Stack

Full architecture diagram and responsibility boundaries: references/stack-architecture.md

---

Remote Deployment Protocol

When deploying a fix to a running host:

1. AUDIT:   ssh host 'pueue status --json' -> count running/queued/failed
2. DECIDE:  Wait for running jobs? Kill? Let them finish with old code?
3. PULL:    ssh host 'cd ~/project && git fetch origin main && git reset --hard origin/main'
4. VERIFY:  ssh host 'cd ~/project && python -c "import pkg; print(pkg.__version__)"'
5. UPGRADE: ssh host 'cd ~/project && uv pip install --python .venv/bin/python --refresh pkg==X.Y.Z'
6. RESTART: ssh host 'pueue restart <failed_id>' OR add fresh jobs
7. MONITOR: ssh host 'pueue status --group mygroup'

Critical: Step 1 (AUDIT) is mandatory. Skipping it is the root cause of cascade failures.

See: references/deployment-checklist.md for full protocol.

---

Concurrency Safety Decision Tree

Adding a new parameter to a resumable job function?
|-- Is it job-differentiating (two jobs can have different values)?
|   |-- YES -> Add to checkpoint filename
|   |          Add to pueue job label
|   |          Add to remote checkpoint key
|   |-- NO  -> Skip (e.g., verbose, notify are per-run, not per-job)
|
|-- Does the function delete files?
|   |-- YES -> Use missing_ok=True
|   |          Use atomic write for creates
|   |-- NO  -> Standard operation
|
|-- Does the function write to shared storage?
    |-- YES -> Force deduplication after write
    |          Use UPSERT semantics where possible
    |-- NO  -> Standard operation

---

Autoscaler

Dynamic parallelism tuning for pueue groups based on host CPU and memory. Full details: references/autoscaler.md

CPU < 40% AND MEM < 60%  ->  SCALE UP (+1 per group)
CPU > 80% OR  MEM > 80%  ->  SCALE DOWN (-1 per group)
Otherwise                 ->  HOLD

Key principle: Ramp up incrementally (not to max). Job memory grows over time -- jumping to max parallelism risks OOM when all jobs peak simultaneously.

---

Project-Specific Extensions

This skill provides universal patterns that apply to any distributed job pipeline. Projects should create a local extension skill (e.g., myproject-job-safety) in their .claude/skills/ directory that provides:

Two-layer invocation pattern: When this skill is triggered, also check for and invoke any local *-job-safety skill in the project's .claude/skills/ directory for project-specific configuration.

devops-tools:distributed-job-safety    (universal patterns - this skill)
  + .claude/skills/myproject-job-safety  (project-specific config)
  = Complete operational knowledge

---

SOTA Alternative: Temporal for Durable Workflows

For structured, repeatable job pipelines, Temporal provides built-in enforcement of many invariants in this skill:

When to consider Temporal: When your pipeline has well-defined activities (not ad-hoc shell commands), needs dedup/idempotency guarantees, or when the overhead of pueue guardrails (autoscaler agents, manual retry classification) exceeds the overhead of running a Temporal server.

Install: pip install temporalio (Python SDK), brew install temporal (CLI + dev server).

Lesson from 2026-03-04 incident: 5 autonomous Claude Code agents monitoring 60 pueue jobs created ~12,800 runaway tasks because pueue's restart creates new tasks (not in-place), agents had no mutation budgets, and persistent failures were blindly retried. Temporal prevents all three failure modes natively.

---

References

• Anti-Patterns -- 17 production failure patterns (AP-1 through AP-17)
• Concurrency Invariants -- Formal invariant specifications (INV-1 through INV-9)
• Deployment Checklist -- Step-by-step remote deployment protocol
• Environment Gotchas -- Host-specific pitfalls (G-1 through G-17)
• Stack Architecture -- Mise + Pueue + systemd-run layer diagram
• Autoscaler -- Dynamic parallelism tuning patterns
• Cross-reference: devops-tools:pueue-job-orchestration -- Pueue basics, dependency chaining, installation
• SOTA Alternative: Temporal -- Durable workflow orchestration with built-in dedup and retry

Post-Execution Reflection

After this skill completes, check before closing:

1. Did the command succeed? — If not, fix the instruction or error table that caused the failure. 2. Did parameters or output change? — If the underlying tool's interface drifted, update Usage examples and Parameters table to match. 3. Was a workaround needed? — If you had to improvise (different flags, extra steps), update this SKILL.md so the next invocation doesn't need the same workaround.

Only update if the issue is real and reproducible — not speculative.

Anti-Patterns (Learned from Production)

AP-1: Redeploying Without Checking Running Jobs

Symptom: Killed running jobs, requeued new ones. Old checkpoint files from killed jobs persisted, causing collisions with new jobs.

Fix: Always run state audit before redeployment:

pueue status --json | jq '[.tasks[] | select(.status | keys[0] == "Running")] | length'
# If > 0, decide: wait, kill gracefully, or abort

See: Deployment Checklist

AP-2: Checkpoint Filename Missing Job Parameters

Symptom: FileNotFoundError on checkpoint delete -- Job A deleted Job B's checkpoint.

Root cause: Filename {item}_{start}_{end}.json lacked a differentiating parameter. Two jobs for the same item at different configurations shared the file.

Fix: Include ALL differentiating parameters: {item}_{config}_{start}_{end}.json

AP-3: Trusting pueue restart Logs

Symptom: pueue log <id> shows old error after pueue restart, appearing as if the restart failed.

Root cause: Pueue appends output to existing log. After restart, the log contains BOTH the old failed run and the new attempt.

Fix: Check timestamps in the log, or add a new fresh job instead of restarting:

# More reliable than restart
pueue add --group mygroup --label "BTCUSDT@750-retry" -- <same command>

AP-4: Assuming PyPI Propagation Is Instant

Symptom: uv pip install pkg==X.Y.Z fails with "no version found" immediately after publishing.

Root cause: PyPI CDN propagation takes 30-120 seconds.

Fix: Use --refresh flag to bust cache:

uv pip install --refresh --index-url https://pypi.org/simple/ mypkg==<version>

AP-5: Confusing Editable Source vs. Installed Wheel

Symptom: Updated pip package to latest, but uv run still uses old code.

Root cause: uv.lock has source = { editable = "." } -- uv run reads Python files from the git working tree, not from the installed wheel.

Fix: On remote hosts, git pull updates the source that uv run reads. Pip install only matters for non-editable environments.

AP-6: Sequential Phase Assumption

Symptom: Phase 2 jobs started while Phase 1 was still running for the same item, creating contention.

Root cause: All phases queued simultaneously.

Fix: Either use pueue dependencies (--after <id>) or queue phases sequentially after verification:

# Queue Phase 1, wait for completion, then Phase 2
pueue add --label "phase1" -- run_phase_1
# ... wait and verify ...
pueue add --label "phase2" -- run_phase_2

AP-7: Manual Post-Processing Steps

Symptom: Queue batch jobs, print "run optimize after they finish."

# WRONG
postprocess_all() {
    queue_batch_jobs
    echo "Run 'pueue wait' then manually run optimize and validate"  # NO!
}

Fix: Wire post-processing as pueue --after dependent jobs:

# RIGHT
postprocess_all() {
    JOB_IDS=()
    for param in 250 500 750 1000; do
        job_id=$(pueue add --print-task-id --group mygroup \
            --label "ITEM@${param}" -- uv run python process.py --param "$param")
        JOB_IDS+=("$job_id")
    done
    # Chain optimize after ALL batch jobs
    optimize_id=$(pueue add --print-task-id --after "${JOB_IDS[@]}" \
        -- clickhouse-client --query "OPTIMIZE TABLE mydb.mytable FINAL")
    # Chain validation after optimize
    pueue add --after "$optimize_id" -- uv run python scripts/validate.py
}

Cross-reference: See devops-tools:pueue-job-orchestration Dependency Chaining section for full --after patterns.

AP-8: Hardcoded Job IDs in Pipeline Monitors

Symptom: Background monitor crashes with empty variable or wrong comparison after jobs are removed, re-queued, or split into per-year jobs.

Root cause: Monitor uses grep "^14|" to find specific job IDs. When those IDs no longer exist (killed, removed, replaced by per-year splits), the grep returns empty and downstream comparisons fail.

Fix: Detect phase transitions by group completion patterns, not by tracking individual job IDs. Use group_all_done() to check if all jobs in a pueue group have finished.

Principle: Pueue group names and job labels are stable identifiers. Job IDs are ephemeral.

Cross-reference: See devops-tools:pueue-job-orchestration Pipeline Monitoring section for the full group_all_done() implementation and integrity check patterns.

AP-9: Sequential Processing When Epoch Resets Enable Parallelism

Symptom: A multi-year job runs for days single-threaded while 25+ cores sit idle. ETA: 1,700 hours.

Root cause: Pipeline processor resets state at epoch boundaries (yearly, monthly) — each epoch is already independent. But the job was queued as one monolithic range.

Fix: Split into per-epoch pueue jobs running concurrently:

# WRONG: Single monolithic job, wastes idle cores
pueue add -- process --start 2019-01-01 --end 2026-12-31  # 1,700 hours single-threaded

# RIGHT: Per-year splits, 5x+ speedup on multi-core
for year in 2019 2020 2021 2022 2023 2024 2025 2026; do
    pueue add --group item-yearly --label "ITEM@250:${year}" \
        -- process --start "${year}-01-01" --end "${year}-12-31"
done

When this applies: Any pipeline where the processor explicitly resets state at time boundaries (ouroboros pattern, rolling windows, annual rebalancing). If the processor carries state across boundaries, per-epoch splitting is NOT safe.

Cross-reference: See devops-tools:pueue-job-orchestration Per-Year Parallelization section for full patterns.

AP-10: State File Bloat Causing Silent Performance Regression

Symptom: Job submission that used to take 10 minutes now takes 6+ hours. No errors — just slow. Pipeline appears healthy but execution slots sit idle waiting for new jobs to be queued.

Root cause: Pueue's state.json grows with every completed task. At 50K+ completed tasks (80-100MB state file), each pueue add takes 1-2 seconds instead of <100ms. This is invisible — no errors, no warnings, just gradually degrading throughput.

Why it's dangerous: The regression is proportional to total completed tasks across the daemon's lifetime. A sweep that runs 10K jobs/day hits the problem by day 5. The first day runs fine, creating a false sense of security.

Fix: Treat state.json as infrastructure that requires periodic maintenance:

# Before bulk submission: always clean
pueue clean -g mygroup 2>/dev/null || true

# During long sweeps: clean between batches
# (See pueue-job-orchestration skill for full batch pattern)

# Monitor state size as part of health checks
STATE_FILE="$HOME/.local/share/pueue/state.json"
ls -lh "$STATE_FILE"  # Should be <10MB for healthy operation

Invariant: state.json size should stay below 50MB during active sweeps. Above 50MB, pueue add latency exceeds 500ms and parallel submission gains vanish.

Cross-reference: See devops-tools:pueue-job-orchestration State File Management section for benchmarks and the periodic clean pattern.

AP-11: Wrong Working Directory in Remote Pueue Jobs

Symptom: Jobs fail immediately (exit code 2) with can't open file 'scripts/populate.py': [Errno 2] No such file or directory.

Root cause: ssh host "pueue add -- uv run python scripts/process.py" queues the job with the SSH session's cwd (typically $HOME), not the project directory. The script path is relative, so pueue looks for ~/scripts/process.py instead of ~/project/scripts/process.py.

Fix: Use -w (preferred) or cd && to set the working directory:

# WRONG: pueue inherits SSH cwd ($HOME)
ssh host "pueue add --group mygroup -- uv run python scripts/process.py"

# RIGHT (preferred): -w flag sets working directory explicitly
ssh host "pueue add -w ~/project --group mygroup -- uv run python scripts/process.py"

# RIGHT (alternative): cd first, then pueue add inherits project cwd
ssh host "cd ~/project && pueue add --group mygroup -- uv run python scripts/process.py"

Note: Pueue v4 does have -w / --working-directory. Use it as the primary approach. Fall back to cd && for SSH-piped commands where -w path expansion may differ. On macOS, -w /tmp resolves to /private/tmp (symlink).

Test: After queuing, verify the Path column in pueue status shows the project directory, not $HOME.

AP-12: Per-File SSH for Bulk Job Submission

Symptom: Submitting 300K jobs takes days because each pueue add requires a separate SSH round-trip from the local machine to the remote host.

Root cause: The submission script runs locally and calls ssh host "pueue add ..." per job. Each SSH connection has ~50-100ms overhead. At 300K jobs: 300K \* 75ms = 6.25 hours just for SSH, before any submission latency.

Fix: Generate a commands file locally, rsync it to the remote host, then run xargs -P on the remote host to eliminate SSH overhead entirely:

# Step 1 (local): Generate commands file
bash gen_commands.sh > /tmp/commands.txt

# Step 2 (local): Transfer to remote
rsync /tmp/commands.txt host:/tmp/commands.txt

# Step 3 (remote): Feed via xargs -P (no SSH per-job)
ssh host "xargs -P16 -I{} bash -c '{}' < /tmp/commands.txt"

Invariant: Bulk submission should run ON the same host as pueue. The only SSH call should be to start the feeder process, not per-job.

AP-13: SIGPIPE Under set -euo pipefail

Symptom: Script exits with code 141 (128 + SIGPIPE=13) on harmless pipe operations.

Root cause: ls *.sql | head -10 — head reads 10 lines then closes stdin. ls gets SIGPIPE writing to closed pipe. Under set -o pipefail, this propagates as exit 141.

Fix: Avoid piping to head in strict-mode scripts:

# WRONG (exit 141)
ls /tmp/sql/*.sql | head -10

# RIGHT (temp file)
ls /tmp/sql/*.sql > /tmp/filelist.txt
head -10 /tmp/filelist.txt

AP-14: False Data Loss From Variable-Width NDJSON Output

Symptom: wc -l shows fewer lines than expected. Appears as 3-6% "data loss".

Root cause: Configs with 0 signals after feature filtering produce 1 NDJSON line (skipped entry), not N barrier lines. Example: 95 normal × 3 + 5 skipped × 1 = 290 (not 300).

Fix: Account for variable output width in line count validation:

expected = N_normal * barriers_per_query + N_skipped * 1 + N_error * 1

AP-15: Cursor File Deletion on Completion

Symptom: ETL/indexer job succeeds, but next invocation does a full re-run instead of incremental resume.

Root cause: Code deletes the cursor/checkpoint/offset file after processing completes (e.g., CURSOR_FILE.unlink() in the "done" branch). The cursor IS the resume state — deleting it forces a full re-index.

Fix: Never delete checkpoint files on success. Add a filename-based fallback for recovery:

# WRONG
if not has_more_data:
    cursor_file.unlink()  # "Clean up" destroys resume state

# RIGHT
# Leave cursor in place. Next run reads it, queries for new data, finds none, exits quickly.
# Add fallback: derive position from output filenames if cursor is lost.

See also: G-17

AP-16: Using mise [env] for Secrets Consumed by Pueue/Cron/Systemd Jobs

Symptom: Jobs work in interactive shell but fail in pueue/cron/systemd with empty env vars.

Root cause: mise [env] variables require mise activation in the shell. Pueue jobs, cron jobs, and systemd services run in clean shells without mise. Workarounds (eval "$(mise env)" inside jobs) introduce trust issues, version incompatibilities, and __MISE_DIFF leakage over SSH. # PROCESS-STORM-OK (documentation of anti-pattern)

Fix: Use python-dotenv + .env for secrets. Use mise.toml [tasks] for task definitions only:

# mise.toml — tasks only, no [env] for secrets
[tasks.backfill]
run = "bash scripts/backfill.sh"

[tasks.ingest]
run = "bash scripts/ingest.sh"

# scripts/backfill.sh — just cd so python-dotenv finds .env
pueue add -- bash -c 'cd ~/project && uv run python my_indexer.py'

# my_indexer.py — loads .env from cwd automatically
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.getenv("API_KEY")  # Works in interactive shell AND pueue jobs

See also: G-15

AP-17: Unscoped Glob Consumes Artifacts From Other Pipeline Categories

Symptom: Phase A aggregation produces correct results. Phase B aggregation finds empty input (no files). Or Phase B produces mixed/contaminated results containing Phase A's data.

Root cause: Both phases write artifacts to the same directory with filenames that differ only by a dimension NOT included in the cleanup glob. Phase A's glob matches ALL artifacts, consuming or deleting Phase B's files.

# WRONG: Unscoped glob matches ALL categories
chunk_files = folds_dir.glob("_chunk_*.parquet")  # Eats long AND short
for p in chunk_files:
    p.unlink()  # Deletes short's chunks too

# RIGHT: Category-scoped glob
chunk_files = folds_dir.glob(f"_chunk_{direction}_*.parquet")
for p in chunk_files:
    p.unlink()  # Only deletes this direction's chunks

The pattern: This occurs whenever:

1. Two pipeline phases share an output directory 2. Artifacts are named by a subset of discriminating dimensions 3. A glob pattern doesn't include ALL discriminating dimensions 4. One phase runs before the other and "cleans up" shared files

Fix: Apply INV-9 — include ALL category dimensions in filenames AND scope all globs to the current category. Add post-merge validation to catch contamination early.

Discovery: Gen720 WFO pipeline (2026-02-17). LONG aggregation consumed SHORT Parquet chunks via _chunk_*.parquet glob, producing a mixed 8.7M-row Parquet. SHORT aggregation found 0 chunks.

Autoscaler

Pueue has no resource awareness. The autoscaler complements it with dynamic parallelism tuning.

How it works: Reads CPU load + available memory, then adjusts pueue parallel N per group.

CPU < 40% AND MEM < 60%  ->  SCALE UP (+1 per group)
CPU > 80% OR  MEM > 80%  ->  SCALE DOWN (-1 per group)
Otherwise                 ->  HOLD

Incremental Scaling Protocol

Don't jump to max capacity. Ramp up in steps and verify stability at each level:

Step 1: Start with conservative defaults (e.g., group1=2, group2=3)
Step 2: After jobs stabilize (~5 min), probe: uptime + free -h + ps aux
Step 3: If load < 40% cores AND memory < 60% available:
        Bump by +1-2 jobs per group
Step 4: Wait ~5 min for new jobs to reach peak memory
Step 5: Probe again. If still within 80% margin, bump again
Step 6: Repeat until load ~50% cores OR memory ~70% available

Why incremental: Job memory footprint grows over time (a job may start at ~500 MB and peak at 5+ GB). Jumping straight to max parallelism risks OOM when all jobs hit peak simultaneously.

Safety bounds: Each group should have min/max limits the autoscaler won't exceed. It should also check per-job memory estimates before scaling up (don't add a 5 GB job if only 3 GB available).

Dynamic Adjustment

Pueue supports live tuning without restarting jobs:

# Scale up when resources are available
pueue parallel 4 --group group1
pueue parallel 5 --group group2

# Scale down if memory pressure detected
pueue parallel 2 --group group1

Per-Family Groups

When jobs have vastly different resource profiles, give each family its own pueue group. This prevents a single high-memory job type from starving lighter jobs:

# Example: high-volume symbols need fewer concurrent jobs (5 GB each)
pueue group add highvol-yearly --parallel 2

# Low-volume symbols can run more concurrently (1 GB each)
pueue group add lowvol-yearly --parallel 6

Concurrency Invariants

Formal specifications for concurrent job safety. Each invariant includes a violation scenario and enforcement pattern.

---

INV-1: Checkpoint File Isolation

Statement: For any two concurrent jobs J_a and J_b where params(J_a) != params(J_b), the checkpoint paths must be distinct: checkpoint_path(J_a) != checkpoint_path(J_b).

Violation scenario:

J_a = (symbol=BTCUSDT, threshold=1000, start=2024-01-01, end=2024-12-31)
J_b = (symbol=BTCUSDT, threshold=750,  start=2024-01-01, end=2024-12-31)

# Without threshold in filename:
checkpoint_path(J_a) = checkpoints/BTCUSDT_2024-01-01_2024-12-31.json
checkpoint_path(J_b) = checkpoints/BTCUSDT_2024-01-01_2024-12-31.json  # COLLISION

# J_a finishes, deletes checkpoint
# J_b tries to read/delete -> FileNotFoundError

Enforcement:

def get_checkpoint_path(symbol, threshold, start_date, end_date, ...):
    filename = f"{symbol}_{threshold}_{start_date}_{end_date}.json"
    return checkpoint_dir / filename

Verification: For every pair of pueue job labels in the same group, assert their checkpoint paths differ.

---

INV-2: Idempotent Cleanup

Statement: For any file deletion operation delete(path), the operation must be idempotent: calling delete(path) when the file does not exist must not raise an exception.

Violation scenario:

# J_a finishes at T=100, calls: checkpoint_path.unlink()  -> OK (file deleted)
# J_b finishes at T=101, calls: checkpoint_path.unlink()  -> FileNotFoundError!

# Even worse with TOCTOU:
# J_a: if checkpoint_path.exists():  -> True (T=100)
# J_b:   checkpoint_path.unlink()    -> OK (T=100.5, deletes file)
# J_a:   checkpoint_path.unlink()    -> FileNotFoundError! (T=101, file already gone)

Enforcement:

# ALWAYS use missing_ok=True
path.unlink(missing_ok=True)

# NEVER use exists() + unlink() pair
# NEVER use try/except for this (verbose, error-prone)

---

INV-3: Atomic State Writes

Statement: A checkpoint file must always contain either the complete previous state or the complete new state, never a partial write.

Violation scenario:

# Job writing checkpoint (no atomic write):
open("checkpoint.json", "w")
write(first_half_of_json)     # <- process killed here by OOM
# File now contains truncated JSON
# On resume: json.loads() -> JSONDecodeError

Enforcement: Tempfile + fsync + atomic rename:

fd, temp_path = tempfile.mkstemp(dir=path.parent, prefix=".ckpt_", suffix=".tmp")
with os.fdopen(fd, "w") as f:
    f.write(json.dumps(data))
    f.flush()
    os.fsync(f.fileno())         # Force to disk
os.replace(temp_path, path)      # POSIX guarantees atomic rename

Recovery: The checkpoint loader should handle JSONDecodeError gracefully by returning None, triggering a fresh start from the last known-good state.

---

INV-4: Environment Override Scoping

Statement: An environment variable override for a pueue job MUST NOT affect other concurrent jobs or the host's default configuration.

Violation scenario:

# WRONG: Global modification
export MY_APP_MIN_THRESHOLD=250
pueue add -- uv run python script.py --threshold 250
pueue add -- uv run python script.py --threshold 1000  # Also uses 250 minimum!

# Even worse: editing .mise.toml

Enforcement: Use env prefix per-job when overrides are truly needed:

pueue add -- env MY_APP_MIN_THRESHOLD=250 uv run python script.py --threshold 250
pueue add -- env MY_APP_MIN_THRESHOLD=250 uv run python script.py --threshold 1000
# Each job gets its own environment scope

Preferred: Set the correct value in .mise.toml so per-job overrides are unnecessary.

---

INV-5: Post-Write Deduplication

Statement: After writing data to a storage backend that uses eventual-consistency deduplication (e.g., ClickHouse ReplacingMergeTree), an explicit deduplication step must follow.

Violation scenario:

# Job fails at day 2024-06-15, retries from checkpoint
# Day 2024-06-15 is reprocessed, rows are INSERT'd again
# ClickHouse ReplacingMergeTree: dedup happens in background merge (hours later)
# Query immediately after: sees duplicate rows for 2024-06-15

Enforcement:

# After processing loop completes:
with MyCache() as cache:
    cache.deduplicate(symbol, threshold)

Applies to: Any storage engine with lazy deduplication (ClickHouse ReplacingMergeTree, Parquet append, eventual-consistency stores).

---

INV-6: Gate Before Compute

Statement: Validation gates (input registry, parameter bounds) MUST execute before any expensive operation (data fetch, computation, cache write).

Violation scenario:

# WRONG: Validate after fetch
data = fetch_data(symbol, start, end)   # 10 minutes of download
validate_input(symbol)                   # Fails! Wasted 10 minutes.

# WORSE: No validation at all
data = fetch_data("MATIC_USDT", ...)    # Typo, but fetches garbage data
result = process(data)                   # Computes from garbage
cache.store(result)                      # Stores garbage permanently

Enforcement:

def run_resumable_job(symbol, ...):
    validate_input(symbol)                      # FIRST
    start_date = validate_and_clamp(symbol, start_date)  # SECOND
    checkpoint_path = get_checkpoint_path(...)   # THEN proceed

---

INV-7: Per-Job Memory Isolation

Statement: Each concurrent job must have an enforced upper bound on physical memory consumption, preventing any single job from starving the host via swap thrashing.

Violation scenario:

# Job memory profile: starts ~500 MB, peaks at 5+ GB over hours
# 6 concurrent heavy jobs peak simultaneously:
# 6 * 5 GB = 30 GB demand on 61 GB host -> triggers swap
# All jobs now swap-thrashing -> load 50+ -> SSH unresponsive -> host frozen

Enforcement: systemd-run --user --scope with cgroups v2:

systemd-run --user --scope \
    -p MemoryMax=8G \
    -p MemorySwapMax=0 \
    uv run python scripts/process.py --symbol BTCUSDT --threshold 250

Critical: MemorySwapMax=0 is mandatory. Without it, Linux memory overcommit allows processes to spill into swap, defeating the purpose of MemoryMax.

Verification (while job is running):

SCOPE=$(pueue log <id> | grep "Running as unit" | grep -o "run-r[a-z0-9]*.scope")
CGROUP=$(find /sys/fs/cgroup/user.slice -name "$SCOPE" -type d | head -1)
cat $CGROUP/memory.current    # Should be < MemoryMax
cat $CGROUP/memory.max        # Should match MemoryMax
cat $CGROUP/memory.swap.max   # Should be 0

Platform: Linux with cgroups v2 only. Falls back to plain execution on macOS. Bypass with MY_APP_NO_CGROUP=1.

---

INV-8: Monitor by Stable Identifiers, Not Ephemeral IDs

Statement: Pipeline monitors and orchestration scripts must identify jobs by stable attributes (group names, label patterns) — never by ephemeral numeric IDs that change when jobs are removed, re-queued, or restructured.

Violation scenario:

# Monitor hardcodes job IDs from initial queue submission
optimize_job=14
detect_job=15
backfill_jobs=(16 17 18)

# Later: jobs 16-18 are killed and replaced with per-year splits (IDs 21-47)
# Monitor still checks job 16 -> empty result -> crash or false positive

Enforcement: Use group names and label patterns:

# Query by group (stable)
group_jobs=$(pueue status --json | jq -r \
    '.tasks | to_entries[] | select(.value.group == "btc-yearly") | .value.id')

# Query by label pattern (stable)
optimize_job=$(pueue status --json | jq -r \
    '.tasks | to_entries[] | select(.value.label == "optimize-table:final") | .value.id')

Principle: Pueue group names and job labels are chosen by the user and remain stable. Job IDs are auto-incremented integers that shift with every queue mutation.

---

INV-9: Derived Artifact Category Isolation

Statement: For any two pipeline phases P_a and P_b that write derived artifacts to a shared directory, every artifact filename must include ALL dimensions that differentiate P_a from P_b. Glob patterns used for reading, merging, or deleting artifacts must be scoped to the executing phase's dimensions.

Violation scenario:

P_a = (direction=long, formation=exh_l, symbol=SOLUSDT, threshold=500)
P_b = (direction=short, formation=exh_s, symbol=SOLUSDT, threshold=500)

# Without direction in filename:
artifact(P_a) = folds/_chunk_exh_l_SOLUSDT_500.parquet
artifact(P_b) = folds/_chunk_exh_s_SOLUSDT_500.parquet

# P_a merges with: glob("_chunk_*.parquet")
# COLLISION: glob matches BOTH P_a and P_b artifacts
# P_a merges all into long_folds.parquet (now contaminated with SHORT data)
# P_a deletes all chunks (P_b's chunks are gone)
# P_b runs: glob("_chunk_*.parquet") -> 0 files -> empty output

Enforcement:

# Include ALL category dimensions in filename
chunk_path = folds_dir / f"_chunk_{direction}_{formation}_{symbol}_{threshold}.parquet"

# Scope glob to current phase's category
chunk_files = folds_dir.glob(f"_chunk_{direction}_*.parquet")

# Post-merge validation
merged_df = pl.concat([pl.read_parquet(p) for p in chunk_files])
expected_strategies = {"standard"} if direction == "long" else {"A_mirrored", "B_reverse"}
actual = set(merged_df["strategy"].unique().to_list())
assert actual == expected_strategies, f"Category contamination: expected {expected_strategies}, got {actual}"

Verification: After merging derived artifacts, assert that category columns contain only the expected values for the current phase. This catches contamination even if filenames are accidentally unscoped.

Relationship to INV-1: INV-1 ensures runtime checkpoint uniqueness. INV-9 extends the same principle to derived artifacts that persist across pipeline phases and may be consumed by later phases running in different category contexts.

---

Testing Invariants

Verify these invariants hold after any code change:

# INV-1: Filename uniqueness
path_a = get_checkpoint_path("BTCUSDT", 1000, "2024-01-01", "2024-12-31")
path_b = get_checkpoint_path("BTCUSDT", 750,  "2024-01-01", "2024-12-31")
assert path_a != path_b, "INV-1 violated: same path for different thresholds"

# INV-2: Idempotent delete
from pathlib import Path
p = Path("/tmp/test_inv2.json")
p.unlink(missing_ok=True)  # Should not raise even if file doesn't exist
p.unlink(missing_ok=True)  # Second call also safe

# INV-3: Atomic write recovery
# Kill process during checkpoint.save(), verify file is either old or new, never partial

# INV-4: Env scoping
# Run two pueue jobs with different env overrides, verify they don't interfere

# INV-5: Post-dedup
# Write duplicate rows, verify deduplicate() removes them

# INV-6: Gate ordering
# Call run_resumable_job("INVALID_INPUT", ...) -> must fail before any I/O

# INV-7: Memory isolation
# Run job under systemd-run with MemoryMax, verify cgroup limits are enforced

# INV-8: Monitor by stable identifiers
# After re-queuing a job, verify monitoring scripts still find it by group/label
# (not by old job ID which no longer exists)

# INV-9: Derived artifact category isolation
# Write artifacts with two different category values to same directory
# Verify that merging for category A does not include category B's artifacts
# Verify that cleanup for category A does not delete category B's artifacts

Deployment Checklist

Step-by-step protocol for deploying code changes to a remote host running pueue jobs.

Critical principle: AUDIT before MUTATE. Never modify running state without understanding it first.

---

Pre-Deployment Audit

Run BEFORE any code changes on the remote host.

# 1. Count running/queued/failed jobs
ssh host 'pueue status --json' | python3 -c "
import json, sys
data = json.load(sys.stdin)
tasks = data.get('tasks', {})
states = {'Running': 0, 'Queued': 0, 'Success': 0, 'Failed': 0}
for t in tasks.values():
    status = t.get('status', '')
    if isinstance(status, dict):
        key = list(status.keys())[0]
        if key == 'Done':
            result = status['Done'].get('result', '')
            if result == 'Success' or (isinstance(result, dict) and 'Success' in result):
                states['Success'] += 1
            else:
                states['Failed'] += 1
        else:
            states[key] = states.get(key, 0) + 1
print(states)
"

# 2. List active job labels (what's actually running)
ssh host 'pueue status --json' | jq -r '
  [.tasks[] | select(.status | keys[0] == "Running") | .label] | .[]
'

# 3. Check for stale checkpoints
ssh host 'ls -la ~/.cache/myapp/checkpoints/'

---

Forensic Database Audit

Before deployment, audit the database for corruption that the fix addresses:

# ClickHouse: Check for corruption indicators
ssh host "clickhouse-client --query \"
  SELECT symbol, threshold,
    count() as rows,
    countIf(value < 0) as neg_values,
    round(countIf(value < 0) * 100.0 / count(), 2) as pct_corrupt
  FROM mydb.mytable
  GROUP BY symbol, threshold
  HAVING neg_values > 0
  ORDER BY symbol, threshold
  FORMAT PrettyCompact\""

This baseline is critical for:

1. Confirming the scope of corruption (which items/parameters affected) 2. Deciding which jobs need --force-refresh vs checkpoint resume 3. Post-deployment verification (expect zero corrupt rows after reprocessing)

---

Decision Matrix

Never: Kill running jobs and immediately requeue without cleaning up checkpoints.

---

Force-Refresh vs Checkpoint Resume

When restarting jobs after a code upgrade, choose based on data integrity:

Critical: Jobs with clean checkpoints should NOT use --force-refresh -- it deletes the checkpoint and all cached data, losing hours/days of progress.

# Clean data, just upgrade -- resume from checkpoint
pueue add --label "DOGEUSDT@250" -- uv run python process.py --symbol DOGEUSDT --threshold 250

# Corrupt data -- wipe and restart
pueue add --label "ITEM@250" -- uv run python process.py --symbol ITEM --threshold 250 --force-refresh

---

Deployment Steps

Step 1: Pull Code

ssh host 'cd ~/project && git fetch origin main && git reset --hard origin/main'

Verify: Check the commit hash matches expected release.

ssh host 'cd ~/project && git log --oneline -1'

Step 2: Upgrade Package (if non-editable env exists)

# For project .venv (editable source, updated by git pull)
# No action needed - uv run reads from working tree

# For standalone .venv (non-editable, needs pip upgrade)
ssh host 'cd ~/project && uv pip install --python .venv/bin/python --refresh mypkg==<version>'

Step 3: Verify Fix Is Active

Use inspect.getsource() to confirm the deployed code contains the expected fix:

ssh host 'cd ~/project && .venv/bin/python -c "
import inspect, mypkg.checkpoint as cp
src = inspect.getsource(cp.get_checkpoint_path)
assert \"threshold\" in src, \"FIX NOT APPLIED: threshold not in checkpoint path\"
print(\"OK: fix verified\")
"'

Adapt the assertion to match whatever the fix changes. The key pattern is: inspect the source code of the fixed function and assert the fix signature is present.

Step 4: Handle Failed Jobs

# Option A: Add fresh replacement job (preferred)
ssh host 'pueue add --group mygroup --label "SYMBOL@THRESHOLD-retry" -- <command>'

# Option B: Restart in-place (may show stale logs -- see AP-3)
ssh host 'pueue restart <job_id>'

Step 5: Monitor

# Watch specific job
ssh host 'pueue follow <job_id>'

# Periodic status check
ssh host 'pueue status --group mygroup'

---

Post-Deployment Verification

After all jobs complete:

# 1. Check for failures
ssh host 'pueue status --json' | jq '[.tasks[] | select(.status.Done.result != "Success")] | length'

# 2. Run domain-specific validation script
ssh host 'cd ~/project && uv run python scripts/validate.py'

# 3. Clean up completed jobs
ssh host 'pueue clean'

---

Emergency: Killing All Jobs

If absolutely necessary (corrupted state, runaway processes):

# 1. Kill all running jobs
ssh host 'pueue kill --all'

# 2. Remove all jobs from queue
ssh host 'pueue clean'
ssh host 'pueue status --json' | jq -r '.tasks | keys[]' | while read id; do
    ssh host "pueue remove $id"
done

# 3. Clean stale checkpoints
ssh host 'rm -f ~/.cache/myapp/checkpoints/*.json'

# 4. Unpause groups (pueue kill pauses groups as safety measure)
ssh host 'pueue start --all'

After emergency cleanup: Deploy fresh code, then requeue from scratch.

Environment Gotchas

Host-specific pitfalls encountered during remote deployments. Each gotcha includes the symptom, root cause, and fix.

---

G-1: PEP 668 Externally-Managed-Environment

Symptom: pip install mypkg fails with "externally managed environment" on Python 3.12+.

Root cause: PEP 668 (Python 3.12+) blocks pip in system-managed environments to prevent breakage.

Fix: Use uv instead of pip:

# WRONG
pip install mypkg

# RIGHT
uv pip install --python .venv/bin/python mypkg

Affected hosts: Any host with Python 3.12+ managed by system package manager.

---

G-2: ClickHouse Docker Auth Failure

Symptom: Authentication failed for user 'default' with local Docker ClickHouse.

Root cause: ClickHouse Docker image requires explicit password configuration, even if empty.

Fix:

docker run -e CLICKHOUSE_PASSWORD= clickhouse/clickhouse-server

---

G-3: uv run Editable vs. Wheel Confusion

Symptom: Installed new wheel version via pip, but uv run still executes old code.

Root cause: uv.lock defines source = { editable = "." }. When running uv run python script.py inside the project directory, uv reads Python files directly from the git working tree, NOT from the installed wheel.

Implications:

• git pull on remote host updates the code that uv run uses
• pip install only matters for standalone venvs outside the project
• Version shown by import pkg; pkg.__version__ may differ from actual source code behavior

Decision tree:

Is the remote host's project directory a git checkout?
|-- YES -> git pull updates the code uv run uses
|          pip install is irrelevant for uv run
|-- NO  -> pip install / uv pip install is required

---

G-4: Pueue Daemon Not Running

Symptom: pueue add fails with "Connection refused" or hangs.

Root cause: Pueue daemon (pueued) not started on the host.

Fix:

# Start daemon (persists across SSH disconnects)
pueued -d

# Verify
pueue status

Auto-start on Linux (systemd):

systemctl --user enable --now pueued

---

G-5: Pueue Groups Paused After Kill

Symptom: After pueue kill --all, new jobs stay in Queued state forever.

Root cause: pueue kill pauses all affected groups as a safety measure.

Fix: Unpause groups after kill:

pueue start --all

---

G-6: Pandas 3.0 Datetime Resolution

Symptom: Timestamps are off by factor of 1000 (seconds instead of milliseconds).

Root cause: Pandas 3.0 defaults to datetime64[us] (microsecond). .astype("int64") returns microseconds, not nanoseconds. So // 10**6 produces seconds instead of milliseconds.

Fix: Use explicit unit conversion:

# WRONG (breaks on pandas 3.0):
df["timestamp_ms"] = df["timestamp"].astype("int64") // 10**6

# RIGHT (works on pandas 2.x and 3.0):
df["timestamp_ms"] = df["timestamp"].dt.as_unit("ms").astype("int64")

Defensive guard: Add a scale check before database writes:

def _guard_timestamp_ms_scale(df):
    """Reject writes where timestamps are seconds instead of milliseconds."""
    if df["timestamp_ms"].min() < 1_000_000_000_000:
        raise ValueError("timestamp_ms appears to be seconds, not milliseconds")

---

G-7: SSH Key Not Available on Remote Host

Symptom: git clone git@github.com:... fails with "Permission denied (publickey)".

Root cause: Remote host doesn't have SSH key configured for GitHub.

Fix: Use HTTPS with token instead:

git remote set-url origin https://github.com/owner/repo.git
# Token is provided via GH_TOKEN env var from mise

Or use git init + fetch pattern:

cd ~/project
git init
git remote add origin https://github.com/owner/repo.git
git fetch origin main
git reset --hard origin/main

---

G-8: mise trust Required on Fresh Checkout

Symptom: mise run <task> fails with "Config files are not trusted".

Root cause: mise requires explicit trust for config files to prevent supply-chain attacks from untrusted repos.

Fix: Run once per checkout:

cd ~/project && mise trust

---

G-9: PyPI Propagation Delay

Symptom: uv pip install pkg==X.Y.Z fails immediately after publishing.

Root cause: PyPI CDN propagation takes 30-120 seconds globally.

Fixes (in order of preference):

# 1. Bust uv's index cache
uv pip install --refresh pkg==X.Y.Z

# 2. Force primary index
uv pip install --refresh --index-url https://pypi.org/simple/ pkg==X.Y.Z

# 3. Wait and retry
sleep 60 && uv pip install pkg==X.Y.Z

---

G-10: MemoryMax Without MemorySwapMax=0

Symptom: systemd-run --scope -p MemoryMax=256M doesn't prevent allocation beyond 256 MB.

Root cause: Linux memory overcommit + swap. Without MemorySwapMax=0, the cgroup can spill into swap, effectively making MemoryMax a soft limit.

Fix: Always pair MemoryMax with MemorySwapMax=0:

# WRONG: Process escapes into swap
systemd-run --user --scope -p MemoryMax=2G <command>

# RIGHT: Hard memory limit, no swap escape
systemd-run --user --scope -p MemoryMax=2G -p MemorySwapMax=0 <command>

Verification:

SCOPE=$(pueue log <id> | grep "Running as unit" | grep -o "run-r[a-z0-9]*.scope")
CGROUP=$(find /sys/fs/cgroup/user.slice -name "$SCOPE" -type d | head -1)
cat $CGROUP/memory.swap.max   # Must be 0

When OOM-killed: Exit code 137 (SIGKILL). Pueue marks the job as Failed.

---

G-11: Rust Not in PATH via uv run on Remote Hosts

Symptom: maturin develop --uv fails with "rustc not installed or not in PATH".

Root cause: uv run inherits a minimal PATH that doesn't include ~/.cargo/bin.

Fix: Prepend cargo bin to PATH:

# Interactive
PATH="$HOME/.cargo/bin:$PATH" uv run maturin develop --uv

# In pueue jobs
pueue add -- env PATH="/home/user/.cargo/bin:$PATH" uv run maturin develop --uv

---

G-12: Pueue add Inherits SSH cwd, Not Project Directory

Symptom: Pueue job fails instantly with No such file or directory for a relative script path.

Root cause: When running ssh host "pueue add -- cmd", pueue records the working directory as the SSH session's cwd (typically $HOME). Relative paths in the command resolve against $HOME, not the project root.

Fix: Always cd to the project directory in the same shell command:

# WRONG
ssh host "pueue add -- uv run python scripts/process.py"
# Job runs in $HOME, fails to find scripts/process.py

# RIGHT
ssh host "cd ~/project && pueue add -- uv run python scripts/process.py"
# Job runs in ~/project, relative paths work correctly

Verify: Check the Path column in pueue status output — it should show the project directory.

---

G-13: SIGPIPE (Exit 141) Under set -euo pipefail

Symptom: Bash script exits with code 141 on ls | head, cat | head, or similar pipe-to-head patterns.

Root cause: Under set -o pipefail, when head closes its stdin after reading N lines, the upstream command receives SIGPIPE (signal 13). Exit code = 128 + 13 = 141.

Fix: Avoid piping to head/tail -n in strict-mode scripts. Write to temp file first:

# WRONG (exit 141 under set -euo pipefail)
ls /tmp/gen600_sql/2down/*.sql | head -10

# RIGHT
find /tmp/gen600_sql/2down/ -name '*.sql' -print0 | head -z -n 10

# RIGHT (temp file approach)
ls /tmp/gen600_sql/2down/*.sql > /tmp/filelist.txt
head -10 /tmp/filelist.txt

---

G-14: Pipe Subshell Data Loss in While Loops

Symptom: while read ... done > $TMPOUT produces empty or truncated output when the input comes from a pipe.

Root cause: echo "$OUTPUT" | while read ...; do ...; done > file runs the while loop in a subshell (because it's the right side of a pipe). Variables set inside the loop are lost when the subshell exits. More critically, output redirection may not flush correctly under concurrent execution.

Fix: Use process substitution to keep the while loop in the main shell:

# WRONG (subshell, data loss risk)
echo "$OUTPUT" | while IFS=$'\t' read -r col1 col2 col3; do
    echo "processed: $col1"
done > "$TMPOUT"

# RIGHT (process substitution, main shell)
while IFS=$'\t' read -r col1 col2 col3; do
    echo "processed: $col1"
done < <(echo "$OUTPUT" | tail -n +2) > "$TMPOUT"

Affected: Any bash script that parses multi-line command output (ClickHouse TSV, CSV, etc.) into a while-read loop.

---

G-15: Pueue Jobs Cannot See mise [env] Variables

Symptom: Pueue job fails with empty env vars (e.g., MissingSchema: Invalid URL '') even though mise env shows them in your interactive shell.

Root cause: Pueue jobs run in a clean shell — no .bashrc, no .zshrc, no mise activation. Variables defined in mise.toml [env] are only available in shells that have run mise activation or eval "$(mise env)". # PROCESS-STORM-OK (documentation only)

Fix: For Python applications, use python-dotenv + .env file in the project root. The pueue job only needs cd $PROJECT_DIR — dotenv auto-loads .env from cwd at runtime:

# WRONG (pueue can't see mise [env] vars)
pueue add -- uv run python my_script.py  # POLYGON_RPC="" → crash

# WRONG (mise env works but requires trust + version compat)
pueue add -- bash -c 'export MISE_YES=1 && cd ~/project && eval "$(mise env)" && uv run python my_script.py'  # PROCESS-STORM-OK

# RIGHT (python-dotenv loads .env from cwd — zero binary dependencies)
pueue add -- bash -c 'cd ~/project && uv run python my_script.py'
# Requires: load_dotenv() in your Python entry point + .env file in project root

Architecture: Use mise.toml for task definitions only, .env for secrets (gitignored, loaded by python-dotenv at runtime). This is the most portable pattern across macOS/Linux, interactive/daemon shells, and local/remote execution.

Affected: Any pueue/cron/systemd job running Python code that expects mise-managed env vars.

---

G-16: mise Trust Errors Over SSH (\_\_MISE_DIFF Leakage)

Symptom: Every SSH command to a remote host fails with mise ERROR: Config files not trusted, even with mise trust run on the remote.

Root cause: The local machine's __MISE_DIFF environment variable (set by mise shell activation) propagates through SSH sessions. The remote mise sees this variable, interprets it as a stale diff, and triggers trust validation against remote configs — which fails because the serialized diff references local paths.

Fix: Either unset the variable locally before SSH, or set MISE_YES=1 on the remote:

# Option 1: Unset locally before SSH (per-command)
unset __MISE_DIFF && ssh host 'cd ~/project && mise env'

# Option 2: Set MISE_YES=1 in remote's .bashrc/.profile
# This auto-trusts all configs (OK for single-user servers)
echo 'export MISE_YES=1' >> ~/.bashrc

# Option 3: Trust the specific config on both sides
mise trust .               # On local machine
ssh host 'mise trust .'    # On remote machine

Affected: Any workflow where a macOS dev machine SSHes into Linux servers that also have mise installed. Most common with rsync + ssh or pueue remote orchestration.

---

G-17: Cursor/Checkpoint File Deletion Destroys Incremental Resume

Symptom: An indexer or ETL job runs successfully, but the next invocation does a full re-run instead of resuming from where it left off.

Root cause: The job deletes its cursor/checkpoint/offset file on completion (e.g., CURSOR_FILE.unlink() after the final batch). This was likely added to "clean up" but destroys the state needed for incremental runs.

Fix: Never delete checkpoint files on success. The checkpoint IS the proof of completion — it tells the next run "start from here":

# WRONG (deletes proof of progress)
def run(self):
    while has_more_data():
        batch = fetch_next_batch()
        save_batch(batch)
        CURSOR_FILE.write_text(str(batch.last_id))
    CURSOR_FILE.unlink()  # ← BUG: next run starts from scratch

# RIGHT (preserve checkpoint for incremental resume)
def run(self):
    while has_more_data():
        batch = fetch_next_batch()
        save_batch(batch)
        CURSOR_FILE.write_text(str(batch.last_id))
    # Done — cursor stays. Next run reads it and finds nothing new.

Bonus: Add a filename-based fallback for recovery if the cursor file is lost:

# Derive progress from output files (e.g., trades_100_200.parquet → resume from 200)
import re
pattern = re.compile(r"output_(\d+)_(\d+)\.parquet")
max_id = max(int(m.group(2)) for f in DATA_DIR.glob("output_*.parquet") if (m := pattern.match(f.name)))

Affected: Any ETL pipeline, web scraper, or blockchain indexer with resumable backfilling.

---

Quick Reference Table

Evolution Log

Convention: Reverse chronological order (newest on top, oldest at bottom). Prepend new entries.

---

2026-02-26: Initial Evolution Log

Status: Skill is in use and maintained. Track improvements here.

Purpose

This evolution log tracks updates to the skill. Each entry should note:

• What changed (content, structure, tooling)
• Why it changed (bug fix, feature request, best practice)
• Files affected

How to Use

1. When updating SKILL.md or references, add an entry here with the date 2. Keep entries reverse-chronological (newest first) 3. Link to ADRs or GitHub issues when relevant 4. Reference specific line changes when helpful

---

The Mise + Pueue + systemd-run Stack

mise (environment + task discovery)
  |-- .mise.toml [env] -> SSoT for defaults
  |-- .mise/tasks/jobs.toml -> task definitions
  |     |-- mise run jobs:submit-all
  |     |     |-- submit-all.sh (orchestrator)
  |     |           |-- pueue add (per-unit, NOT per-query)
  |     |                 |-- submit_unit.sh (per unit)
  |     |                       |-- xargs -P16 (parallel queries)
  |     |                             |-- wrapper.sh (per query)
  |     |                                   |-- clickhouse-client < sql_file
  |     |                                   |-- flock + append NDJSON
  |     |
  |     |-- mise run jobs:process-all (Python pipeline variant)
  |     |     |-- job-runner.sh (orchestrator)
  |     |           |-- pueue add (per-job)
  |     |                 |-- systemd-run --scope -p MemoryMax=XG -p MemorySwapMax=0
  |     |                       |-- uv run python scripts/process.py
  |     |                             |-- run_resumable_job()
  |     |                                   |-- get_checkpoint_path() -> param-aware
  |     |                                   |-- checkpoint.save() -> atomic write
  |     |                                   |-- checkpoint.unlink() -> missing_ok=True
  |     |
  |     |-- mise run jobs:autoscale-loop
  |           |-- autoscaler.sh --loop (60s interval)
  |                 |-- reads: free -m, uptime, pueue status --json
  |                 |-- adjusts: pueue parallel N --group <group>

Responsibility Boundaries