When an AI agent runs a multi-step pipeline, every step creates temporary files. Article drafts, image uploads, JSON payloads, log files. Over fifty runs, these files accumulate. Some get cleaned up, some don't. And the ones that don't cause the next run to fail in confusing ways.
I hit this exact problem with my publishing pipeline. A failed cleanup from run #12 left a stale devto_article.json in the working directory. Run #13 picked it up, parsed it, and published a draft with last week's title. The logs showed "JSON loaded successfully" — which was technically true. The file was valid JSON. It just belonged to the wrong run.
The fix was a Guard class that sits between the pipeline and the filesystem. Every file the pipeline creates must be registered before the pipeline starts. Any file that appears without registration halts the pipeline immediately. Run identity gets embedded into every file, so even if a cleanup fails, the next run can tell the file doesn't belong.
The Problem With Temp Files
Temp files are invisible by design. You create them, use them, delete them. But when deletion fails — file lock, process crash, permission error — the file becomes a ghost. It exists on disk but nobody remembers it's there.
The next run scans the directory, finds the ghost, and treats it as intentional. This is especially dangerous for JSON files because they're always valid. A stale manifest.json looks identical to a fresh one. The only difference is the content, and the loader doesn't check content provenance.
Here's a concrete example from my pipeline:
# The naive approach — just check if the file exists
def load_manifest(path):
if not path.exists():
return None
return json.loads(path.read_text())
This code returns valid data from any run, any day, any context. It answers "can I read this file?" but not "should I read this file?" That distinction is the entire bug.
The Guard Pattern
The Guard class solves this by requiring every temp file to be registered before the pipeline starts. Registration means the file path is added to an allowlist. Any write to a path outside the allowlist raises an error.
from pathlib import Path
class Guard:
def __init__(self, allowed_temps: list[Path]):
self.allowed = {p.resolve() for p in allowed_temps}
self.seen = set()
def register(self, path: Path):
self.allowed.add(path.resolve())
def check(self, path: Path):
resolved = path.resolve()
if resolved not in self.allowed:
raise RuntimeError(f"Untracked file: {path.name}")
self.seen.add(resolved)
def verify_cleanup(self):
remaining = self.allowed - self.seen
if remaining:
for p in remaining:
if p.exists():
print(f" Warning: {p.name} was registered but never created")
The Guard also tracks which files were actually created versus which were just registered. After the pipeline finishes, verify_cleanup() reports files that were registered but never materialized — a sign that a step was skipped or failed silently.
Run Identity in Every File
The second layer of defense is run identity. Every temp file includes the run ID either in its name or as a field in its content. This makes stale files self-identifying.
from datetime import datetime
import json
from pathlib import Path
class RunScopedFile:
def __init__(self, run_id: str, base_dir: Path):
self.run_id = run_id
self.base_dir = base_dir / run_id
self.base_dir.mkdir(parents=True, exist_ok=True)
def path_for(self, name: str) -> Path:
return self.base_dir / name
def write_json(self, name: str, data: dict):
path = self.path_for(name)
data["_run_id"] = self.run_id
data["_created_at"] = datetime.utcnow().isoformat()
path.write_text(json.dumps(data, indent=2))
return path
def load_json(self, name: str, expected_run_id: str):
path = self.path_for(name)
if not path.exists():
return None
data = json.loads(path.read_text())
if data.get("_run_id") != expected_run_id:
raise ValueError(f"File {name} belongs to run {data.get('_run_id')}, not {expected_run_id}")
return data
Each run gets its own directory. Files from different runs never collide. The _run_id field inside the JSON provides a secondary check — even if a file ends up in the wrong directory, the content proves it doesn't belong.
The Atomic Write Pattern
Race conditions between the planner (who writes the manifest) and the worker (who reads it) are another source of stale data. If the worker reads the file while the planner is still writing, it gets a partial or inconsistent state.
The fix is atomic write: write to a temp file, flush to disk, then rename.
import os
def atomic_write(path: Path, data: str):
temp = path.with_suffix(".tmp")
temp.write_text(data, encoding="utf-8")
temp.flush()
os.fsync(temp.fileno())
os.replace(temp, path)
The worker never watches the .tmp file. It waits for the final path. If the file appears, it's complete. No polling, no partial reads, no corrupted state.
Cross-Platform File Locking
On Windows, file locking is different from Linux. The fcntl module doesn't exist. The msvcrt module provides locking() instead. My pipeline needs to run on both platforms, so I use a conditional import.
import sys
def acquire_lock(path: Path):
if sys.platform == "win32":
import msvcrt
fd = os.open(path, os.O_RDWR | os.O_CREAT)
msvcrt.locking(fd, msvcrt.LK_LOCK, 1)
return fd
else:
import fcntl
fd = os.open(path, os.O_RDWR | os.O_CREAT)
fcntl.flock(fd, fcntl.LOCK_EX)
return fd
def release_lock(fd):
if sys.platform == "win32":
import msvcrt
msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
else:
import fcntl
fcntl.flock(fd, fcntl.LOCK_UN)
os.close(fd)
This ensures that even if two pipeline instances overlap, they can't write to the same file simultaneously. The lock is acquired before any read or write and released after the operation completes.
Putting It All Together
The full pipeline startup now looks like this:
def run_pipeline(run_id: str, work_dir: Path):
scoped = RunScopedFile(run_id, work_dir)
guard = Guard([
scoped.path_for("manifest.json"),
scoped.path_for("article.md"),
scoped.path_for("payload.json"),
])
# Write manifest atomically
manifest = {"title": "...", "target": "hashnode", "body": "..."}
manifest_path = scoped.write_json("manifest.json", manifest)
guard.check(manifest_path)
# Verify on load
loaded = scoped.load_json("manifest.json", run_id)
guard.check(manifest_path)
return loaded
Every file is accounted for. Every file knows which run it belongs to. Every write is atomic. Every lock is cross-platform.
What Changed After 50 Runs
Before this pattern, my pipeline had a stale file incident roughly every 10 runs. The most common failure mode was a JSON file from a previous run being picked up by the next run. The logs showed nothing wrong because the file was valid JSON and the loader treated existence as freshness.
After implementing the Guard + run-scoped directories + atomic writes, I've gone 50 consecutive runs without a single temp-file-related failure. The fix didn't make the pipeline faster — it made failures impossible to miss. A mismatched run ID is a hard error. A file outside the allowlist is a hard error. A partial write from a crashed process leaves a .tmp file that the worker ignores.
The most important change was psychological. Before, I never knew if a temp file was safe to delete. After, I knew exactly which files belonged to which run. The Guard provided a single source of truth for the filesystem state.
Stale temp files are a silent failure mode. The Guard pattern makes them noisy.
What About You
Have you ever had a stale temp file cause a pipeline to execute the wrong plan? How do you track which files belong to which run? I'd love to hear about your approach.
Originally published on Dispatch.
Top comments (0)