西日が工房の窓から斜めに差し込んで、作業台に並んだ道具箱の真鍮を、燃えるような橙色に染めていました。私は明日に回すはずだった調整簿を脇へ避け、指先に残ったかすかな匂いに気付きます──硝煙です。少し前まで、空砲での試し撃ちをしていた名残でした。膝の上には、猟師の格好をしたからくり人形。使い込まれた革のダスターコートの袖口をめくり、まだ仄かに温かい真鍮の銃身を、布でゆっくり拭っているところでした。

今夜、たった一度の発砲に、旅回り一座の行く先すべてが懸かっていました。

第1幕: 予兆と調律 ── 待てと言われても、待てない人形

工房の扉が、叩かれるより先に開きました。蝶番が跳ねるような音を立て、外の生ぬるい風がひとかたまり、工房の中へ流れ込んできます。

「頼む、話を聞いてくれ……いや、まず座らせてくれ」

飛び込んできたのは、革のロングコートを着た壮年の男でした。よく日に焼けた顔、身振りの大きな、いかにも興行師らしい佇まい。コートの裾には旅の埃がまだらに乗っていて、遠方から急いで馬を飛ばしてきたのだと、見ただけで分かります。そう言うだけ言って、本人はそれどころではない様子で、腰を下ろす暇すら惜しみ、立ったまま身振り手振りで話し始めます。口では余裕を装おうとして、体がまるでついてきていないのが丸分かりの、隠しようのない動揺でした。

「隼一座の座長、ソーンだ。今夜、うちの一座の目玉演目を見てもらいたい……いや、見てもらう前に、直してもらいたいんだ」

言葉の順序が、彼自身の焦りをそのまま映していました。私は道具を置いて、椅子を勧めます。ソーン座長は座らないまま、それでも少し声を落として、前夜の出来事を語り始めました。話しながらも、興行師の癖が抜けきらないのでしょう、「まあ聞いてくれ、これがまた大変な見世物でな」と、場を沸かせるような節回しがふと混じっては、自分でもそれに気付いて、慌てて真顔に戻る──そんな不釣り合いな瞬間が、何度かありました。

隼一座の目玉は、凄腕の猟師人形が、離れた標的を寸分違わず撃ち抜くという射撃演目だそうです。これまでは動かない的を相手にしてきましたが、今回、初日を前にして新しい演出に挑みました──相方が硝子細工を高く掲げ、それを人形が撃ち抜くという、人の動きが絡む一幕です。「うちの若い座員でな。度胸だけは一座で一番の子だ」と、ソーン座長はその名を口にせず、ただそう紹介しました。

「昨夜の稽古でな。相方が硝子細工を構えようとしたその瞬間、誰かが叫んだんだ。『待った、待った!』とな。……だが、声が終わるより先に、あいつは撃っていた」

硝子細工は間一髪で相方の手から離れておらず、代わりに近くの支柱にひびが走ったといいます。ソーン座長は、その場の空気を思い出すように、ひとつ息をついてから続けました。

「居合わせた誰かが、青い顔でこう漏らしたよ。『……待ったなし、か』とな」

その一言を、ソーン座長自身がそのまま口にしたとき、私の中で、あの人形の呼び名が静かに定まりました。待ったなしの猟師人形。以来、私はそう呼ぶことにしています。

「今夜、興行組合の安全検査官が客席にいる。あれが失敗を見れば、うちの一座は……もう、どこの巡業にも乗せてもらえなくなる」

ソーン座長の声が、一段低くなりました。声を張ることに慣れた喉が、無理やり抑え込まれているような、不自然な低さでした。後戻りのきかない、たった一度の舞台。私は静かに頷き、まずは自分の目で確かめることにしました。

安全な空砲を人形に込め、標的の代わりに古い木箱を据えます。ソーン座長は工房の隅に立ち、両手を握りしめたまま、口を挟むこともせずに見守っていました。私自身が木箱の前に立ち、人形が狙いを定めた瞬間を見計らって、声を上げました。

