My AI agent had one job: write an article, post it to a publishing platform, then clean up. Three times out of ten, it would crash mid-pipeline, leave a draft orphaned, lose the temp files, and try to publish the same article again fresh. The result: duplicate drafts, half-failed posts, and — worst case — two identical articles live on the same feed.
The first time it happened, I thought it was a fluke. A network timeout during image upload, no big deal. The second time, I blamed the API rate limit. The third time, I sat and watched the logs scroll by. The agent created a draft, got a 503 error during the image upload step, then on retry it couldn't find the temp files — they were already cleaned up by a failed cleanup handler. So it created a brand new draft with a slightly different title. Now I had three drafts, two of which were the same article, and no way to tell which one was the canonical version.
The problem wasn't the network. The problem was that every step of the pipeline assumed the previous step had succeeded. There was no shared state, no recovery mechanism, no way to answer "what phase are we in?" after a crash. Each run was a blank slate, and blank slates don't know they're repeating work.
I needed a way to make the pipeline crash-safe. Not "eventually consistent" crash-safe. Atomic crash-safe. Here's the state machine pattern that did it.
I split the pipeline into phases, each with a single responsibility. A run ID ties everything together. A manifest file tracks which temp files belong to which run. The state machine refuses to advance unless the current phase's gates are all green.
Here's the core data model. Every field in RunState is written to disk before and after each phase — that's the atomicity guarantee:
from __future__ import annotations
import json
from pathlib import Path
from enum import Enum
from datetime import datetime
from typing import Optional
class Phase(Enum):
LOAD = ""
SELECT = "topic_selection"
DEDUP = "topic_dedup"
WRITE = "article_generation"
GATE = "de_ai_gate"
VALIDATE = "pre_publish_validate"
PUBLISH = "publish"
VERIFY = "post_publish_verify"
CLEAN = "cleanup"
class RunState:
def __init__(self, run_id: str, manifest_path: Path):
self.run_id = run_id
self.manifest_path = manifest_path
self.current_phase = None
self.phase_results = {}
def save(self):
data = {
"run_id": self.run_id,
"current_phase": self.current_phase.value if self.current_phase else None,
"phase_results": self.phase_results,
"updated_at": datetime.now().isoformat(),
}
self.manifest_path.write_text(json.dumps(data, ensure_ascii=False, indent=2))
@classmethod
def load(cls, manifest_path: Path):
if not manifest_path.exists():
return None
data = json.loads(manifest_path.read_text(encoding="utf-8"))
state = cls(data["run_id"], manifest_path)
state.current_phase = Phase(data["current_phase"]) if data["current_phase"] else None
state.phase_results = data["phase_results"]
return state
The Pipeline class runs the loop. Each phase handler is a method named _phase_{name}
. Before every handler, the state is saved. After every handler, the result is saved. If the process dies between two phases, the next run picks up exactly where it left off:
class Pipeline:
def __init__(self, state: RunState, temp_files: list[Path]):
self.state = state
self.temp_files = temp_files
self.guard = Guard(temp_files)
def run(self, phases: list[Phase]):
for phase in phases:
print(f"[{self.state.run_id}] Phase: {phase.value}")
self.state.current_phase = phase
self.state.save()
handler = getattr(self, f"_phase_{phase.value}")
handler()
self.state.phase_results[phase.value] = True
self.state.save()
def _phase_cleanup(self):
for f in self.temp_files:
if f.exists():
f.unlink()
print(f" cleaned: {f.name}")
self.state.manifest_path.unlink(missing_ok=True)
The Guard class solves a more subtle problem. When your publishing pipeline also has a "skill consistency" checker (like a linter that flags unexpected files), temp files get flagged as errors. The Guard registers every temp file upfront so the checker knows to ignore them:
class Guard:
def __init__(self, allowed_temps: list[Path]):
self.allowed = {p.resolve() for p in allowed_temps}
def is_temp_artifact(self, path: Path) -> bool:
return path.resolve() in self.allowed
def verify_no_untracked_writes(self, workspace: Path):
for f in workspace.iterdir():
if f.is_file() and not self.is_temp_artifact(f):
raise RuntimeError(
f"Untracked file: {f.name}. "
"Add to temp manifest or refactor."
)
Three design decisions make this pattern crash-safe in practice.
First, disk-first state. The state machine writes to disk before AND after every phase handler. If the agent crashes during _phase_validate()
, the next run picks up manifest.json
, sees current_phase: "pre_publish_validate"
, and knows exactly where to resume. No "where was I?" guessing.
I tested this by killing the process mid-run. I killed it during _phase_write
, during _phase_publish
, and during _phase_cleanup
. Every time, the next run read the manifest, found the phase it was on, and resumed from that exact point. No duplicate drafts, no orphaned temp files, no skipped phases. The key insight: the save happens before the handler, so even if the handler fails halfway, the manifest already records that we've entered this phase.
Second, the temp-file manifest. Before the pipeline starts, every file it will create is registered. The Guard class enforces this: if a write happens to a file not in the manifest, the pipeline halts. This caught me twice — once when I used write_file instead of open() in Python (the write_file tool created a file the Guard didn't know about), and once when a logging library created a .log file in the working directory.
The manifest also serves as a recovery checklist. When the pipeline resumes after a crash, it compares the current filesystem state against the manifest. Missing files get recreated. Extra files get flagged. It's a poor man's filesystem snapshot that costs exactly one JSON write per phase.
Third, idempotent phase handlers. Each phase function checks if its work is already done before doing it again. _phase_dedup()
reads the existing article list and only writes new data if it hasn't already. This means replaying Phase 2 after a crash is safe — it won't double-check or double-write.
def _phase_dedup(self):
dedup_flag = self.state.manifest_path.parent / f"{self.state.run_id}_dedup_done"
if dedup_flag.exists():
print(" dedup already done, skipping")
return
dedup_flag.write_text("done")
The idempotency trick is the flag file. Each phase writes a {run_id}_{phase}_done
marker file. The phase handler checks for its marker first. If the marker exists, the phase is skipped. This is a belt-and-suspenders approach on top of the state machine's phase tracking — because the state machine tracks the current phase, but the flag file tracks whether the phase's work was actually completed. One tracks intent, the other tracks execution.
Windows file locking is different. On Linux, fcntl.flock() works fine. On Windows, you need msvcrt.locking(). I hit this when two pipeline runs accidentally overlapped (I was testing recovery logic) and tried to write the same manifest file. The fix was a cross-platform lock:
import sys
from pathlib import Path
def safe_write(path: Path, data: str):
if sys.platform == "win32":
import msvcrt
with open(path, "w", encoding="utf-8") as f:
msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)
f.write(data)
msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
else:
import fcntl
with open(path, "w", encoding="utf-8") as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
f.write(data)
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
The manifest file itself is a temp file. If you include it in the temp-file manifest, the Guard will let it through. But the _phase_cleanup
handler deletes the manifest last — after all other temps are gone — so a crash during cleanup doesn't lose the state record. I learned this the hard way: I deleted the manifest first, then the process crashed on the next temp file. On restart, there was no manifest, so the pipeline started fresh and created a duplicate article.
State machines are only as good as their phase boundaries. I initially made phases too granular (14 phases for a 10-step pipeline). The state file was overwritten constantly, and the I/O overhead added ~2 seconds per run. I collapsed related steps: de_ai_gate + validate became one phase, publish + verify became another. Down to 7 phases, no perceptible overhead.
The first run is always special. When RunState.load() returns None (no manifest exists), the pipeline should start fresh — but it needs to generate a run_id and create the manifest. The first phase handler needs to bootstrap the state. I forgot this and my initial run crashed because _phase_load()
tried to read a manifest that didn't exist yet.
File paths on Windows vs Unix. On Linux, temp files go to /tmp/
. On Windows, there's no /tmp
— you need to use the working directory or a temp subdirectory. I hardcoded /tmp/devto_article.json
in the first version and it failed silently on Windows. The fix: always use relative paths resolved against the project root.
Don't let the manifest grow stale. If you update the pipeline code (add a new phase, rename an existing one), the old manifest file becomes incompatible. The RunState.load() method will fail trying to parse a phase name that no longer exists. I added a version field to the manifest data and a migration mechanism: if the version doesn't match the current code version, the manifest is discarded and the pipeline starts fresh.
I've seen people use SQLite for pipeline state, Redis for distributed pipelines, and even JSON files on S3. My approach is deliberately minimal — a single JSON file, a guard class, and a loop. It works because the pipeline is single-threaded and runs on one machine.
What's your approach to crash-safe automation? Do you reach for a workflow engine (Temporal, Prefect) or roll your own state machine? I'd love to hear what's worked for you.
I also publish practical engineering notes on Dispatch.