Managed Agents API でバックグラウンドタスクを安全に走らせる設計

import time, hashlib
 
def idempotency_key(task: dict) -> str:
    # タスク内容から安定した一意キーを作る
    raw = f"{task['type']}:{task['target']}:{task['date']}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]
 
def run_managed_agent(task: dict, *, max_seconds=600, budget_units=50):
    key = idempotency_key(task)
 
    # 冪等チェック: 処理済みなら何もしない
    if store.already_done(key):
        return {"status": "skipped", "key": key}
 
    started = time.monotonic()
    handle = agents.launch(           # 隔離環境にエージェントを起動
        prompt=task["prompt"],
        budget_units=budget_units,    # 予算上限を起動時に渡す
    )
 
    while True:
        state = agents.poll(handle.id)
        if state.finished:
            store.mark_done(key)
            return {"status": "ok", "result": state.result, "key": key}
 
        # タイムアウト: 壁時計で外から打ち切る
        if time.monotonic() - started > max_seconds:
            agents.cancel(handle.id)
            store.mark_failed(key, reason="timeout")
            return {"status": "timeout", "key": key}
 
        time.sleep(5)

def log_run(record: dict):
    # 構造化ログを1行で残す（後から集計しやすい形にする）
    line = (
        f"ts={record['ts']} key={record['key']} "
        f"status={record['status']} elapsed={record['elapsed']}s "
        f"units={record['units']}"
    )
    append_log(line)