echo6-docs/vault/runbooks/pipeline-patterns.md

---
title: "Pipeline & Wrapper Patterns"
type: runbook
tags: []
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'
<paste wrapper script here>
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 <typical args here>"
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 <service-name>"
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 <original-target> $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="<extract from $@>"
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 <small-input-args>"
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 <large-input-args>"
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 <large-input-1> &"; sleep 2
ssh $TARGET_HOST "$BINARY <large-input-2>"            # 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: <property> <operator> <value> → SKIP (reason: "<why>")
Criterion 2: <property> <operator> <value> → SKIP (reason: "<why>")
Criterion 3: <property> 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 <probe-specific-flags> "$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).*