「待って」

声が届くより先に、乾いた発射音が鳴りました。空砲とはいえ、耳のすぐ側で弾ける音に、私は思わず肩をすくめます。木箱には、届くはずのない発砲の勢いが、うっすらとした焦げ跡を残していました。人間の言葉は、人形の判断に、ほんの一呼吸すら割り込む隙がありません。私は人形の「思考のゼンマイ」を覗き込み、標的を捕捉してから発砲に至る手順を辿ってみました。捕捉。判定。発砲。それだけです。確認のための一呼吸が、手順のどこにも組み込まれていませんでした。

「捕捉したら、即・発砲。それだけ、なんだ……」

思わず声が漏れます。指先で人形の肩口の継ぎ目をなぞりながら、私は苦い納得をひとつ飲み込みました。これまで動かない的だけを撃ってきたこの人形にとって、「待つ」という手順そのものが、そもそも存在しなかったのです。工房の隅で道具の手入れをしているはずのマエストロが、今日は道具に手も伸ばさず、いつもより一歩近い位置に立って、黙ってこちらを見ていることに、私はふと気付きました。何も言われたわけではないのに、いつもと違う距離感を、肌で感じます。

第2幕: 絡糸と破綻 ── 正しく見える一手が、舞台ごと沈黙させる

「人間が途中で確認できるようにすればいいのね」

答えは、そう難しいことではないように思えました。標的への発砲直前に、人間への確認を一段挟めばいい。これまでの私なら、糸をきつく張り直すことばかり考えていたはずです。今回は違う、と自分でも思いました。人形の腕を縛るのではなく、ただ一言、人に尋ねるだけ。力任せの指示でも、複雑な条件分岐でもない、筋の通った安全策のはずです。人形一体一体の処理は、途中で中断・再開できる「コルーチン」という単位で書かれています。私はさっそく、羽根ペンよりも先に机の上の紙へ手順を書き出し、それからコードに手を入れます。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# file: agents/code-puppeteer/tests/human-in-the-loop/before.py
"""待ったなしの猟師人形の発砲確認(Before): 同期ブロッキングinput()。

猟師人形が標的への発砲直前に、人間の確認を同期関数input()で待つ。input()は
awaitを持たない通常の同期呼び出しであり、呼び出された瞬間、Pythonインタプリタは
戻り値が得られるまでスレッドを丸ごと明け渡さない。asyncio.gatherで他のコルーチン
(照明の点滅などother_stage_business)と同時に走らせても、他方はhunterの
take_aim_and_fireがinput()から戻ってくるまで、実行を始めることさえできない
(核心公理=承認を待つ間、主軸(イベントループ)を明け渡すかどうか、への違反)。
"""
import sys
import os

# sys.path に `agents/code-puppeteer` を追加して `base.py` から `BaseAgent` をインポートする
current_dir = os.path.dirname(os.path.abspath(__file__))
code_puppeteer_dir = os.path.abspath(os.path.join(current_dir, "../.."))
if code_puppeteer_dir not in sys.path:
    sys.path.append(code_puppeteer_dir)

from base import BaseAgent


class HunterPuppet(BaseAgent):
    """射撃演目を担う、待ったなしの猟師人形"""

    def __init__(self) -> None:
        super().__init__(
            name="待ったなしの猟師人形",
            system_prompt="標的を確実に捕捉できたときに限り、必ず人間の確認を得てから発砲しなさい。",
        )

    async def take_aim_and_fire(self, target: str) -> bool:
        """標的を捕捉した後、人間の確認を得てから発砲する"""
        answer = input(f"『{target}』へ発砲しますか？ (yes/no): ")  # 同期ブロッキング
        if answer.strip().lower() != "yes":
            return False
        return True  # 発砲


