Chen Yuan

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

Enter fullscreen mode Exit fullscreen mode

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)

Enter fullscreen mode Exit fullscreen mode

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."
                )

Enter fullscreen mode Exit fullscreen mode

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")

Enter fullscreen mode Exit fullscreen mode

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)

Enter fullscreen mode Exit fullscreen mode

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.