Fire-and-Forget on a Cron That Never Loses a Result: Reclaiming Gemini Background Executions with a Submission Ledger

import sqlite3, time
 
def open_ledger(path="reclaim_ledger.db"):
    db = sqlite3.connect(path, isolation_level=None)  # autocommit
    db.execute("PRAGMA journal_mode=WAL")
    db.execute("""
      CREATE TABLE IF NOT EXISTS jobs (
        idem         TEXT PRIMARY KEY,   -- idempotency key (uniqueness of a logical job)
        handle       TEXT,               -- Interactions API handle name
        status       TEXT NOT NULL,      -- reserving / submitted / done / consumed / failed
        submitted_at REAL,
        updated_at   REAL NOT NULL
      )
    """)
    db.execute("CREATE INDEX IF NOT EXISTS idx_status ON jobs(status)")
    return db

import hashlib, json
 
def idempotency_key(job: dict) -> str:
    canonical = json.dumps(job, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:32]
 
def reserve(db, idem) -> bool:
    try:
        db.execute(
            "INSERT INTO jobs(idem, status, updated_at) VALUES (?, 'reserving', ?)",
            (idem, time.time()),
        )
        return True          # new reservation succeeded
    except sqlite3.IntegrityError:
        return False         # already exists (= prevented a double submit)

from google import genai
 
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
def submit(db, job: dict):
    idem = idempotency_key(job)
    if not reserve(db, idem):
        return  # this job is already in flight; do nothing (idempotent)
 
    try:
        op = client.interactions.create(
            model="gemini-flash-latest",
            input=job["input"],
            background=True,            # fire and exit
            metadata={"idem": idem},    # key for reverse lookup later
        )
    except Exception:
        # if the submit failed, roll the reservation back so the next tick can resend
        db.execute("UPDATE jobs SET status='reserving', updated_at=? WHERE idem=?",
                   (time.time(), idem))
        raise
 
    db.execute(
        "UPDATE jobs SET handle=?, status='submitted', submitted_at=?, updated_at=? WHERE idem=?",
        (op.name, time.time(), time.time(), idem),
    )

def reclaim(db, on_result):
    rows = db.execute(
        "SELECT idem, handle FROM jobs WHERE status='submitted' AND handle IS NOT NULL"
    ).fetchall()
 
    for idem, handle in rows:
        op = client.interactions.get(name=handle)
        if not op.done:
            continue                         # still running; revisit next tick
 
        if getattr(op, "error", None):
            db.execute("UPDATE jobs SET status='failed', updated_at=? WHERE idem=?",
                       (time.time(), idem))
            continue
 
        # done. advance to 'done' before the side effect to guard against double processing
        db.execute("UPDATE jobs SET status='done', updated_at=? WHERE idem=?",
                   (time.time(), idem))
        on_result(idem, op.response)         # the side effect: persist, publish, etc.
        db.execute("UPDATE jobs SET status='consumed', updated_at=? WHERE idem=?",
                   (time.time(), idem))

def resume_unfinished(db, on_result):
    # advanced to 'done' but never 'consumed' = the process died inside the side effect
    rows = db.execute("SELECT idem, handle FROM jobs WHERE status='done'").fetchall()
    for idem, handle in rows:
        op = client.interactions.get(name=handle)
        on_result(idem, op.response)
        db.execute("UPDATE jobs SET status='consumed', updated_at=? WHERE idem=?",
                   (time.time(), idem))

def recover_orphans(db, max_age=120):
    stale = db.execute(
        "SELECT idem FROM jobs WHERE status='reserving' AND updated_at < ?",
        (time.time() - max_age,),
    ).fetchall()
    if not stale:
        return
    stale_keys = {row[0] for row in stale}
 
    # reverse-look-up real API-side jobs by idempotency key, write the handle back
    for op in client.interactions.list(filter="background=true"):
        idem = (op.metadata or {}).get("idem")
        if idem in stale_keys:
            db.execute(
                "UPDATE jobs SET handle=?, status='submitted', updated_at=? WHERE idem=?",
                (op.name, time.time(), idem),
            )
            stale_keys.discard(idem)
 
    # whatever stays in stale_keys has no API-side job = the submit itself failed.
    # leave it as 'reserving' so the next submit resends it

def tick(jobs_for_today):
    db = open_ledger()
    # 1) sweep up the leftovers first (orphans, last run's submissions, half-done side effects)
    recover_orphans(db)
    reclaim(db, on_result=persist_article)
    resume_unfinished(db, on_result=persist_article)
    # 2) submit this run's work (idempotent, so a double launch is safe)
    for job in jobs_for_today:
        submit(db, job)
    db.close()