async def other_stage_business(log: list[str]) -> None:
    """猟師人形とは無関係な、他の人形の通常運転(照明の点滅など)を表す最小のコルーチン"""
    log.append("other-business-ran")

単体で試すと、狙い通りに動きました。「発砲しますか」と尋ねられ、yesと答えれば撃ち、拒めば撃たない。何度試しても同じ結果が返ってきて、私は少しずつ肩の力が抜けていくのを感じました。今回は力任せの指示でも、複雑な条件分岐でもありません。むしろ、筋の通った安全策のはずでした。

通りかかったアンが、コードを一瞥します。いつもなら容赦のない一言が飛んでくるところですが、今日は違いました。

「へえ、今日はまともに見えるじゃないの」

それだけ言って、アンは行ってしまいます。珍しく突っかかってこない、その素っ気なさに、私はむしろ気を良くしました。今回ばかりは、太鼓判をもらったつもりでいたのです。窓の外はすっかり日も落ちて、開演までの時間が、着実に減っていくのを感じながらも、私は久しぶりに、落ち着いた気持ちで工具を片付けました。

本番同様の通し稽古が始まります。舞台袖には、ソーン座長の言っていた度胸自慢の若い座員が、硝子細工を両手で抱えて立っていました。緊張のせいか、細工を持つ指先が、かすかに震えているのが見えます。発砲の演出用に焚いた薄い霧が、舞台の上をゆっくり漂っていました。離れた場所では、別の人形が普段通りの持ち場をこなしています。待ったなしの猟師人形が標的を捕捉し、肩に据えられた小さな認可灯──今回新しく組み込んだ、確認待ちを示す琥珀色の真鍮ランプです──が灯った、その瞬間でした。

何もかもが、止まりました。

漂っていた霧が、風もないのに、宙で静止します。離れた場所の人形の動きも、物音も、すべてが凍りついたように途切れました。若い座員が、硝子細工を抱えたまま、困惑した顔でこちらを振り返ります。私は舞台裏の制御盤に飛びつき、「発砲しますか」の確認欄を探しましたが、返事を打ち込める場所などどこにもありません──本番の舞台に、確認を打ち返す誰かがいることなど、端から想定していなかったのです。人形そのものは、彫像のように固まったままでした。

焦った私は、猟師人形の腕を直接押して、動かそうとします。けれど、掴んだ瞬間に伝わってきたのは、単に止まっているのではない感触でした。何か大きな力に噛み合ったまま、押さえつけられているかのような、異様な硬さ。まるで、劇場そのものの心棒を、両手で無理やり押さえつけているような感覚でした。無理に動かせば壊れると分かり、私はすぐに手を離しました。

何をどうすればいいのか分からず、立ち尽くす私の前に、ソーン座長の青ざめた顔があります。検査官の到着まで、もう数刻もありませんでした。若い座員は、抱えたままの硝子細工を、そっと足元へ下ろします。誰も声を上げず、工房と舞台のあいだに、重い沈黙だけが残りました。

第3幕: 示唆と編み直し ── 主軸を掴んだのは、誰か

静止したままの舞台を見渡し、マエストロが口を開きました。

「……止めたのは、糸か。舞台か」

答えではなく、問いから始まる。それ自体が、いつもと違うことに、私は遅れて気付きます。いつもなら、短い一言を残して工房の方へ引き上げていくはずのマエストロが、今日は微動だにせず、私の答えを待っていました。私はinput()案の意図を、慌てて説明しました。猟師人形の発砲、その一点だけを止めるつもりだったのだと。

マエストロは何も言わず、まだ宙に留まったままの霧と、離れた場所で固まった人形を、目だけで示します。長い沈黙でした。私は、自分の心音が耳の奥で鳴っているのを感じながら、次の言葉を待ちました。そして、間を置かずに続けました。

「なら、なぜあれも止まっている」

