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.
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.
The Fix
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.
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 = "loading"
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 state machine itself is a simple loop. Each phase is a function that returns True (advance) or raises an exception (halt). Before advancing, the state is saved to disk.
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 sits between the pipeline and the file system. It knows exactly which files are temporary publish artifacts and ignores them during skill-consistency checks.
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."
)
Why This Works
Three design decisions make this pattern crash-safe.
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.
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, and once when a logging library created a .log file in the working directory.
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):
manifest = self.state.manifest_path
dedup_flag = manifest.parent / f"{self.state.run_id}_dedup_done"
if dedup_flag.exists():
print(" dedup already done, skipping")
return
# ... actual dedup logic ...
dedup_flag.write_text("done")
Gotchas
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 and tried to write the same manifest file. The fix was a cross-platform lock:
import sys
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.
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.
What about you?
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.
Source: DEV Community
