プロンプトを直した後、過去の成果物をどこまで作り直すか — 予算で区切る計画的バックフィルの設計

# edit_guard.py — 人手修正の検知
# 何を解決するか: バックフィルが手修正済みの成果物を旧に復す事故を止める
import hashlib
 
def content_hash(text: str) -> str:
    normalized = " ".join(text.split())  # 空白差で誤検知しない
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
 
def is_human_edited(current_text: str, stored_hash: str | None) -> bool:
    if stored_hash is None:
        # ハッシュ未記録の旧データは「編集済み扱い」に倒す（安全側）
        return True
    return content_hash(current_text) != stored_hash
 
def save_generated(conn, item_id: str, text: str, prompt_ver: int) -> None:
    conn.execute(
        """UPDATE artifacts
           SET body = ?, gen_hash = ?, prompt_version = ?, human_edited = 0
           WHERE item_id = ?""",
        (text, content_hash(text), prompt_ver, item_id),
    )

# backfill_runner.py — 予算上限つきの再開可能ランナー
# 何を解決するか: 途中で止まっても状態が壊れない・使いすぎない実行制御
import json, os, time
from google import genai
 
DAILY_TOKEN_BUDGET = 600_000     # 日次予算（入力+出力トークン）
STATE_PATH = "backfill_state.json"
 
def load_state() -> dict:
    if os.path.exists(STATE_PATH):
        with open(STATE_PATH) as f:
            return json.load(f)
    return {"date": "", "spent_tokens": 0, "done_ids": []}
 
def run_batch(candidates: list[str], build_request, apply_result) -> None:
    state = load_state()
    today = time.strftime("%Y-%m-%d")
    if state["date"] != today:
        state = {"date": today, "spent_tokens": 0, "done_ids": state["done_ids"]}
 
    client = genai.Client()
    for item_id in candidates:
        if item_id in state["done_ids"]:
            continue  # カーソル: 処理済みは飛ばす
        if state["spent_tokens"] >= DAILY_TOKEN_BUDGET:
            break     # 予算到達: 今日はここまで
        req = build_request(item_id)
        resp = client.models.generate_content(
            model="gemini-flash-latest",
            contents=req["contents"],
            config=req["config"],
        )
        usage = resp.usage_metadata
        state["spent_tokens"] += (usage.prompt_token_count or 0) + (
            usage.candidates_token_count or 0
        )
        apply_result(item_id, resp.text)
        state["done_ids"].append(item_id)
        with open(STATE_PATH, "w") as f:
            json.dump(state, f)  # 1件ごとに永続化（途中クラッシュに耐える）

# replace_gate.py — 置換前の新旧比較ゲート
# 何を解決するか: 「新しいほうが良いはず」という思い込みで品質を逆行させない
BANNED = {"最高の", "究極の", "No.1"}
 
def gate(old: dict, new: dict) -> str:
    """'replace' / 'keep' / 'hold' を返す"""
    # 1) 構造検査: タグ数・説明文長が規定内か
    if not (3 <= len(new["tags"]) <= 8):
        return "keep"
    if not (40 <= len(new["description"]) <= 160):
        return "keep"
    # 2) 禁止表現
    if any(b in new["description"] for b in BANNED):
        return "keep"
    # 3) 情報量が明確に痩せたら保留（人の目に回す）
    if len(set(new["tags"])) < len(set(old["tags"])) - 2:
        return "hold"
    return "replace"
 
def apply_with_archive(conn, item_id: str, old_body: str, new_body: str) -> None:
    # 旧値をアーカイブしてから原子的に置換（ロールバック可能に）
    conn.execute(
        "INSERT INTO artifact_history (item_id, body) VALUES (?, ?)",
        (item_id, old_body),
    )
    conn.execute(
        "UPDATE artifacts SET body = ? WHERE item_id = ?", (new_body, item_id)
    )
    conn.commit()