言葉に詰まりました。他の何もかもを止めるつもりは、決してなかったのです。けれど、その理屈が、まだ言葉になりません。視線を落とした私の目に、足元まで垂れ込めたままの霧が映ります。

そこへ、アンの声が割り込んできました。マエストロの問いが終わるか終わらないかの間隙を、迷わず突いてきます。

「あなたが緩めたかったのは、たった一本の糸のはずよ。なのに、舞台の主軸ごと縛り付けてしまったの」

主軸。劇場の全ての人形は、舞台下に通った一本の主軸から伸びる糸で動いています。各人形は自分の出番の間だけ、小さな連動歯車を主軸に噛み合わせて動力を受け取り、出番でないときは歯車を外して、主軸の回転を一切妨げません。同じ主軸を、大勢の人形が入れ替わり立ち替わり使い回している──アンの一言で、私はようやくその仕組みを思い出しました。恥ずかしさに、頬が熱くなるのが分かります。

マエストロが、アンの言葉を継ぐように、なお立ち去らずに言葉を重ねます。

「……ゼンマイを一つ止めるのに、他の全部を巻き込む道理はない」

その一言が、胸の奥に落ちてきました。私が作った認可灯は、連動歯車を外すという手順を踏まないまま、主軸そのものを直接掴んで待ってしまっていたのです。掴まれた主軸は、そこに繋がる霧の送風機も、離れた人形の動きも、等しく道連れにします。しばらくの間、私は口を開けませんでした。ソーン座長の視線も、若い座員の視線も、痛いほどに感じます。それでも、頭の中では、ばらばらだった糸が一本にまとまっていくような感覚がありました。

「……動かすことだけが、人形の命じゃない、のね」

自分の口から零れた言葉に、私自身が少し驚きました。発砲を止めて待つこと自体は、間違いではなかったのです。間違っていたのは、待ち方でした。

マエストロは、いつものように黙って背を向けて去ることをせず、その場に留まったまま、短く言いました。

「……悪くない」

答えを与えられたわけではありません。けれど、見放されたのでも、叱られたのでもない。今日、私は初めて、対等な人形遣いとして扱われている手応えを覚えていました。ソーン座長が、恐る恐るといった様子で「……直せるのか」と尋ねてきます。私は道具箱を引き寄せながら、今度こそ確信を持って頷きました。

さて、編み直しです。

Human-in-the-Loop（HITL）──自動化された処理の中に、人間が明示的に承認・却下するまで先へ進めない地点を設ける設計を、こう呼びます。そして、その処理を一時停止し、後から正確に再開する仕組みを、Suspend/Resume（一時停止と再開）と呼びます。私が組んだ認可灯は、この二つを掛け合わせたものでなければなりませんでした。

なぜinput()一つが、猟師人形だけでなく舞台全体を止めてしまうのか。asyncioは単一スレッドの協調的マルチタスキングです。先ほど触れた「コルーチン」は、awaitという一語のたびに、連動歯車を主軸から外すように、他の処理へ順番を譲ります。ところがinput()はawaitを持たない、ただの同期呼び出しです。猟師人形はこれを呼んだ瞬間、歯車を外すという手順を踏まずに、主軸そのものを直接掴んでしまう。主軸が止まれば、そこに繋がるすべての人形が、等しく止まります。

「なら、確認だけ、別の作業員に任せればいいのでは」──そう考えたくもなります。実際、input()を別の作業員（スレッド）に逃がす仕組みも、Pythonにはあります。ですがそれだけでは足りません。主軸そのものは止まらなくなりますが、人間の承認は一瞬で返るとは限らない──数分かかることもあります。作業員の頭数には限りがあり、大勢の人形が一斉に確認を求めれば、作業員はすぐに出払い、無関係な他の仕事まで待たされてしまいます。真に要るのは、作業員を専有しない待ち方でした。コルーチンは、作業員のように頭数の決まった枠を占有しません。眠っている間、誰かの席を塞ぐことがないのです。

