打ち切りは失敗の合図ではありませんでした — ブロッキング挙動を伴う関数呼び出しの設計

class ResourceQueue:
    """資源キーごとに直列化する。異なるキー同士は従来どおり並列で走る。"""
    def __init__(self):
        self._locks: dict[str, asyncio.Lock] = {}
 
    def lock(self, key: str) -> asyncio.Lock:
        return self._locks.setdefault(key, asyncio.Lock())
 
RESOURCE_OF = {
    "submit_build": "appstore:myapp",     # 同一アプリへの送信は重ねられない
    "convert_assets": "fs:wallpapers",    # 同一ディレクトリの書き換え
    "search_docs": None,                  # 通常ツール。直列化は不要
}
 
async def dispatch(queues, act, name, action_id, dur, timeout=None):
    key = RESOURCE_OF.get(name)
    if key is None:
        return await blocking_call(act, action_id, dur, timeout)
    async with queues.lock(key):
        return await blocking_call(act, action_id, dur, timeout)

import time, uuid
 
class IntentLedger:
    """呼ぶ前に意図を記録し、打ち切り後は再実行ではなく観測で決着させる。"""
    def __init__(self):
        self._rows: dict[str, str] = {}
 
    def open(self, tool: str, args: dict) -> str:
        action_id = f"{tool}:{uuid.uuid4().hex[:12]}"
        self._rows[action_id] = "started"     # 実運用では KV や SQLite など、
        return action_id                      # プロセスが落ちても残る場所へ
 
    def close(self, action_id: str, state: str) -> None:
        self._rows[action_id] = state
 
 
async def call_with_observation(ledger, act, tool, args, dur,
                                timeout=0.20, observe_budget=2.0, poll=0.005):
    action_id = ledger.open(tool, args)
    result = await blocking_call(act, action_id, dur, timeout=timeout)
 
    if result == "ok":
        ledger.close(action_id, "done")
        return "done", action_id
 
    # 打ち切り。ここで再試行しない。効果が現れたかどうかだけを見に行く
    t0 = time.perf_counter()
    while time.perf_counter() - t0 < observe_budget:
        if act.effects > 0:                   # 実運用では action_id で外部状態を照会する
            ledger.close(action_id, "done")
            return "done", action_id
        await asyncio.sleep(poll)
 
    # 観測予算を使い切っても判定できないときだけ、人間の判断へ渡す
    ledger.close(action_id, "unknown")
    return "unknown", action_id

TOOL_RESULT_TEMPLATE = {
    "done":    "完了しました（action_id={aid}）。次の手順へ進んでください。",
    "unknown": ("この操作の結果は確認できていません（action_id={aid}）。"
                "同じ操作を再実行してはいけません。"
                "利用者に状態の確認を依頼し、確認が取れるまで待ってください。"),
    "failed":  "失敗が確認されました（action_id={aid}）。再実行して構いません。",
}
 
def to_tool_response(state: str, action_id: str) -> dict:
    return {"status": state, "message": TOOL_RESULT_TEMPLATE[state].format(aid=action_id)}

TOOLS = [
    {
        "name": "submit_build",
        "description": "指定バージョンのビルドを送信する。完了まで戻らない。",
        "parameters": {
            "type": "object",
            "properties": {"version": {"type": "string"}},
            "required": ["version"],
        },
        # 以下は自前のメタデータ。API へは渡さず、実行側だけが読む
        "_blocking": True,
        "_resource": "appstore:myapp",
        "_timeout_ms": 200,
        "_on_timeout": "observe",     # "retry" は選ばせない
    },
    {
        "name": "search_docs",
        "description": "ドキュメントを検索する。",
        "parameters": {"type": "object", "properties": {"q": {"type": "string"}}, "required": ["q"]},
        "_blocking": False,
        "_resource": None,
        "_timeout_ms": 5000,
        "_on_timeout": "retry",
    },
]
 
def to_api_tools(tools):
    """アンダースコア始まりの自前キーを落として API へ渡す形にする。"""
    return [{k: v for k, v in t.items() if not k.startswith("_")} for t in tools]