投げて終了する定期実行で、結果を取りこぼさない — Gemini バックグラウンド実行を再取得台帳で回す設計

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,   -- 冪等キー（論理ジョブの一意性）
        handle       TEXT,               -- Interactions API のハンドル名
        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          # 新規予約に成功
    except sqlite3.IntegrityError:
        return False         # 既に存在（＝二重送信を防いだ）

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  # 同じジョブは既に投入済み。何もしない（冪等）
 
    try:
        op = client.interactions.create(
            model="gemini-flash-latest",
            input=job["input"],
            background=True,            # 投げて終了する
            metadata={"idem": idem},    # 後で逆引きするための鍵
        )
    except Exception:
        # 送信に失敗したら予約を戻し、次ティックで再送できるようにする
        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                         # まだ処理中。次ティックで再訪
 
        if getattr(op, "error", None):
            db.execute("UPDATE jobs SET status='failed', updated_at=? WHERE idem=?",
                       (time.time(), idem))
            continue
 
        # 完了。後段へ渡す前に done へ進めて二重処理を防ぐ
        db.execute("UPDATE jobs SET status='done', updated_at=? WHERE idem=?",
                   (time.time(), idem))
        on_result(idem, op.response)         # ここで保存・公開などの副作用
        db.execute("UPDATE jobs SET status='consumed', updated_at=? WHERE idem=?",
                   (time.time(), idem))

def resume_unfinished(db, on_result):
    # done まで進んだが consumed されていない＝副作用の途中で落ちたジョブ
    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}
 
    # API 側に実在するジョブを冪等キーで逆引きし、台帳へハンドルを書き戻す
    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)
 
    # ここで残った stale_keys は API 側にも無い＝送信自体が失敗。
    # reserving のまま残し、次の submit で再送させる

def tick(jobs_for_today):
    db = open_ledger()
    # 1) まず取りこぼしを拾う（孤児・前回投げたぶん・副作用の途中落ち）
    recover_orphans(db)
    reclaim(db, on_result=persist_article)
    resume_unfinished(db, on_result=persist_article)
    # 2) 今回ぶんを投げる（冪等なので二重起動しても安全）
    for job in jobs_for_today:
        submit(db, job)
    db.close()