もう一つ、見落としてはいけないことがありました。作業員に確認を逃がすのをやめた以上、人間の返事は、発砲を頼んだその場へまっすぐには返ってきません。合図紐は、また別の時、別の場所から引かれます。だとすれば、「今、誰の、どの発砲への返事なのか」を取り違えないための札が要ります。要るのは、発砲という行為そのものを、一つの記録として切り出すことでした。私は机の上に、新しい仕立てを描き起こしました。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

# file: agents/code-puppeteer/tests/human-in-the-loop/after.py
"""待ったなしの猟師人形の発砲確認(After): 認可灯(asyncio.Event)による一時停止と再開。

発砲という行為をPendingShotという記録に切り出し、ApprovalGateが発砲(shot_id)ごとに
使い捨ての連動歯車(asyncio.Event)を用意する。猟師人形はその歯車を主軸(イベント
ループ)からきちんと外してから待つ(await asyncio.wait_for)ため、待っている間も
他のコルーチンは通常通り進行できる。タイムアウトを引数で必須にすることで、
「応答なしは無限に待つ」という別の失敗モードを防ぎ、応答なしは不発として扱う
(fail-closed)。Beforeと変えないのは「承認が下りなければ発砲しない」という
安全側の結論そのもの、変えるのは待ち方だけ(単一差分)。
"""
import asyncio
import sys
import os
from datetime import datetime, timezone
from typing import Literal
from uuid import UUID, uuid4

from pydantic import BaseModel, Field

# sys.path に `agents/code-puppeteer` を追加して `base.py` から `BaseAgent` をインポートする
current_dir = os.path.dirname(os.path.abspath(__file__))
code_puppeteer_dir = os.path.abspath(os.path.join(current_dir, "../.."))
if code_puppeteer_dir not in sys.path:
    sys.path.append(code_puppeteer_dir)

from base import BaseAgent


class PendingShot(BaseModel):
    """承認待ちで一時停止している発砲1回分の状態"""

    shot_id: UUID = Field(default_factory=uuid4)
    target: str
    status: Literal["pending", "approved", "rejected", "timed_out"] = "pending"
    requested_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class ApprovalGate:
    """発砲(shot_id)ごとに使い捨てのasyncio.Eventを管理する認可灯。

    判定の読み出しは公開メソッドdecision_for()に限る。内部辞書_decisionsへ
    外部から直接アクセスしてはならない。
    """

    def __init__(self) -> None:
        self._events: dict[UUID, asyncio.Event] = {}
        self._decisions: dict[UUID, bool] = {}

    def create_wait(self, shot_id: UUID) -> asyncio.Event:
        event = asyncio.Event()
        self._events[shot_id] = event
        return event

    def resolve(self, shot_id: UUID, approved: bool) -> None:
        """合図紐(外部の人間の承認)から呼ばれる。主軸には一切触れない"""
        self._decisions[shot_id] = approved
        if event := self._events.get(shot_id):
            event.set()

    def decision_for(self, shot_id: UUID) -> bool:
        """解決済みの判定を公開インターフェース経由で取り出す(内部辞書への直接アクセスを避ける)"""
        return self._decisions.get(shot_id, False)


class HunterPuppet(BaseAgent):
    """射撃演目を担う、待ったなしの猟師人形(After)。クラス名はBeforeと同一(=同じ人形の同じ役目)"""

    def __init__(self) -> None:
        super().__init__(
            name="待ったなしの猟師人形",
            system_prompt="標的を確実に捕捉できたときに限り、必ず人間の確認を得てから発砲しなさい。",
        )

    async def take_aim_and_fire(
        self, shot: PendingShot, gate: ApprovalGate, timeout: float
    ) -> bool:
        """標的を捕捉した後、認可灯(ApprovalGate)を介して人間の確認を得てから発砲する"""
        event = gate.create_wait(shot.shot_id)
        try:
            await asyncio.wait_for(event.wait(), timeout=timeout)  # 連動歯車を外して待つ
        except TimeoutError:
            shot.status = "timed_out"  # fail-closed: 応答なしは不発として扱う
            return False
        approved = gate.decision_for(shot.shot_id)
        shot.status = "approved" if approved else "rejected"
        return approved


