--- title: "Pipeline & Wrapper Patterns" type: runbook tags: [tooling] aliases: [] related: - [[meshtastic-sidecar-node]] - [[meshtastic-headscale-runbook]] - [[headscale-onboard-node]] - [[idahomesh-vpn-device-setup]] - [[syncthing-add-node]] updated: 2026-06-18 --- # Pipeline & Wrapper Patterns Three composable patterns for adding pre-flight logic to tools and pipelines you don't fully control: 1. **Binary Wrapper Interception** — the *delivery mechanism*. Transparently replace a CLI binary with a wrapper that runs custom logic, then `exec`s the real binary. The caller (service, cron, runner) never knows. 2. **GPU/CPU Fallback Routing** — *logic you inject*. Probe a job, route small→GPU / large→CPU, and gate concurrency with `flock` so excess jobs fail-fast and re-queue instead of OOM-killing each other. 3. **Pre-Flight Probe Gate** — *logic you inject*. Cheaply inspect each input and skip the expensive step when the work would be wasted (wrong format, already optimized, corrupt, too large). They compose: **Pattern 1 is how you deploy; Patterns 2 and 3 are two kinds of pre-flight logic you put inside the wrapper.** The running example throughout is the Whisper auto-captioning / PeerTube transcoder / [[recon]] extraction stack on cortex. ## Contents - [Pattern 1 — Binary Wrapper Interception](#pattern-1--binary-wrapper-interception) - [Pattern 2 — GPU/CPU Fallback Routing](#pattern-2--gpucpu-fallback-routing) - [Pattern 3 — Pre-Flight Probe Gate](#pattern-3--pre-flight-probe-gate) --- # Pattern 1 — Binary Wrapper Interception Transparently intercept a CLI binary with a wrapper script that adds pre-flight logic (routing, validation, logging) before exec-ing the real binary. The caller — whether a service, cron job, or another script — never knows the difference. Use this when you need to modify the behavior of a tool that's called by a system you don't control (e.g., a runner, scheduler, or third-party service), without changing the caller's config or code. ### Prerequisites - The binary to intercept is installed and working - You have root/sudo access on the target machine - The caller invokes the binary by absolute path or via PATH lookup ### Inputs Prompt the user for all of these before executing: ``` TARGET_HOST= # SSH alias or IP (e.g., cortex). Use "localhost" if local. BINARY_NAME= # Name of the binary to intercept (e.g., "whisper-ctranslate2") BINARY_PATH= # Full path to the binary (e.g., "/usr/local/bin/whisper-ctranslate2") WRAPPER_NAME= # Name for the wrapper script (e.g., "whisper-smart") WRAPPER_DIR= # Directory for the wrapper (e.g., "/usr/local/bin") REAL_SUFFIX= # Suffix for the renamed real binary (default: "-real") ``` ### Step 1: Locate the Real Binary Find the actual binary or symlink that will be intercepted. ```bash ssh $TARGET_HOST "ls -la $BINARY_PATH && file $BINARY_PATH" ``` If it's already a symlink, follow it to the real target: ```bash ssh $TARGET_HOST "readlink -f $BINARY_PATH" ``` **Gate:** Must return a valid file. Record the real binary location — you'll need it for the rename. ### Step 2: Rename the Real Binary Move the original binary out of the way so the wrapper can take its place. ```bash ssh $TARGET_HOST "sudo mv $BINARY_PATH ${BINARY_PATH}${REAL_SUFFIX}" ``` If the original was a symlink (e.g., pip-installed Python tool): ```bash # Preserve the symlink target REAL_TARGET=$(ssh $TARGET_HOST "readlink -f $BINARY_PATH") ssh $TARGET_HOST "sudo rm $BINARY_PATH && sudo ln -s $REAL_TARGET ${BINARY_PATH}${REAL_SUFFIX}" ``` **Gate:** ```bash ssh $TARGET_HOST "ls -la ${BINARY_PATH}${REAL_SUFFIX}" ssh $TARGET_HOST "${BINARY_PATH}${REAL_SUFFIX} --version 2>/dev/null || ${BINARY_PATH}${REAL_SUFFIX} --help 2>/dev/null | head -1" ``` The renamed binary must exist and be executable. If not, undo immediately: ```bash ssh $TARGET_HOST "sudo mv ${BINARY_PATH}${REAL_SUFFIX} $BINARY_PATH" ``` ### Step 3: Write the Wrapper Script Create the wrapper at `$WRAPPER_DIR/$WRAPPER_NAME`. The wrapper must: 1. Accept all original arguments (`$@`) 2. Perform pre-flight logic (inspection, routing, logging) 3. `exec` the real binary with (possibly modified) arguments 4. Never silently swallow errors — if pre-flight fails, exit with a meaningful code Template: ```bash #!/bin/bash # Wrapper for $BINARY_NAME — transparently intercepts calls # Real binary at: ${BINARY_PATH}${REAL_SUFFIX} LOGFILE="/tmp/${WRAPPER_NAME}.log" # ──── Pre-flight logic ──── # Add your inspection, routing, or validation here. # Example: inspect input files, check resource availability, choose parameters. # Parse arguments to find relevant inputs (file paths, flags, etc.) # This section is use-case specific. # ──── Logging ──── echo "[WRAPPER] $(date) args: $@" >> "$LOGFILE" # ──── Execute real binary ──── # Use exec to replace this process — caller sees the real binary's exit code, # stdout, stderr, and signal handling as if wrapper didn't exist. exec ${BINARY_PATH}${REAL_SUFFIX} "$@" ``` Deploy the wrapper: ```bash ssh $TARGET_HOST "sudo tee $WRAPPER_DIR/$WRAPPER_NAME > /dev/null << 'WRAPPER' WRAPPER sudo chmod +x $WRAPPER_DIR/$WRAPPER_NAME" ``` **Gate:** ```bash ssh $TARGET_HOST "ls -la $WRAPPER_DIR/$WRAPPER_NAME && head -1 $WRAPPER_DIR/$WRAPPER_NAME" ``` Must show executable permissions and `#!/bin/bash` shebang. ### Step 4: Install the Symlink Replace the original binary path with a symlink to the wrapper. ```bash ssh $TARGET_HOST "sudo ln -sf $WRAPPER_DIR/$WRAPPER_NAME $BINARY_PATH" ``` **Gate:** ```bash ssh $TARGET_HOST "ls -la $BINARY_PATH && ls -la ${BINARY_PATH}${REAL_SUFFIX}" ``` Should show: ``` BINARY_PATH -> WRAPPER_DIR/WRAPPER_NAME (wrapper) BINARY_PATH-real -> /path/to/actual/binary (real binary) ``` ### Step 5: Test the Interception Run the binary as the caller would. The wrapper should intercept transparently. ```bash # Direct invocation ssh $TARGET_HOST "$BINARY_PATH --version" # Check wrapper log ssh $TARGET_HOST "tail -5 /tmp/${WRAPPER_NAME}.log" ``` The `--version` output should come from the real binary. The log should show the wrapper fired. Test with actual workload arguments: ```bash ssh $TARGET_HOST "$BINARY_PATH " ssh $TARGET_HOST "tail -1 /tmp/${WRAPPER_NAME}.log" ``` **Gate:** Both must succeed. If the binary fails or produces different output than before, the wrapper has a bug — check argument passing (quoting, `$@` vs `$*`). ### Step 6: Verify Service Integration If the binary is called by a service (systemd, cron, etc.), restart that service and confirm it picks up the wrapper. ```bash ssh $TARGET_HOST "sudo systemctl restart " ssh $TARGET_HOST "sleep 5 && tail -5 /tmp/${WRAPPER_NAME}.log" ``` The log should show entries from the service's invocations, not just your manual tests. ### Rollback ```bash ssh $TARGET_HOST "sudo rm $BINARY_PATH && sudo mv ${BINARY_PATH}${REAL_SUFFIX} $BINARY_PATH" # Or if the original was a symlink: ssh $TARGET_HOST "sudo rm $BINARY_PATH && sudo ln -s $BINARY_PATH" ``` No service restart needed — next invocation hits the real binary directly. ### Key Principles 1. **`exec` is mandatory.** Without `exec`, the wrapper runs the binary as a child process, which breaks signal handling (SIGTERM won't reach the real binary) and doubles PID usage. `exec` replaces the wrapper process entirely. 2. **Use `"$@"` not `$@` or `$*`.** Quoted `"$@"` preserves argument boundaries. Unquoted `$@` splits arguments with spaces. `$*` merges all arguments into one string. 3. **Appended flags override earlier ones.** Many CLI tools (argparse, getopt) use last-value-wins for duplicate flags. The wrapper can append `--flag value` after `"$@"` to force overrides without removing the caller's original flags. 4. **Exit codes matter.** If pre-flight fails, exit with a non-zero code the caller understands. Some callers retry on specific exit codes (e.g., PeerTube runner retries on exit 1). 5. **Log to /tmp, not the service's log directory.** The wrapper log is a debug artifact, not part of the service's data. `/tmp` is cleaned on reboot, which is fine for wrapper logs. ### Troubleshooting - **Wrapper not being called:** Check the symlink chain (`ls -la $BINARY_PATH`). If the service uses a hardcoded absolute path that bypasses PATH, the symlink might be in the wrong location. - **Arguments with spaces break:** Use `"$@"` (quoted) in the exec line, not `$@` (unquoted). - **Service fails after wrapper install:** Check the shebang (`#!/bin/bash`), permissions (`chmod +x`), and that `exec` is present. - **Wrapper log is empty:** The service might be calling a different path. Check `which $BINARY_NAME` vs what the service config specifies. ### Example — Whisper transcription routing (PeerTube runner on cortex) The [[peertube-remote-runner]] calls `whisper-ctranslate2` for auto-captioning. The smart wrapper intercepts this to route short videos to GPU and long videos to CPU (the routing logic itself is **Pattern 2** below). ``` BINARY_NAME=whisper-ctranslate2 BINARY_PATH=/usr/local/bin/whisper-ctranslate2 WRAPPER_NAME=whisper-smart REAL_SUFFIX=-real Symlink chain: /usr/local/bin/whisper-ctranslate2 → /usr/local/bin/whisper-smart /usr/local/bin/whisper-ctranslate2-real → /home/zvx/.local/bin/whisper-ctranslate2 Wrapper logic: - ffprobe audio duration from first non-flag argument - < 1hr → exec with --device cuda --compute_type float16 - >= 1hr → exec with --device cpu --compute_type int8 - Appends --model medium after $@ (last-value-wins override) ``` --- # Pattern 2 — GPU/CPU Fallback Routing Route workloads to GPU or CPU based on pre-flight inspection of job properties (duration, file size, resolution, complexity). Small jobs go to GPU for speed; large jobs fall back to CPU to avoid VRAM exhaustion. Concurrent job control via `flock` prevents OOM kills — excess jobs fail fast and re-queue instead of competing for memory. Use this when you have a GPU workload where some jobs exceed VRAM capacity, and the system needs to handle both small and large jobs without manual intervention or OOM kills. **Deploy it via Pattern 1** (the router script *is* the wrapper's pre-flight logic). ### Prerequisites - NVIDIA GPU with working drivers (`nvidia-smi` returns output) - Both GPU and CPU execution paths available for the workload - A probe tool to inspect job properties before execution (e.g., `ffprobe`, `mediainfo`, `file`, `wc`) - A caller that retries on non-zero exit codes (scheduler, job queue, runner) ### Inputs ``` TARGET_HOST= # Machine with GPU (e.g., cortex) WORKLOAD_BINARY= # The tool that processes jobs (e.g., "whisper-ctranslate2-real") PROBE_TOOL= # Tool to inspect job properties (e.g., "ffprobe", "mediainfo") GPU_VRAM_MB= # Total VRAM available (e.g., 16384 for 16GB) WORKLOAD_VRAM_MB= # VRAM used per GPU job (e.g., 3700) WORKLOAD_RAM_MB= # RAM used per CPU job (e.g., 11000) THRESHOLD_VALUE= # Cutoff for GPU vs CPU routing (e.g., 3600 for seconds) THRESHOLD_UNIT= # What the threshold measures (e.g., "seconds", "bytes", "pixels") MAX_GPU_JOBS= # Max concurrent GPU jobs (e.g., 2) MAX_CPU_JOBS= # Max concurrent CPU jobs (e.g., 1) GPU_ARGS= # Arguments for GPU execution (e.g., "--device cuda --compute_type float16") CPU_ARGS= # Arguments for CPU execution (e.g., "--device cpu --compute_type int8") ``` ### Step 1: Determine the Routing Threshold Profile representative workloads to find the VRAM crossover point. ```bash # Run a small job on GPU, monitor VRAM ssh $TARGET_HOST "nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits" # Run the workload... check peak VRAM during execution # Run a large job on GPU, watch for OOM — if it exceeds VRAM, that's your upper bound ``` Set the threshold conservatively below the point where GPU jobs start failing. Common strategies: | Workload Type | Probe Property | Typical Threshold | |---------------|----------------|-------------------| | Audio transcription | Duration (seconds) | 1-2 hours | | Image generation | Resolution (megapixels) | Based on model VRAM curve | | Video encoding | Duration × resolution | Derived from VRAM budget | | LLM inference | Token count / context length | Model-specific | **Gate:** You must have a clear, measurable property that predicts VRAM usage. If the relationship is unpredictable, this pattern won't work — use a different strategy (e.g., try GPU first, fall back on OOM). ### Step 2: Write the Probe Function ```bash # Generic probe template probe_workload() { local INPUT="$1" local METRIC=0 if [[ -n "$INPUT" && -f "$INPUT" ]]; then # Example: audio/video duration via ffprobe METRIC=$($PROBE_TOOL -v quiet -show_entries format=duration \ -of csv=p=0 "$INPUT" 2>/dev/null | cut -d. -f1) METRIC=${METRIC:-0} # Example: file size in bytes # METRIC=$(stat -c%s "$INPUT" 2>/dev/null) # Example: image resolution (width × height) # METRIC=$($PROBE_TOOL -v quiet -show_entries stream=width,height \ # -of csv=p=0 "$INPUT" 2>/dev/null | awk -F, '{print $1*$2}') fi echo "$METRIC" } ``` **Gate:** Test the probe against known inputs: ```bash probe_workload /path/to/small/input # Should be < THRESHOLD_VALUE probe_workload /path/to/large/input # Should be >= THRESHOLD_VALUE ``` ### Step 3: Implement the Router ```bash #!/bin/bash # GPU/CPU Fallback Router — routes jobs based on $THRESHOLD_UNIT inspection THRESHOLD=$THRESHOLD_VALUE LOGFILE="/tmp/workload-router.log" GPU_LOCK="/tmp/gpu-workload.lock" CPU_LOCK="/tmp/cpu-workload.lock" # ──── Probe ──── INPUT="" METRIC=$(probe_workload "$INPUT") # ──── Route ──── if (( METRIC < THRESHOLD )); then MODE="GPU"; DEVICE_ARGS="$GPU_ARGS"; LOCK_FILE="$GPU_LOCK"; MAX_CONCURRENT=$MAX_GPU_JOBS else MODE="CPU"; DEVICE_ARGS="$CPU_ARGS"; LOCK_FILE="$CPU_LOCK"; MAX_CONCURRENT=$MAX_CPU_JOBS fi # ──── Concurrency control ──── if (( MAX_CONCURRENT == 1 )); then # Single-job lock: flock with fail-fast exec 9>"$LOCK_FILE" if ! flock --nonblock 9; then echo "[ROUTER] $(date) mode=${MODE}-BLOCKED metric=${METRIC} (slot full, exiting)" >> "$LOGFILE" exit 1 # Caller should retry later fi fi # For MAX_CONCURRENT > 1, use numbered lock files: # for i in $(seq 0 $((MAX_CONCURRENT - 1))); do # SLOT_LOCK="${LOCK_FILE}.${i}" # exec 9>"$SLOT_LOCK" # if flock --nonblock 9; then break; fi # Got a slot # if (( i == MAX_CONCURRENT - 1 )); then # echo "[ROUTER] $(date) mode=${MODE}-BLOCKED metric=${METRIC} (all slots full)" >> "$LOGFILE" # exit 1 # fi # done # ──── Log and execute ──── echo "[ROUTER] $(date) mode=$MODE metric=${METRIC} args: $@" >> "$LOGFILE" exec $WORKLOAD_BINARY "$@" $DEVICE_ARGS ``` **Key design decisions:** - **`flock --nonblock`**: Non-blocking lock attempt. If the slot is taken, exit immediately instead of waiting — this prevents queue starvation where all runner slots block waiting for CPU jobs. - **Exit code 1**: The caller (runner, scheduler) should interpret this as "retry later." Most job queues do by default. - **`exec`**: Replace the router process with the workload binary so signals, exit codes, and resource limits pass through cleanly. - **Lock files in `/tmp`**: Auto-cleaned on reboot. No stale locks after crashes. ### Step 4: Integrate with the Caller Deploy the router using **Pattern 1 (Binary Wrapper Interception)**: 1. Rename the real binary: `mv $BINARY → ${BINARY}-real` 2. Write the router script (above) as the wrapper 3. Symlink: `ln -sf /path/to/router $BINARY` Or, if the caller supports configurable command paths, point it directly at the router. ### Step 5: Verify Both Paths ```bash # GPU path — submit a small job, expect mode=GPU + GPU utilization ssh $TARGET_HOST "$BINARY " ssh $TARGET_HOST "nvidia-smi --query-gpu=utilization.gpu,memory.used --format=csv,noheader" ssh $TARGET_HOST "tail -1 /tmp/workload-router.log" # mode=GPU # CPU path — submit a large job, expect mode=CPU + idle GPU ssh $TARGET_HOST "$BINARY " ssh $TARGET_HOST "free -h" ssh $TARGET_HOST "tail -1 /tmp/workload-router.log" # mode=CPU # Concurrency — start one CPU job, immediately try a second ssh $TARGET_HOST "$BINARY &"; sleep 2 ssh $TARGET_HOST "$BINARY " # second exits immediately, code 1 ssh $TARGET_HOST "grep BLOCKED /tmp/workload-router.log" # mode=CPU-BLOCKED ``` ### Step 6: Tune and Monitor ```bash ssh $TARGET_HOST "grep -c 'mode=GPU' /tmp/workload-router.log" ssh $TARGET_HOST "grep -c 'mode=CPU' /tmp/workload-router.log" ssh $TARGET_HOST "grep -c 'BLOCKED' /tmp/workload-router.log" ``` If BLOCKED count is high relative to CPU count, the threshold may be too aggressive (routing too many jobs to CPU). Raise the threshold or increase `MAX_CPU_JOBS` if RAM allows. ### Memory Budget Worksheet ``` GPU path: VRAM per job: $WORKLOAD_VRAM_MB MB Max GPU jobs: $MAX_GPU_JOBS Total GPU VRAM: $GPU_VRAM_MB MB Headroom: GPU_VRAM_MB - (WORKLOAD_VRAM_MB × MAX_GPU_JOBS) MB → must be positive CPU path: RAM per job: $WORKLOAD_RAM_MB MB Max CPU jobs: $MAX_CPU_JOBS System RAM: $(free -m | awk '/Mem:/{print $2}') MB Other processes: ~2-4 GB (OS, services, buffers) Headroom: SystemRAM - OtherProcs - (WORKLOAD_RAM_MB × MAX_CPU_JOBS) MB → must be positive systemd MemoryMax: Set to MAX(GPU peak, CPU peak) + 20% buffer ``` ### Troubleshooting - **GPU job OOM-kills despite being under threshold:** Threshold too high, or VRAM varies by input beyond what the probe measures. Lower it or add a secondary probe (e.g., resolution in addition to duration). - **CPU jobs pile up and exhaust RAM:** `MAX_CPU_JOBS` too high, or `flock` isn't working. Check lock files in `/tmp/` and the `exec 9>` redirect. - **All jobs route to CPU:** The probe returns 0 or fails silently. Test it manually: `$PROBE_TOOL -v quiet -show_entries format=duration -of csv=p=0 /path/to/input`. Empty result usually means a permissions/path issue. - **Blocked jobs never get retried:** The caller doesn't retry on exit 1. Match the exit code to what the caller expects (some want 75 for "temporary failure"). - **Lock files persist after crash:** `/tmp` clears on reboot, so locks self-heal. For immediate cleanup: `rm /tmp/cpu-workload.lock`. ### Example — Whisper auto-captioning on PeerTube runner (cortex) ``` WORKLOAD_BINARY=/usr/local/bin/whisper-ctranslate2-real PROBE_TOOL=ffprobe GPU_VRAM_MB=16384 # RTX A4000 WORKLOAD_VRAM_MB=3700 # Whisper medium on float16 WORKLOAD_RAM_MB=11000 # Whisper medium on CPU int8 (peak for 9.5hr video) THRESHOLD_VALUE=3600 # 1 hour in seconds THRESHOLD_UNIT=seconds MAX_GPU_JOBS=2 # Runner concurrency=2, both can be GPU MAX_CPU_JOBS=1 # Only 1 CPU job at a time (11GB peak, 20G MemoryMax) GPU_ARGS="--device cuda --compute_type float16" CPU_ARGS="--device cpu --compute_type int8" Result: 4100+ videos captioned. ~20 videos over 1 hour routed to CPU. GPU jobs: ~3.7GB VRAM, 88-99% GPU utilization. CPU jobs: ~8-11GB RAM, serialized via flock. MemoryMax=20G on the runner service as safety net. ``` --- # Pattern 3 — Pre-Flight Probe Gate Insert a cheap inspection step before expensive processing in a pipeline. Probe the input (ffprobe, mediainfo, file headers, checksums) to skip work that will be wasted — wrong format, already optimized, below quality threshold, or too large to process safely. Log every decision for an audit trail. Keep a post-processing safety net as backup. Use this when your pipeline processes files in bulk and a significant percentage of inputs don't need the expensive step, or when processing the wrong input would waste time, storage, or GPU cycles. Like Pattern 2, the gate can live inside a Pattern 1 wrapper, or inline in a processing loop. ### Prerequisites - A pipeline with at least one expensive processing step (transcoding, inference, embedding, etc.) - A probe tool that can inspect inputs cheaply (< 1 second per file) - Clear criteria for what constitutes a "skip" vs "process" decision ### Inputs ``` PIPELINE_NAME= # Human-readable name (e.g., "video-transcoder", "pdf-extractor") PROBE_TOOL= # Inspection tool (e.g., "ffprobe", "mediainfo", "file", "pdfinfo") INPUT_DIR= # Where the pipeline reads inputs (e.g., "/opt/pipeline/incoming") OUTPUT_DIR= # Where processed outputs go (e.g., "/opt/pipeline/processed") SKIP_DIR= # Where skipped inputs go (e.g., "/opt/pipeline/skipped") FAIL_DIR= # Where failed inputs go (e.g., "/opt/pipeline/failed") LOG_FILE= # Decision log path (e.g., "/opt/pipeline/logs/probe-gate.log") ``` ### Step 1: Define Skip Criteria Enumerate the conditions under which a file should skip the expensive step. Be specific — vague criteria lead to false positives. | Check | Probe Command | Skip When | |-------|---------------|-----------| | Video codec | `ffprobe -show_entries stream=codec_name` | Already target codec (e.g., already HEVC) | | Audio bitrate | `ffprobe -show_entries stream=bit_rate` | Below minimum quality threshold | | Resolution | `ffprobe -show_entries stream=width,height` | Below minimum (e.g., < 360p) | | Duration | `ffprobe -show_entries format=duration` | Exceeds safe processing limit | | File size | `stat -c%s` | Zero bytes, or exceeds storage budget | | PDF pages | `pdfinfo file.pdf \| grep Pages` | Too many pages for OCR budget | | Image format | `file --mime-type` | Already target format | | Container format | `ffprobe -show_entries format=format_name` | Unsupported container | | Corruption | `ffprobe -v error` exit code | Non-zero = corrupt file | | Existing output | `test -f $OUTPUT_DIR/$(basename)` | Output already exists (dedup) | **Gate:** Write your criteria as a decision table: ``` Criterion 1: → SKIP (reason: "") Criterion 2: → SKIP (reason: "") Criterion 3: not available → SKIP (reason: "probe failed") Default: → PROCESS ``` ### Step 2: Write the Probe Gate Function ```bash #!/bin/bash # Pre-flight probe gate for $PIPELINE_NAME # Returns: 0 = process, 1 = skip, 2 = fail (corrupt/unreadable) LOGFILE="$LOG_FILE" probe_gate() { local INPUT="$1" local BASENAME=$(basename "$INPUT") local TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S') # ──── Existence check ──── if [[ ! -f "$INPUT" ]]; then echo "[$TIMESTAMP] FAIL $BASENAME reason=file_not_found" >> "$LOGFILE" return 2 fi # ──── Size check ──── local SIZE=$(stat -c%s "$INPUT" 2>/dev/null) if (( SIZE == 0 )); then echo "[$TIMESTAMP] SKIP $BASENAME reason=zero_bytes size=0" >> "$LOGFILE" return 1 fi # ──── Probe the input (adapt to your probe tool and criteria) ──── local PROBE_OUTPUT PROBE_OUTPUT=$($PROBE_TOOL "$INPUT" 2>/dev/null) local PROBE_EXIT=$? if (( PROBE_EXIT != 0 )); then echo "[$TIMESTAMP] FAIL $BASENAME reason=probe_failed exit=$PROBE_EXIT" >> "$LOGFILE" return 2 fi # ──── Apply skip criteria ──── local CODEC=$(echo "$PROBE_OUTPUT" | grep codec_name | head -1 | cut -d= -f2) if [[ "$CODEC" == "hevc" ]]; then echo "[$TIMESTAMP] SKIP $BASENAME reason=already_hevc codec=$CODEC" >> "$LOGFILE" return 1 fi local HEIGHT=$(echo "$PROBE_OUTPUT" | grep '^height=' | head -1 | cut -d= -f2) if (( HEIGHT < 240 )); then echo "[$TIMESTAMP] SKIP $BASENAME reason=below_min_resolution height=$HEIGHT" >> "$LOGFILE" return 1 fi # ──── Passed all checks ──── echo "[$TIMESTAMP] PASS $BASENAME codec=$CODEC height=${HEIGHT} size=$SIZE" >> "$LOGFILE" return 0 } ``` **Key design decisions:** - **Return codes**: 0 = process (matches shell "success" convention), 1 = skip, 2 = fail. Callers branch on `$?`. - **Structured log lines**: Every decision logged with timestamp, verdict, filename, reason — parseable by grep/awk. - **Probe errors = FAIL, not SKIP**: If the probe itself fails, the file might be corrupt — route to fail dir for inspection rather than silently skipping. ### Step 3: Integrate into the Pipeline **Option A — Inline in processing loop:** ```bash for INPUT in "$INPUT_DIR"/*; do probe_gate "$INPUT" case $? in 0) process_file "$INPUT"; mv "$INPUT" "$OUTPUT_DIR/" ;; # Expensive step 1) mv "$INPUT" "$SKIP_DIR/" ;; # Skipped — preserve for audit 2) mv "$INPUT" "$FAIL_DIR/" ;; # Failed probe — needs investigation esac done ``` **Option B — As a pre-filter in a Pattern 1 wrapper** (add the gate to the wrapper before exec): ```bash probe_gate "$INPUT_FILE" GATE_RESULT=$? if (( GATE_RESULT == 1 )); then echo "[WRAPPER] $(date) SKIPPED: $INPUT_FILE" >> "$LOGFILE" exit 0 # Success — nothing to do fi if (( GATE_RESULT == 2 )); then echo "[WRAPPER] $(date) PROBE FAILED: $INPUT_FILE" >> "$LOGFILE" exit 1 # Error — caller should retry or alert fi exec $REAL_BINARY "$@" # Gate passed — proceed with expensive processing ``` **Option C — In a Python pipeline script:** ```python import subprocess, shutil, os def probe_gate(input_path: str) -> tuple[str, dict]: """Returns (verdict, metadata) where verdict is 'process', 'skip', or 'fail'.""" if not os.path.exists(input_path): return 'fail', {'reason': 'file_not_found'} size = os.path.getsize(input_path) if size == 0: return 'skip', {'reason': 'zero_bytes', 'size': 0} result = subprocess.run( ['ffprobe', '-v', 'quiet', '-show_entries', 'stream=codec_name,height', '-of', 'flat', input_path], capture_output=True, text=True ) if result.returncode != 0: return 'fail', {'reason': 'probe_failed', 'exit': result.returncode} # Parse and apply criteria... return 'process', {'codec': codec, 'height': height, 'size': size} ``` ### Step 4: Add Post-Processing Safety Net The probe gate is the primary filter, but add a post-processing check as backup. This catches cases where the probe was wrong (e.g., file reported as H.264 but was actually corrupt). ```bash post_process_check() { local OUTPUT="$1" local TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S') # Size gate: output should be at least 10% of input size local INPUT_SIZE=$2 local OUTPUT_SIZE=$(stat -c%s "$OUTPUT" 2>/dev/null) if (( OUTPUT_SIZE < INPUT_SIZE / 10 )); then echo "[$TIMESTAMP] POST-FAIL $OUTPUT reason=output_too_small input=${INPUT_SIZE} output=${OUTPUT_SIZE}" >> "$LOGFILE" return 1 fi # Integrity check: verify output is valid $PROBE_TOOL -v error "$OUTPUT" 2>/dev/null if (( $? != 0 )); then echo "[$TIMESTAMP] POST-FAIL $OUTPUT reason=output_corrupt" >> "$LOGFILE" return 1 fi echo "[$TIMESTAMP] POST-PASS $OUTPUT size=$OUTPUT_SIZE" >> "$LOGFILE" return 0 } ``` ### Step 5: Reporting ```bash echo "=== Probe Gate Report ===" echo "Processed: $(grep -c ' PASS ' $LOG_FILE)" echo "Skipped: $(grep -c ' SKIP ' $LOG_FILE)" echo "Failed: $(grep -c ' FAIL ' $LOG_FILE)" echo "Skip reasons:" grep ' SKIP ' $LOG_FILE | grep -oP 'reason=\S+' | sort | uniq -c | sort -rn echo "Failed files:" grep ' FAIL ' $LOG_FILE | tail -10 ``` ### Troubleshooting - **Probe is slow (> 1s/file):** Some tools read more than necessary. For ffprobe, use `-analyzeduration 1000000 -probesize 1000000`. For large PDFs, `pdfinfo` beats opening the file in Python. - **Probe reports wrong codec/format:** Container and stream codecs can mismatch. Probe the stream level: `ffprobe -v quiet -select_streams v:0 -show_entries stream=codec_name -of csv=p=0 "$INPUT"`. - **Skipped files that should have been processed:** Review the skip log; lower the threshold or add exceptions. The skip directory preserves files for re-processing if criteria change. - **Post-processing catches failures the probe missed:** The safety net working as intended. Investigate why the probe missed it; add a new criterion if the pattern is common. ### Example A — PeerTube H.265 transcoding pipeline (cortex) ``` PIPELINE_NAME=video-transcoder PROBE_TOOL=ffprobe INPUT_DIR=/opt/bulk-import/completed OUTPUT_DIR=/opt/bulk-import/transcoded SKIP_DIR=/opt/bulk-import/skipped FAIL_DIR=/opt/bulk-import/failed Probe criteria: - codec_name == "hevc" → SKIP (already H.265) - height < 240 → SKIP (too low quality to bother) - duration == 0 → FAIL (corrupt or audio-only) - probe exit != 0 → FAIL (unreadable) Post-processing safety net: - output size < 10% of input → FAIL (transcode produced garbage) - ffprobe on output fails → FAIL (corrupt output) Result: Saved ~15% of GPU cycles by skipping already-optimized files. ``` ### Example B — PDF extraction pipeline (RECON on VM 1130) ``` PIPELINE_NAME=pdf-extractor PROBE_TOOL=pdfinfo INPUT_DIR=/mnt/library/incoming OUTPUT_DIR=/opt/recon/extracted Probe criteria: - Pages > 500 → route to Gemini Vision (OCR too slow) - Pages == 0 → FAIL (corrupt PDF) - File size < 1KB → SKIP (empty/placeholder) - Encrypted: yes → SKIP (can't extract without password) - Already in SQLite status table → SKIP (dedup) ``` --- *Merged 2026-06-15 from `binary-wrapper-interception.md`, `gpu-cpu-fallback-routing.md`, and `pipeline-probe-gate.md` (originally written 2026-02-17).*