元データを消しても生成物は残る — Gemini 派生データに削除を伝播させる台帳設計

# Before: 生成物をそのまま保存する（由来の記録なし）
def cache_summary(cache, prompt_hash: str, summary_text: str):
    cache.set(f"summary:{prompt_hash}", summary_text)
 
def index_review(vector_db, review_id: str, text: str, embedding: list):
    vector_db.upsert(
        id=f"rev-{review_id}",
        vector=embedding,
        metadata={"excerpt": text[:200]},
    )

# After: 書き込みと同時に「source_id → 派生キー」を台帳に記録する
import sqlite3, time
 
def _ledger():
    con = sqlite3.connect("provenance.db")
    con.execute("""CREATE TABLE IF NOT EXISTS derived (
        source_id TEXT NOT NULL,
        sink      TEXT NOT NULL,
        sink_key  TEXT NOT NULL,
        created_at REAL NOT NULL,
        PRIMARY KEY (source_id, sink, sink_key)
    )""")
    return con
 
def record(source_id: str, sink: str, sink_key: str):
    con = _ledger()
    con.execute(
        "INSERT OR IGNORE INTO derived VALUES (?, ?, ?, ?)",
        (source_id, sink, sink_key, time.time()),
    )
    con.commit()
    con.close()
 
def cache_summary(cache, prompt_hash: str, summary_text: str, source_ids: list):
    key = f"summary:{prompt_hash}"
    cache.set(key, summary_text)
    for sid in source_ids:
        record(sid, "response_cache", key)
 
def index_review(vector_db, review_id: str, text: str, embedding: list):
    key = f"rev-{review_id}"
    vector_db.upsert(id=key, vector=embedding,
                     metadata={"excerpt": text[:200]})
    record(review_id, "vector_index", key)

def request_deletion(source_id: str, reason: str = "user_delete"):
    con = _ledger()
    con.execute("""CREATE TABLE IF NOT EXISTS tombstone (
        source_id TEXT NOT NULL,
        sink      TEXT NOT NULL,
        sink_key  TEXT NOT NULL,
        reason    TEXT,
        status    TEXT DEFAULT 'pending',
        done_at   REAL
    )""")
    con.execute("""
        INSERT INTO tombstone (source_id, sink, sink_key, reason)
        SELECT source_id, sink, sink_key, ? FROM derived
        WHERE source_id = ?
    """, (reason, source_id))
    con.commit()
    con.close()
 
HANDLERS = {
    "response_cache": lambda key: cache.delete(key),
    "vector_index":   lambda key: vector_db.delete(id=key),
    "file_search":    lambda key: fs_client.delete_document(key),
    "request_log":    lambda key: log_store.redact(key),   # 削除でなく黒塗り
}
 
def propagate(batch: int = 200):
    con = _ledger()
    rows = con.execute("""
        SELECT rowid, sink, sink_key FROM tombstone
        WHERE status = 'pending' LIMIT ?
    """, (batch,)).fetchall()
    for rowid, sink, key in rows:
        try:
            HANDLERS[sink](key)
            con.execute("UPDATE tombstone SET status='done', done_at=? WHERE rowid=?",
                        (time.time(), rowid))
        except KeyError:
            con.execute("UPDATE tombstone SET status='no_handler' WHERE rowid=?", (rowid,))
        except Exception:
            con.execute("UPDATE tombstone SET status='retry' WHERE rowid=?", (rowid,))
    con.commit()
    con.close()

def verify_sweep(deleted_ids: list) -> dict:
    residuals = {"vector_index": 0, "response_cache": 0, "file_search": 0}
    for sid in deleted_ids:
        if vector_db.fetch(id=f"rev-{sid}") is not None:
            residuals["vector_index"] += 1
        for key in cache.scan(f"summary:*"):
            meta = ledger_sources(key)  # 台帳から source_ids を逆引き
            if sid in meta:
                residuals["response_cache"] += 1
        hits = fs_client.search(store="reviews", query=None,
                                filter={"source_id": sid})
        residuals["file_search"] += len(hits)
    return residuals