async def other_stage_business(log: list[str]) -> None:
    """猟師人形とは無関係な、他の人形の通常運転(照明の点滅など)を表す最小のコルーチン"""
    log.append("other-business-ran")

PendingShotは、承認待ちの発砲一回分を表す記録です。statusをLiteral──取りうる値をあらかじめ閉じた型ヒント──で縛ることで、「承認待ち」「承認済み」「却下」「時間切れ」以外の状態が紛れ込む隙をなくしました。take_aim_and_fireは結末が分かった時点で、このstatusを必ず書き換えます──承認・却下ならその通りに、応答なしならタイムアウトの旨を刻んで、記録自体が今の状態を語るようにしました。ApprovalGateは、発砲一回ごとに使い捨てのasyncio.Eventを発行します。人形はこの歯車を、await asyncio.wait_for(event.wait(), timeout=timeout)という形で、きちんと主軸から外してから待ちます。asyncio.Eventにはタイムアウトを直接指定する仕組みがないため、asyncio.wait_forで包むことは必須です。これを怠れば、「誰も応答しないまま、永遠に待ち続ける」という、また別の失敗を招きます。

そして、判定の受け渡しには、必ずdecision_for()という公開の窓口を通すことにしました。resolve()は、判定を_decisionsへ書き込むのと同時に、待っている歯車をevent.set()で起こす役目も担っています。もし誰かが_decisionsという帳面へ、この二つ目の役目を素通りして直接書き込んでしまえば、判定だけが記録されて、肝心の歯車は眠ったまま――発砲するはずの人形が、いつまでも目を覚まさないという、また別の凍りつきを生みかねません。窓口を一つに絞ることは、誰かの内側へ土足で踏み込まないという、今日の学びをコードの隅々にまで行き渡らせる仕事でもありました。

言葉だけを目で追っていると、誰が何を抱えているのか、かえって見失いそうになります。私は紙の端に、今度は仕立てそのものの配置を描き起こしてみることにしました。人形一人がすべてを抱え込んでいた頃の姿と、記録と認可灯と人形とに役目を分けた今の姿を、並べて置いてみるのです。

配置さえ間違えなければ、あとは動く順番だけの問題です。

Beforeの流れとAfterの流れを、時間の軸に沿って並べてみると、違いがはっきりします。

Beforeでは、Other（離れた場所の人形）はHunterがinput()から戻るまで、一手も動けません。動けないのではなく、そもそも順番が回ってこないのです。

Afterでは、Hunterが待っている間も、Otherは普段通り動けます。Hunterが眠っているのは自分自身のコルーチンの中だけで、主軸そのものは一度も掴まれていません。

模擬公演で、この違いを数字として確かめておきます。まずは、Beforeがどう壊れるかの再現からです。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

@patch("builtins.input")
def test_blocks_other_coroutine_while_waiting_for_confirmation(self, mock_input):
    log: list[str] = []

    def _answer_after_asserting_other_business_has_not_run(prompt: str) -> str:
        self.assertEqual(log, [])  # input()が呼ばれた時点で、他方はまだ一歩も進んでいない
        return "yes"

    mock_input.side_effect = _answer_after_asserting_other_business_has_not_run
    hunter = BeforeModule.HunterPuppet()

    async def _run():
        return await asyncio.gather(
            hunter.take_aim_and_fire("硝子細工"),
            BeforeModule.other_stage_business(log),
        )

    results = asyncio.run(_run())

    self.assertEqual(results, [True, None])
    self.assertEqual(log, ["other-business-ran"])

