Production AI Automation Notes #5: SQLite dedup + crash safety for cross-run AI pipelines \u2014 WAL mode, seen_items, audit log

Production AI Automation Notes #5: SQLite dedup + crash safety for cross-run AI pipelines (WAL mode, seen_items table, audit log)

Last tested: May 2026: modernc.org/sqlite, Go 1.25, draftcat v0.1

Follow @renezander030 for the rest of the series. One entry per week, all backed by code from shipping pipelines.

A scheduled AI pipeline that polls Gmail every five minutes has two failure modes the demo version ignores: it re-processes the same inputs every tick, and it loses in-flight state when the host reboots. Both are solved with the same boring tool. SQLite, configured correctly.

Part 5 of Production AI Automation Notes. Reference implementation: draftcat (Go, MIT). All code below is lifted verbatim from state.go.

• #1 Agent Approval Gates: draft, validate, approve, dispatch, audit
• #2 Token Budgets: publishing today, link will be added on release
• #3 Agentic Task System: vector recall for long-running agents
• #4 Driving CapCut from LLM agents: deterministic dispatch for video drafts
• #5 SQLite dedup + crash safety: this entry

Three lines, once at startup. Skipping any costs data on the first hard reboot.

if _, err := db.Exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=5000;`); err != nil {
    return nil, fmt.Errorf("state store pragmas: %w", err)
}

journal_mode=WAL flips SQLite from the default rollback journal to a write-ahead log. Readers stop blocking writers. A crash leaves a clean main DB file plus a WAL segment the next open replays. Without WAL mode, a pipeline reading its own state while another tick writes serializes on the file lock.

synchronous=NORMAL keeps fsync on transaction commit but skips the fsync after each WAL frame. The OS flushes on checkpoint. You trade "every byte durable before commit" for roughly an order of magnitude in write throughput. Acceptable for a pipeline state store, because a crash window of seconds is fine when upstream items are still in Gmail and re-fetchable.

busy_timeout=5000 tells SQLite to retry for five seconds before returning SQLITE_BUSY. Default is zero. Without it, two pipelines writing in the same millisecond produce spurious errors. Five seconds is long enough that real contention resolves, short enough that a genuinely stuck writer surfaces.

Four columns, one composite primary key.

CREATE TABLE IF NOT EXISTS seen_items (
    pipeline TEXT NOT NULL,
    scope    TEXT NOT NULL,
    item_id  TEXT NOT NULL,
    seen_at  INTEGER NOT NULL,
    PRIMARY KEY (pipeline, scope, item_id)
);

The composite key gives per-pipeline isolation. Two pipelines polling the same Gmail account never conflict on message ids, because each writes into its own row space. The scope column splits a pipeline into channels: one pipeline can dedup gmail and ghl_contacts separately in the same table.

seen_at is a Unix timestamp, used for forensics and TTL sweeping. Dedup logic never reads it.

The biggest design decision is when you write the seen marker: after fetch, or after the pipeline finishes successfully. Draftcat writes at fetch.

The reason is asymmetric cost. A message fetched, processed, sent to Telegram, then failing on a later step has already fired the human-visible side effect. Marking seen only on success means the next tick re-fetches, re-runs the LLM, and re-sends. A duplicate notification is more expensive than a dropped one.

Rule: mark seen the moment fetch returns. If the downstream blows up, the item is gone. Replay is an operator command, and the audit table records what was dropped.

Append-only, one row per pipeline execution.

CREATE TABLE IF NOT EXISTS pipeline_runs (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    pipeline    TEXT NOT NULL,
    started_at  INTEGER NOT NULL,
    ended_at    INTEGER NOT NULL,
    status      TEXT NOT NULL,
    error_text  TEXT
);
CREATE INDEX IF NOT EXISTS idx_runs_pipeline ON pipeline_runs(pipeline, started_at DESC);

Each row records pipeline name, start and end timestamps, status (ok/error), and error text. The /status pipeline_name operator command reads this table newest-first. When something breaks, scroll back to the last green run, and the gap to the first red run is your investigation window.

FilterUnseen is one IN query against the composite key. SQLite plans the expanded placeholder list well below a batch size of fifty.

func (s *StateStore) FilterUnseen(pipeline, scope string, ids []string) ([]string, error) {
    if len(ids) == 0 {
        return nil, nil
    }
    placeholders := strings.TrimRight(strings.Repeat("?,", len(ids)), ",")
    q := fmt.Sprintf(`SELECT item_id FROM seen_items WHERE pipeline=? AND scope=? AND item_id IN (%s)`, placeholders)
    args := make([]interface{}, 0, len(ids)+2)
    args = append(args, pipeline, scope)
    for _, id := range ids {
        args = append(args, id)
    }
    rows, err := s.db.Query(q, args...)
    if err != nil {
        return nil, err
    }
    defer rows.Close()
    seen := make(map[string]struct{}, len(ids))
    for rows.Next() {
        var id string
        if err := rows.Scan(&id); err != nil {
            return nil, err
        }
        seen[id] = struct{}{}
    }
    if err := rows.Err(); err != nil {
        return nil, err
    }
    out := make([]string, 0, len(ids))
    for _, id := range ids {
        if _, ok := seen[id]; !ok {
            out = append(out, id)
        }
    }
    return out, nil
}

MarkSeen runs one transaction with a prepared statement reused across the batch. INSERT OR IGNORE swallows duplicates, the right semantics for a set-membership table.

func (s *StateStore) MarkSeen(pipeline, scope string, ids []string) error {
    if len(ids) == 0 {
        return nil
    }
    tx, err := s.db.Begin()
    if err != nil {
        return err
    }
    stmt, err := tx.Prepare(`INSERT OR IGNORE INTO seen_items (pipeline, scope, item_id, seen_at) VALUES (?, ?, ?, ?)`)
    if err != nil {
        _ = tx.Rollback()
        return err
    }
    defer stmt.Close()
    now := time.Now().Unix()
    for _, id := range ids {
        if _, err := stmt.Exec(pipeline, scope, id, now); err != nil {
            _ = tx.Rollback()
            return err
        }
    }
    return tx.Commit()
}

Both methods are safe to call concurrently. The busy_timeout=5000 pragma absorbs cross-pipeline contention.

Every fetch step follows the same pattern: fetch a batch, dedup, short-circuit if empty, store for downstream steps, mark seen. This is the gmail_unread step from main.go:

case "gmail_unread":
    emails, err := gmail.FetchUnread(10)
    if err != nil {
        return fmt.Errorf("[step:%s] gmail fetch failed: %w", step.Name, err)
    }
    if len(emails) == 0 {
        log.Printf("no unread emails, skipping pipeline")
        return nil
    }
    emails = dedupByID(pipeline.Name, "gmail", emails,
        func(e Email) string { return e.ID })
    if len(emails) == 0 {
        log.Printf("all unread emails already processed, skipping pipeline")
        return nil
    }
    data["emails"] = FormatEmailsForPrompt(emails)

dedupByID is a generic helper wrapping FilterUnseen + MarkSeen with a key function. If the state layer is misconfigured or locked, dedup degrades to "process everything" and logs a warning, on purpose. Silently dropping items on a transient state-store issue is worse than running the LLM twice.

• Cross-host coordination. A SQLite file lives on one filesystem. For pipelines across two hosts, switch to Postgres with the same schema, use Redis with a TTL set, or accept that the duplicate-process window equals your sync interval.
• Network-attached storage. WAL mode assumes OS lock primitives work. NFS lies about locks. Keep the SQLite file on a local disk.
• Long-term growth. seen_items grows forever unless you sweep. A cron deleting rows where seen_at < now - 90 days is enough for most pipelines.
• Replay semantics. Marking seen at fetch means a failed downstream is a dropped item. If your domain cannot tolerate that, write seen on success and add an explicit dedup window on the side effect itself.

draftcat ships:

• state.go: full StateStore, schema init, FilterUnseen, MarkSeen, RecordRun, RecentRuns
• main.go: dedupByID helper and deterministic steps (gmail_unread, ghl_new_contacts, ghl_new_conversations)
• pipelines/: YAML configs wiring dedup into real pipelines
• README.md: operator commands including /status

MIT license. Go 1.25, pure-Go SQLite via modernc.org/sqlite so there is no CGO requirement.

If you ship something similar with a different storage layer, drop a comment with the rough shape: Postgres advisory locks, Redis Bloom filter, Cloudflare KV with versioned keys, in-memory hash flushed on shutdown. Curious how multi-host setups handle the "two workers fetched the same item in the same tick" race without a queue broker.

• Production AI Automation Notes #1: Agent Approval Gates: the gate pattern this state store underlies
• Production AI Automation Notes #3: Agentic Task System: vector recall pairs with this dedup layer
• Production AI Automation Notes #4: Driving CapCut from LLM agents: deterministic dispatch with the same audit log shape
• pipeline_engine.go walkthrough: the engine that calls FilterUnseen and MarkSeen between steps
• YAML config for AI pipelines: how the scope strings get assigned per fetch action

2026-05-25: initial publish