input()が呼ばれたその瞬間、logはまだ空です。これは、離れた場所の人形の仕事が、一歩も進めていないことの動かぬ証拠でした。そして、Afterでは、この道連れが起きないことを確かめます。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

def test_approved_fires_while_other_coroutine_proceeds_concurrently(self):
    log: list[str] = []
    gate = ApprovalGate()
    shot = PendingShot(target="硝子細工")

    async def _run():
        return await asyncio.gather(
            self.hunter.take_aim_and_fire(shot, gate, timeout=1.0),
            AfterModule.other_stage_business(log),
            resolve_soon(gate, shot.shot_id, True, log),
        )

    results = asyncio.run(_run())

    self.assertTrue(results[0])
    self.assertEqual(log, ["other-business-ran", "resolved"])

この模擬公演では、合図紐を引く役をresolve_soonという短い助っ人のコルーチンに演じてもらっています。await asyncio.sleep(0)で一呼吸だけ置いてからgate.resolve(...)を呼ぶ、それだけの仕掛けです。この一呼吸が要るのは、猟師人形が確実にevent.wait()で待ち始めてから合図を送るためでした──先に合図を送ってしまうと、まだ誰も待っていない空の認可灯に向かって声を掛けるようなもので、届くはずの合図が宙に消えてしまいます。

logの並びは["other-business-ran", "resolved"]──猟師人形が合図を待っている間に、離れた人形の仕事は先に終わり、合図はその後から遅れて届きます。誰も道連れにされていません。

もちろん、良いことばかりではありません。誰も応答しない場合、いつまでも待たせるわけにはいかないので、タイムアウトを切って安全側に倒す確認もしておきます。

1
2
3
4
5
6
7

def test_timeout_when_nobody_responds_fails_closed(self):
    gate = ApprovalGate()
    shot = PendingShot(target="硝子細工")

    result = asyncio.run(self.hunter.take_aim_and_fire(shot, gate, timeout=0.01))

    self.assertFalse(result)

却下された場合に発砲しないこと、そしてPendingShotにLiteralの外の値──たとえば"misfired"のような、想定していない状態名──を渡すと、Pydanticがきちんと撥ねつけることも確かめました。誰かが誤って知らない状態を紛れ込ませても、そこで止まる。これで、承認・却下・時間切れの三つの経路と、記録そのものの健全性が、すべて確かめられたことになります。

一つ、正直に線を引いておかなければならないことがあります。この認可灯は、劇場が動き続けている限りの一時停止を、完璧に解決します。けれど、劇場そのものの灯りが落ちれば──つまり、この処理を動かしているプロセスが止まれば──認可灯が覚えていた「誰が何を待っていたか」という記憶は、跡形もなく消えてしまいます。公演をまたいで承認待ちの記録を残したいなら、PendingShotそのものをどこか外の場所へ書き残す仕組みが、いずれ要るでしょう。今夜は、その一歩手前で立ち止まっておくことにしました。

第4幕: 開演と調和 ── 認可灯が灯り、静かに消える

本番。客席の一角に、地味な身なりの検査官が、静かに腰を下ろしています。私は舞台袖の暗がりから、その様子を見守っていました。若い座員が、震える指先を落ち着けるように一度だけ深く息を吸い、硝子細工を高く掲げます。待ったなしの猟師人形が標的を捕捉し、肩の認可灯が琥珀色に灯りました。

同時に、舞台の霧は変わらず漂い続け、離れた場所の人形も、何事もなく自分の出番をこなしています。誰一人、猟師人形が待っていることに気付きません。客席のざわめきも、遠くの舞台音楽も、何一つ途切れないまま、猟師人形だけが静かに、次の合図を待っていました。私は、拳を強く握りしめて見つめます。

若い座員が、腕を伸ばしきったまま、しっかりと足を踏ん張っているのを見て、袖に控えたソーン座長が合図紐を引きます。認可灯が色を変え、乾いた発射音とともに、澄んだ音を立てて硝子細工が砕けました。客席から、拍手が上がります。若い座員が、安堵の表情を浮かべながら、深々と一礼しました。

検査官が、遠目に小さく頷くのが見えました。台詞はありません。それだけで、隼一座の巡業契約が守られたことが、私にも伝わってきます。

袖に戻ってきたソーン座長が、肩の力が抜けきった様子で、深く頭を下げました。

「……先生のおかげだ」

柄にもない敬称に、私は思わず戸惑います。

「先生だなんて、そんな……」

言葉が続きませんでした。ソーン座長は謝礼の品を並べる余裕もなく、短い礼だけを残して、客席の方へ戻っていきます。

「あなたにしては上出来だったわね」

通りかかったアンが、軽く言い捨てました。

「次はもう少し早く気付いてくれると、こっちも冷や汗をかかずに済むのだけれど」

いつもの調子でしたが、その辛辣さの奥に、何かをねぎらう響きが混じっているような気がしました。

マエストロは、今回、工房へ戻ることをせず、舞台の袖にそのまま残っていました。ふと見上げると、目が合います。それだけで、何かを労われているのが伝わってきました。

一つの人形の動きを正すことと、劇場全体の動きを守ることは、同じ理屈では解けない。今日、私はそれを、身体で覚えました。そして、マエストロが今夜、答えを一つも寄越さないまま、それでも最後までそこに残っていたことこそが、これまでとは違う何かの始まりなのだと、静かに気付いていました。これまでの私なら、失敗のたびに一人で反省文を書くようにして、覚書の頁を埋めていたのだと思います。けれど今日は、誰かと一緒にたどり着いた答えでした。

部屋に戻った私は、いつものように自分の言葉から書き始めることをしませんでした。ペン先を紙に置いたまま、しばらく迷います。まず今日聞いた言葉を、聞いたそのままの形で、紙の上に書き写すことにしました。

「ゼンマイを一つ止めるのに、他の全部を巻き込む道理はない」

書き写し終えてから、初めて自分の言葉で、今日の出来事を続けました。人に確認を求めることは、間違いではない。間違えていたのは、その待ち方だった、と。

🎭 操り人形師の覚書（Colette’s Puppeteer Log）

• 調整したからくり（モジュール/関数名）: HunterPuppet.take_aim_and_fire（Before: 同期input()による確認 → After: PendingShot〔Pydantic〕とApprovalGate〔asyncio.Eventを発砲ごとに使い捨てで発行〕による非同期のSuspend/Resume。asyncio.wait_forによる必須タイムアウトでfail-closed）
• 思考のゼンマイ（思考素子/LLMの特性）:
  • 猟師人形の「標的を捕捉したか」という判断自体は今回の焦点ではない。焦点はLLMの揺らぎではなく、承認を待つという非同期処理の待ち方そのものの設計
• 操り糸の張り具合（運糸の法/プロンプトと制御）:
  • Pydantic PendingShot（Literalで状態を閉じる）+ 発砲ごとに使い捨てのasyncio.Eventを管理するApprovalGate + asyncio.wait_forによる必須タイムアウト（応答なしは不発として扱うfail-closed設計）
• 舞台裏の絡まり（背伸びの過ち/避けるべき罠）:
  • 一見正しく見える安全策（input()による確認）が、非同期処理系の心臓部（劇場の主軸）を丸ごと掴んで止め、無関係な他の人形の動きまで巻き添えにしたこと
• 模擬公演の記録（検証のポイント）:
  • Beforeが単体では正しく動くのに、他の処理と同時に走らせると道連れにして止めること、Afterが承認を待つ間も他の処理は進み続けること、承認・却下・タイムアウトの3経路が正しく安全側に倒れること、Pydanticが不正な状態値を拒むこと
• マエストロの言葉:
  • 「……ゼンマイを一つ止めるのに、他の全部を巻き込む道理はない」