Python API¶

このページでは flower のトップレベル __all__ にある 62 個の公開シンボルを網羅する:シグネチャ、引数、デフォルト値、意味、 公開属性とメソッド。読み終えたら引数を探すためにソースを開く必要はない。

構成は関心事ごとであって、モジュールファイルごとではない ——「coordinator が 自分で手を動かすのをどう止めるか」を知りたいなら hook 層へ;「前のステップの結果を次にどう渡すか」ならworkflowへ。 用語はすべて用語集に従う。

バージョン 0.1.0、依存は claude-agent-sdk>=0.2.152。すべてのシグネチャはソースと逐語対応。

from flower import Runtime, Workflow, Step, coordinator, worker   # 顶层一次导入

このページの内容¶

噛みつく 6 つのデフォルト値¶

この 6 条は些末事ではなく、最も頻出する 6 種類の事故だ。各条は対応する節に完全な説明がある。

ランタイム¶

ソース:flower/core/runtime.py

Runtime は実行の中核だ。ワークスペース、session store、workbench、 resilience 方針、handoff 方針を保持し、外向きには動詞が 1 つしかない:1 ステップを run する。 リトライ、中断後の再開、コンテキスト満杯時の handoff は、すべてこの 1 回の呼び出しの中で完結する。

Runtime¶

Runtime(
    *,
    workspace: str | Path,
    run_dir: str | Path = "runs",
    portable: bool = True,
    trim: TrimPolicy | bool = False,
    ephemeral: EphemeralPolicy | bool = True,
    keep_denials: int = 1,
    workbench: Workbench | bool = False,
    spill_threshold: int | None = 4000,
    resilience: Resilience | bool = True,
    handoff: HandoffPolicy | bool = True,
)

コンストラクタ引数はすべて keyword-only(* が先頭)、workspace は必須。

session store はハードコードされている:常に PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...))。 コンストラクタ引数にバックエンドを差し替える入口はない —— 差し替えたいなら自分で AgentSpec + build_options(session_store=...) を組むか、 構築後に rt.store を上書きする。

構築の最後の 2 ステップは load_dotenv() と check_credentials() で、後者はエラーがあれば raise RuntimeError。 資格情報がない場合は構築段階で落ちるのであって、run() を待たない。

公開属性¶

クラス定数:INTERRUPTED = "interrupted-by-human"、HANDOFF_DUE = "context-full-handoff"、 INTERRUPT_NOTE(中断後の再開時に人の発言の後ろに付ける一節で、「そのとき飛んでいたツール呼び出しが interrupted を 返すのは中断の正常な副作用であって、環境障害ではない」と説明するもの)。

公開メソッド¶

Runtime.run(...)¶

async def run(
    self,
    spec: AgentSpec,
    prompt: str,
    *,
    step_name: str | None = None,
    resume: str | None = None,
    fork: bool = False,
    resume_at: str | None = None,
    on_event: Callable[[Event], None] | None = None,
) -> StepResult

各ステップの冒頭でコンテキスト水位をゼロに戻す(self._ctx, self._warned = 0, False)。その後はループで、出口は 4 つ:

1. 成功 → 抜ける。
2. 人による中断(result.error == INTERRUPTED)→ max_attempts の制約を受けず、ネットワークも待たない。 人の発言を携えて同じ session を resume、attempt -= 1(中断は失敗試行に数えない)、prompt = 人の発言 + INTERRUPT_NOTE。 session_id を得ていなければ止まるしかない。
3. コンテキスト満杯(result.error == HANDOFF_DUE、または handoff.enabled かつ session_id を得ていて is_overflow(...) が成立)→ これも max_attempts の制約を受けない。まず len(result.retired) >= handoff.max_generations を確認し、 超えていれば error を診断メッセージに差し替えて抜ける;そうでなければhandoff document を書き → resume=None, fork=False (完全に新しい session)→ prompt を h.prompt_block() に差し替え → 水位をゼロに戻し → attempt -= 1。
4. リトライ可能な障害 → not resilience.enabled or attempt >= max_attempts なら抜ける; classify(error) がリトライすべきでないと判定すれば抜ける;そうでなければ Event("retry") を発火し、wait_online() でネットワークを待ち、 sleep(delay_for(attempt));session_id を得ていれば resume で継続する(prompt は resilience.resume_prompt に差し替え)、そして result.resumed を True にする。

締めくくり:ended_at を書き、self.results に追記し、manifest.json を書く。

manifest.json はappend セマンティクス:書くたびにディスクを読み直し、run フィールドで重複排除する(自分の行は差し替え、他人の行は残す)。 だから同じ run_dir の下で flower を 2 つ並行して走らせても安全だ —— run_id が衝突しないことが前提。

handoff の 3 つの観測点(いずれも Event("handoff")、payload["phase"] で区別): near(warn_at に接近、世代ごとに 1 回だけ発火)、writing(handoff を書いている最中、十数秒かかる)、 done(payload に degraded / path / sections を含む)。handoff を書くそのターンは replace(spec, max_budget_usd=None) で走らせる —— handoff は必ず書き切れなければならず、予算で詰まってはいけないからだ; さらに on_event=None で、このターンは UI に流さない。

handoff は <workbench.notes>/交接-<步骤名>.md に spill される;workbench がなければ spill されない。文書は変わらず prompt で 引き継ぎ手に渡されるが、事後に読み返せなくなるだけだ。古い handoff は notes/archive/交接/<名>-<时间戳>.md に移される。

StepResult¶

@dataclass
class StepResult:
    step: str
    session_id: str | None = None
    ok: bool = False
    cost_usd: float = 0.0
    num_turns: int = 0
    text: str = ""
    error: str | None = None
    started_at: float = 0.0
    ended_at: float = 0.0
    attempts: int = 1
    errors: list[str] = field(default_factory=list)
    resumed: bool = False
    retired: list[str] = field(default_factory=list)
    context: int = 0

1 ステップ走り終えた後の全帳簿。

ワークフロー¶

ソース:flower/workflow/

ワークフローは順番につないだステップの集まりであり、 加えてステップ間で状態をどう渡すか、いつ早期終了するかを決めるものだ。フレームワークは既製のワークフローを提供しない。ワークフローはあなたが書く —— starter_flow は動くサンプルにすぎない。

型エイリアス Ctx = dict[str, Any](flower.workflow.base.Ctx。flower.workflow.__all__ にはあるが、 トップレベルの __all__ にはない)。

Step¶

@dataclass
class Step:
    name: str
    spec: AgentSpec
    prompt: str | Callable[[Ctx], str]

    resume_from: str | None = None
    fork: bool = False

    retries: int = 0
    gate: Callable[[StepResult, Ctx], bool] | None = None
    on_fail: str = "stop"
    when: Callable[[Ctx], bool] | None = None
    on_reject: Callable[[StepResult, Ctx], str] | None = None
    resume_prompt: str | Callable[[Ctx], str] | None = None
    reduce: Callable[[StepResult, Ctx], str] | None = None

ステップの宣言だ。Step そのものは関数ではない —— 実際に実行するのは Runtime.run(step.spec, prompt, ...)。 先頭3つのフィールドは位置引数なので、Step("取词", terse, "读 seed.txt …") と書ける。

セッションのつなぎ方は3通り(同一ラン内):

on_reject はリトライの意味を変える:

• 与えない → 次の試行は最初から(同じ prompt、同じ resume_from)。
• 与える → 次の試行はさっき否決されたセッションを継続し、prompt はその戻り値に差し替わる。fork は強制的に False。
• 空文字列を返す → 差し戻さず、最初から走る動きに退化する。
• result.session_id が None → これも最初から走る動きに退化する。

on_fail の3つの値:

合格・不合格にかかわらず ctx["_results"][name] = result は必ず書かれる。result.session_id が非空なら ctx["_sessions"] にも書き、lineage.remember(...) する。

Workflow¶

@dataclass
class Workflow:
    steps: list[Step]
    name: str = "workflow"
    context: Ctx = field(default_factory=dict)
    channel: Any = None
    workbench: Any = None
    continuous: bool = True

    async def run(
        self,
        runtime: Runtime,
        *,
        on_event: Callable[[Event], None] | None = None,
        on_step: Callable[[Step, StepResult], None] | None = None,
    ) -> Ctx

Step の列を順に走らせ、最終的な ctx を返す。steps は位置引数なので Workflow([...]) と書ける。

run() が ctx に書き込むプライベートキー(すべて _ 始まりで、ステップ名と衝突しない):

Event("step") の payload:{"index": i, "total": len(steps), "resumed": bool, "woke": int}。

リトライのラベル:0回目は step.name。以降は on_reject があれば f"{name}#round{attempt+1}"、 なければ f"{name}#retry{attempt}"。マニフェストを見れば、そのステップがどう終わったか一目でわかる。 サフィックス付きの名前はプロセスをまたぐリネージには入らない —— Lineage.remember が使うのは元の名前だ。

runtime.on_session は runtime.run の一文だけを覆い、try/finally で gate の前に必ず外れることを保証する。 prompt_cur / resume_cur / fork_cur はローカル変数で、step に書き戻さない —— 同じ Step オブジェクトが2度走る可能性があるからだ。

StepAbort¶

class StepAbort(Exception): ...

gate から投げる = 即座に停止し、もうリトライしない。「False を返す」との違いはこうだ:False は「今回はダメ、もう1ラウンド」。 StepAbort は「もう1回やっても無駄」。

投げたあと:ctx["_aborted"] = str(exc)、passed = False、リトライループを抜ける(残りの retries は消費しない)。 そのあとは通常の失敗として on_fail(デフォルト "stop")に従う。

with_goal は2か所でこれを投げる:ctx["_runtime"] が取れないとき、そして判定結果が unreachable で誰も応答しないときだ。

clarify_step()¶

def clarify_step(
    channel: HumanChannel,
    *,
    brief_path: str | Path,
    prompt: str | Callable[[Ctx], str],
    name: str = "确认需求",
    spec: AgentSpec | None = None,
    instructions: str = "",
    always_ask: bool = False,
    on_fail: str = "stop",
    retries: int = 0,
    **spec_kw,
) -> Step

事前確認を行う Step を生成する:要件を聞き切る → Brief にパース → 4節すべて揃っていれば凍結してディスクに書き出す。

生成される Step の各フィールドはこう埋まる:

• resume_prompt = CLARIFY_RESUME。
• when:always_ask=True → 常に True。そうでなければ Brief.load(brief_path) が揃っていれば ctx に流し込み、 そのうえで False を返す(スキップ) —— スキップするときも流し込まないと、下流が要件を受け取れない。
• gate:Brief.parse(result.text)。欠けていれば ctx[MISSING_KEY] を書いて False を返す。 揃っていれば b.write(brief_path) で凍結し、ctx に流し込んで True を返す。
• reduce:ctx[BRIEF_KEY].prompt_block() を返す。モデルの原文ではない —— 原文にはモデルが余計に書いたものが混ざりうる。
• resume_from はデフォルトの None のまま:次のステップは新しいセッションで、ブリーフだけを受け取り、あの一問一答は受け取らない。 事前確認の問答はコーディネーターの文脈に一度も入っていない。入ってから刈り取られたのではない。

ctx に流し込む3か所:ctx[BRIEF_KEY] = b、ctx[name] = b.prompt_block()、ctx.pop(MISSING_KEY, None)。

goal_step()¶

def goal_step(
    channel: HumanChannel,
    *,
    goal_path: str | Path,
    brief_key: str = "确认需求",
    name: str = "设定目标",
    spec: AgentSpec | None = None,
    instructions: str = "",
    always_set: bool = False,
    on_fail: str = "stop",
    retries: int = 0,
    **spec_kw: Any,
) -> Step

目標を設定する Step を生成する:ジャッジにブリーフを読ませ、目標と判定チェックリストを書かせ、 Goal にパースしてから凍結してディスクに書き出す。clarify_step と同じ形をしている。

can_run の仮引数はない —— 目標設定のジャッジにコマンドを走らせたいなら、**spec_kw 経由で can_run=True を渡すしかない。 渡さなければ Bash は手に入らず、JUDGE_RULES の「まず自分がどんな環境にいるかを見極めろ」という一条が実行できない。

gate はパースと凍結のほかに、もう1つやることがある:目標に [此环境无法验证:…] の項目があるとき、その場で ctx["_on_event"] 経由で Event("task", payload={"unverifiable", "total", "path"}) を発して警告する —— これらの項目の運命は目標を設定するこの瞬間に決まっており、判定の時点ではもう1ラウンド分の作業費を使い切っているからだ。

resume_prompt は設定していない —— 目標設定はそもそもブリーフ全文を再送すべきだ。

with_goal()¶

def with_goal(
    step: Step,
    channel: HumanChannel,
    *,
    goal_path: str | Path,
    spec: AgentSpec | None = None,
    rounds: int = 3,
    instructions: str = "",
    can_run: bool = False,
    name: str | None = None,
    **spec_kw: Any,
) -> Step

既存の Step にゴールガードを被せる:各ラウンドの終わりにジャッジが独立して判定し、 達成していなければ差し戻して作業を続けさせる。

戻り値は replace(step, retries=max(0, rounds - 1), gate=<新しい gate>, on_reject=<新しい on_reject>) —— フィールドを1つずつ組み直すのではなく dataclasses.replace を使う。一度組み直したときに resume_prompt が漏れ、しかもエラーにならず、 継続時にブリーフ全文をもう一度送ってしまっただけだった。

gate は async で、流れはこうだ:

1. ctx["_runtime"] がない → StepAbort を投げる(「Runtime が取れず、目標を判定できない」)。通ったふりをするな。
2. ctx[ROUND_KEY] += 1。
3. 目標を取る:まず ctx[GOAL_KEY] の揃った Goal、なければ Goal.load(goal_path)、それもなければ空の Goal()。
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...)。 ジャッジは独立した1回の Runtime.run であり、resume は常に None —— つねに新しいセッションだ。step_name にはラウンド番号が付くので、 プロセスをまたぐリネージには入らない。
5. Verdict.parse(vr.text) を ctx[VERDICT_KEY] に書く。
6. v.achieved → True を返す。
7. unreachable ではない(v.ok=False の曖昧なケースを含む)→ 曖昧なときはデフォルトの reason を1行補い、False を返す。 曖昧なものは一律で未達成とみなす —— 「いけそうに見える」の一言で作業を打ち切らせてはならない。
8. unreachable → await channel.ask(...) で人に尋ねる。選択肢は3つ:
9. 誰も応答しない(a.state != "answered")→ StepAbort を投げる。空回りを続けるのが最も高くつく選択だ。
10. 「この結果を受け入れて、このまま先へ進む」→ True を返す。
11. 「目標を修正する」→ もう一度新しい目標を尋ね、g.amend(...).write(goal_path) して ctx[GOAL_KEY] を更新し、False を返す。
12. それ以外(人が自分で書いた自由回答を含む)→ 「お前の判定が間違っている」とみなし、人の言い分を v.reason に記録して False を返す。

on_reject は同期だ:ctx[VERDICT_KEY].feedback() を返し、Verdict がなければ "" を返す (最初から走る動きに退化する)。

starter_flow()¶

def starter_flow(
    ask: str,
    *,
    workspace: str | Path = ".",
    run_dir: str | Path = "runs",
    new: bool = False,
    isolate: bool = False,
    clarify_only: bool = False,
    goal: bool = True,
    rounds: int = 3,
    judge_can_run: bool = False,
    max_asks: int | None = None,
    timeout_s: float | None = 1800.0,
    instructions: str = "",
    worker_prompt: str = "你负责实现。每改一处就跑一次验证,别攒到最后。",
    brief_name: str = "需求.md",
    goal_name: str = "目标.md",
    log_name: str = "问答记录.md",
) -> Workflow

そのまま動く3ステップのワークフローを組み立てる:要件確認 → 目標設定 → 作業(ゴールガード付き)。コマンドラインの flower が使っているのもこれだ。

固定の組み立て:

Workflow(name="starter", channel=ch, workbench=wb, steps=[...])
# ch = HumanChannel(log_path=<notes>/问答记录.md, amend_path=<brief_path>,
#                   max_asks=max_asks, timeout_s=timeout_s)
# コーディネーター = coordinator("协调者", "", {"coder": worker(..., isolate=isolate)}, channel=ch)

分岐する挙動:

• isolate=True かつ workspace が git リポジトリでない → ValueError を投げる。Agent ツールがエラーを出すまで待たない (そのときには既に金を使っている)。
• ウェイク判定:Brief.load(brief_path) が存在し complete() ならウェイクとみなす。ウェイクでなく ask が空 → ValueError("要给一句诉求,例如 flower '帮我做一个 X'") を投げる。
• ウェイク時、その一言は3か所に同時に落ちる。1つでも欠ければ黙って効かなくなる:ブリーフへの追記 (ch.amend(said, label="唤醒时追加")。既にファイルにあれば重複して書かない)、 goal_step(always_set=True) によるチェックリストの引き直し(引き直さないとジャッジが読むのは古い目標のまま)、 そしてコーディネーターへ直接届けること(その文脈にあるのは古い目標なので、渡さなければ古い基準で作業して新しい基準で判定されることになる)。

wake_state()¶

def wake_state(
    workspace: str | Path = ".",
    *,
    run_dir: str | Path = "runs",
    isolate: bool = False,
    brief_name: str = "需求.md",
    goal_name: str = "目标.md",
) -> dict

走り出す前の読み取り専用の探査で、1バイトも書かない。 実際に走らせる前に「これは前回の続きなのか、最初からなのか」を人に伝えるためのものだ。

返る dict:

ワークベンチの位置はここと starter_flow の2か所でだけ定義される:isolate=True → <ws>.parent/.flower-<ws.name> (リポジトリの外)。そうでなければ <ws>/.flower。ドライバがブリーフの場所を知りたいときもこの関数を通す —— 自分でパスを組み立てて間違えてもエラーにはならず、黙って効かなくなるだけだ。

ロールファクトリ¶

ソース:flower/core/roles.py

5 つのロールはすべてファクトリ関数だ。各ロール = 注入される 1 段のルールテキスト + 1 組のツール + 1 組の hook。 worker() が返すのは SDK の AgentDefinition(subagent に割り当てる用)で、残り 4 つは AgentSpec (自分でセッションを 1 本立てる)を返す。

ロール自体は hook を持たない —— ツールを止める仕事は Runtime._attempt が spec.delegate_only に応じて自動で装着する。 hook 層を参照。

内部のツール群定数(エクスポートされていないが、デフォルト値を決めている):

COORDINATOR_TOOLS = ["Agent", "TodoWrite", "Read"]
WEB_TOOLS         = ["WebFetch", "WebSearch"]
WORKER_TOOLS      = ["Read", "Write", "Edit", "Bash", "Glob", "Grep", "WebFetch", "WebSearch"]

coordinator()¶

def coordinator(
    name: str,
    instructions: str,
    workers: dict[str, AgentDefinition],
    *,
    channel: Any = None,
    can_read: bool = True,
    glance: bool = True,
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = None,
    max_budget_usd: float | None = None,
    permission_mode: str = "acceptEdits",
    compact: Any = None,
    hooks: dict[str, Any] | None = None,
    env: dict[str, str] | None = None,
) -> AgentSpec

メインスレッド上のコーディネーターを作る:タスクを分解し、割り当て、レポートを読み、決定を下す。 ただし自分では手を動かさない。最初の 3 つは位置引数。

返される AgentSpec で固定される 3 項目:delegate_only=True、agents=workers、 workbench は AgentSpec のデフォルト True のまま。

workers の web ツールがマージされる¶

リストを組み立てた後、coordinator() は各 AgentDefinition.tools を走査し、 WEB_TOOLS(WebFetch、WebSearch、roles.py:33)に含まれるものをコーディネーター自身の allowed_tools にも 1 つ追加する(roles.py:523-526)。

理由:allowed_tools は disallowed_tools と同じくセッション単位だからだ。 この点についてはこの文書中で最も硬い証拠であり —— 影響はメインスレッドだけに留まらない。このセッション単位のリストにないツールは、subagent が呼ぶときにもパーミッション承認を通る。 無人運転では承認する人がいないので、harness は Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected)を返し、モデルは同じ呼び出しを何度もリトライする。実測で踏んだ:ワーカーに WebFetch/WebSearch を追加したのに AgentDefinition.tools にしか書かず、その novel の実行は 20 回以上 user-rejected を出し、1 文字も書けなかった(roles.py:513-518)。

2 つのフィールドのセッション単位という性質は同じだが、症状が違う:disallowed_tools はその場でエラー、 allowed_tools は静かに死ぬまでリトライ。後者のほうが追いにくい。画面上は何もエラーらしく見えないからだ。

マージするのは読み取り専用で副作用のないものだけ。 Write/Edit/Bash は 意図的にマージしない:メインスレッドがそれらを無承認で使えるようになった時点で、 delegate_guard の「コーディネーターは手を動かさない」という壁が無意味になる。しかも subagent の Bash/Write はもともと通る(実測 462 回の許可、roles.py:520-522)。

ソースには disallowed_tools で「調整だけして手を動かさない」を実装するな と明記されている —— あれはセッション単位で、 subagent の Bash/Write まで一緒に禁止してしまう。AgentSpec の警告を参照。 正しいやり方はここにある delegate_only=True + allowed_tools に入れない、であり、 そのうえで delegate_guard が agent_id を見てメインスレッドだけを止める。

channel を渡すと 2 つのツールが一緒に来る。選べない:MCP server を付けた時点で両方が存在し、 allowed_tools は排他ではないので、書いても書かなくても呼べる。無人運転では ask のたびに timeout_s を丸ごと待つ —— その場面では HumanChannel(timeout_s=0) を使う。

worker()¶

def worker(
    description: str,
    prompt: str,
    *,
    tools: list[str] | None = None,
    model: str = "inherit",
    effort: str | int | None = None,
    max_turns: int | None = None,
    permission_mode: str | None = None,
    skills: list[str] | None = None,
    discipline: bool = True,
    isolate: bool = False,
) -> AgentDefinition

実際に作業する subagent の定義を作る。最初の 2 つは位置引数。 返るのは SDK の AgentDefinition で、そのまま coordinator(workers={...}) に入れられる。

isolate=True は workspace が git リポジトリであることを要求する。そうでなければ Agent ツールが直接 "not in a git repository" を返し、 静かに劣化はしない。しかもマークは Python の属性なので —— AgentDefinition に dataclasses.replace() をかけると 失われ、隔離が静かに無効化される。

clarify()¶

def clarify(
    name: str,
    channel: Any,
    *,
    instructions: str = "",
    can_read: bool = True,
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = None,
    max_budget_usd: float | None = None,
) -> AgentSpec

クラリファイアを作る:手を動かす前に要件を明確にする。作業はせず、質問だけをして、最後にちょうど 4 セクションを出力する。

返される AgentSpec:allowed_tools = [channel.tool_name] + (読める場合はその 5 つ)、 mcp_servers = channel.mcp_servers()、workbench=False(書き込みツールを持たないのでインデックスは無意味)、 permission_mode は AgentSpec のデフォルト "default" を継承。 Write / Edit / Bash / Agent はなく、inbox もない(コーディネーターとは違う)。

judge()¶

def judge(
    name: str,
    channel: Any,
    *,
    instructions: str = "",
    can_run: bool = False,
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = None,
    max_budget_usd: float | None = None,
) -> AgentSpec

ジャッジを作る:走り出す前に目標を設定するか、各ラウンドの終わりにそのラウンドを判定するか、どちらかをする。

返される AgentSpec:allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (can_run のとき)["Bash"]、 workbench=False、それ以外は clarify() と同じ。Write / Edit / Agent はなく、inbox もない。

トレードオフ:can_run=True は判定が硬くなる(受け入れコマンドを実際に走らせられる)が、代償としてジャッジがワークスペースを変更できるようになる —— Bash 自体がファイルを書けるからだ。絶対に中立な判定が欲しいなら開けない。

oracle()¶

def oracle(
    name: str = "旁路问答",
    *,
    instructions: str = "",
    model: str | None = None,
    effort: str | None = None,
    max_turns: int | None = 12,
    max_budget_usd: float | None = 0.5,
) -> AgentSpec

オラクルを作る:実行がまだ走っている最中に「いまどこまで進んだ?」と聞くと、直近のイベントとワークベンチを一目見てから答える。 その発言はその実行のコンテキストには入らない。

返される AgentSpec:allowed_tools = ["Read", "Glob", "Grep"](channel なし —— 質問はせず、 答えるだけ)、workbench=True(5 つのロールの中で唯一、コーディネーターでないのにワークベンチを開くもの —— 成果物とノートを読みに行くのが仕事だから)。

5 つのルールテキスト¶

5 つの定数はすべて __all__ に入っており、直接 import して読む・連結する・書き換えることができる。

agent 定義¶

ソース:flower/core/agent.py

AgentSpec は 1 つの専用 agent の完全な宣言であり、build_options がそれを SDK の ClaudeAgentOptions にコンパイルする。 ロールファクトリが返すのがこの AgentSpec だ —— ファクトリの外の組み合わせが必要なら、直接構築すればいい。

AgentSpec¶

@dataclass
class AgentSpec:
    name: str
    instructions: str
    allowed_tools: list[str] = field(default_factory=lambda: ["Read", "Glob", "Grep"])
    disallowed_tools: list[str] = field(default_factory=list)
    model: str | None = None
    effort: str | None = None
    max_turns: int | None = None
    max_budget_usd: float | None = None
    permission_mode: str = "default"
    agents: dict[str, Any] | None = None
    mcp_servers: dict[str, Any] = field(default_factory=dict)
    hooks: dict[str, Any] | None = None
    compact: CompactPolicy | None = None
    env: dict[str, str] = field(default_factory=dict)
    glance: bool = False
    workbench: bool = True
    delegate_only: bool = False

build_options()¶

def build_options(
    spec: AgentSpec,
    *,
    cwd: str | Path | None = None,
    session_store: SessionStore | None = None,
    resume: str | None = None,
    fork: bool = False,
    resume_at: str | None = None,
    use_plugin: bool = True,
    portable: bool = True,
    add_dirs: list[str] | None = None,
    flush: str = "eager",
    prelude: str = "",
) -> ClaudeAgentOptions

AgentSpec を SDK の ClaudeAgentOptions にコンパイルする。Runtime._attempt が内部で呼んでいるのがこれだ。 自分で SDK を駆動する(Runtime を使わない)場合もここから入る。

対応関係:

PLUGIN_DIR はリポジトリルートの plugin/(flower/core/agent.py から 3 階層上)。pip でインストールしただけではこのディレクトリが存在するとは限らないので、 コードは is_dir() で判定している。

CompactPolicy¶

@dataclass
class CompactPolicy:
    mode: str = "auto"
    window: int | None = None

    def env(self) -> dict[str, str]: ...

auto-compact のスイッチパネル。成果物は子プロセスに注入する環境変数の組だ。 compact のアルゴリズム自体は harness のバイナリの中にあり変更できない。変えられるのは「発火するかどうか」だけ。

HandoffPolicy¶

@dataclass
class HandoffPolicy:
    enabled: bool = True
    window: int = field(default_factory=default_window)
    headroom: int = 50_000
    max_generations: int = 8

    @property
    def at(self) -> int: ...        # max(10_000, window - headroom)
    @property
    def warn_at(self) -> int: ...   # max(1_000, at - 20_000)

コンテキストが埋まりかけたときに compact ではなく「引き継ぎ書を書いて新しいセッションに切り替える」ためのポリシーオブジェクト。

default_window()¶

def default_window() -> int

環境変数 ANTHROPIC_MODEL または ANTHROPIC_DEFAULT_OPUS_MODEL のモデル名文字列からコンテキストウィンドウを推測する:

デフォルトは強気の値を取る。大きく見積もっても致命傷ではない:API が prompt is too long で弾き返し、Runtime はこのシグナル (内部の is_overflow)を認識してその場で世代交代する —— ただしその世代の引き継ぎは劣化版になる。

文書¶

ソース:brief.py · handoff.py · goal.py

4 つの dataclass。どれも「モデルの返答を決まった数のセクションにパースし、ディスクに書き出す」ものだ。共通の形: parse() でパース、missing() / complete() で揃っているか確認、to_markdown() は人が読む用、 prompt_block() は下流のモデルが読む用、write() / load() で書き出しと読み戻し。

Brief¶

@dataclass
class Brief:
    goal: str = ""
    accept: str = ""
    bounds: str = ""
    unknowns: str = ""
    path: Path | None = field(default=None, compare=False)

要件確認書。ちょうど 4 セクション、順序は固定で goal → accept → bounds → unknowns。中国語のセクション名はそれぞれ「目标」「验收标准」「边界」「未知与假设」。

パースのルール(間違いが集中するところ):

• フェンスを剥がすとき、**閉じられていない `` や_~に出会ったらそこから後ろを丸ごと捨てる** —— 実測でクラリファイアはコード全体を返答に貼り付けてくる。 モデルの出力が途中で切れると、以降のセクションが一切パースできなくなり、complete()がFalse` になって gate が差し戻す。
• 見出しの正規表現は ## 目标 / **目标** / 目标: / 3. 边界 を許容し、見出しの直後に本文が続く形も許容する。
• 別名テーブルは長さの降順でコンパイルする。そうしないと「未知」が「未知与假设」を先に食ってしまう。
• 同名のセクションが複数回現れた場合は最初に中身のあるものを取る。
• 確認書を手で編集するときに to_markdown() の "(未填)" プレースホルダをそのまま写すと、そのセクションは依然として欠けている扱いになる。

Handoff¶

@dataclass
class Handoff:
    doing: str = ""
    decided: str = ""
    deadends: str = ""
    next: str = ""
    scene: str = ""
    step: str = ""
    path: Path | None = field(default=None, compare=False)

世代交代のときに書く引き継ぎ書。5 セクション。

必須は doing と next の 2 セクションだけ —— 「通らなかった道」を空にできないと強制すると、モデルは捏造する。

同じモジュール内にある、エクスポートされていないが意味的に重要な 3 つのメンバー:is_overflow(*texts) は prompt is too long、 context length exceeded、maximum context length、too many total text bytes、 input length and max_tokens exceed などにマッチし、「ハードエラー」を「その場での世代交代」に変える;HANDOFF_PROMPT は 現在のセッション自身に引き継ぎを書かせるプロンプト({used} {window} の 2 つのプレースホルダを含む。新しいロールではない —— そのコンテキストを持っているのは自分だけだから);degraded(step, prompt, *, why="") は引き継ぎが書けなかったときに機械的に 1 部を組み立て、 scene には元のタスクの先頭 1200 文字を入れる。

Goal¶

@dataclass
class Goal:
    statement: str = ""
    checks: list[str] = field(default_factory=list)
    path: Path | None = None

目標ガードにおける目標 + 判定チェックリスト。

Verdict¶

@dataclass
class Verdict:
    state: str = ""
    reason: str = ""
    failed: list[str] = field(default_factory=list)

ジャッジによる 1 ラウンドの判定結果。ちょうど 3 セクション:結論 / 理由 / 未通過。

parse の識別順序:

1. まず見出しセクションで「结论」/「判定」を取る。
2. 見出しセクションがない場合、全体を strip して fullmatch(r"1|true") → 達成;fullmatch(r"0|false") → 未達。
3. それ以外は結論テキストの中を状態語テーブル(長い語が先)で走査し、最初にヒットした語を取る。 「无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable」はすべて unreachable に寄せる —— 実測で踏んだ:ターゲットプラットフォームが macOS で Linux コンテナ上で走っており、ジャッジがソースの分岐を見ただけで合格を出した。
4. それでも決まらなければ、孤立した \b1\b を探して達成、\b0\b を探して未達。
5. すべて外れたら state=""、ok=False。

unreachable と not_yet は別々の結論だ:前者は「立ち止まって人に聞く」道に進むのであって、「もう 1 ラウンド」ではない。

hook 層¶

ソース:flower/core/guard.py

この層は flower の実行境界である:どのツールを main thread に触らせないか、長すぎる結果をどう刈るか、 どの役割を独立した worktree に送るか、すべて SDK hook が強制する。prompt には頼らない。理由は単純で —— prompt は提案でしかなく、モデルは従わなくてよい:system prompt に「worktree を使うな」と明記していても、 isolate_guard の注入はそのまま効くことを実測している(モデルが渡したのは None、実際に適用されたのは 'worktree')。

エクスポートは 9 つ:HookMatcher を返す guard ファクトリ 5 つ(whitelist_guard は None を返しうる)、組み立て関数 1 つ、マージ関数 1 つ、isolation マーカー関数 2 つ。 手で登録する必要はない —— Runtime が AgentSpec に応じて自動で組み立てる。手で登録するのは自分で SDK を直接駆動する (Runtime を通さない)場合だけだ。

main thread の判定は 1 つの関数に統一されている:_is_main_thread(data) = not data.get("agent_id") —— subagent の tool-lifecycle hook データには agent_id が入り、main thread には入らない。 「main thread だけを止める」guard はすべてこの 1 行に依っている。

ツールグループ定数(モジュールレベル、未エクスポート。ただしデフォルト matcher を決める):

HANDS_ON   = "Bash|Write|Edit|NotebookEdit"
WRITE_ONLY = "Write|Edit|NotebookEdit"
BULKY      = "Bash|Read|Grep|Glob|WebFetch|WebSearch"

早見表:どの guard がどの SDK イベントに付くか¶

この表から読み取れる重要な帰結:Runtime(workbench=False) のときは workbench_hooks が丸ごと登録されない。 そして delegate_only=True の coordinator では whitelist_guard もスキップされる —— main thread には壁が 1 枚も無い。 詳細は Runtime の警告を参照。

whitelist_guard()¶

def whitelist_guard(allowed: list[str] | None, *, role: str = "这个角色") -> HookMatcher | None

allowed_tools を、手を動かす 4 つのツールに対して本当に排他的にする。

• PreToolUse に付く。matcher は "|".join(banned)、banned = Bash Write Edit NotebookEdit のうち allowed に無いもの。
• ヒットしたら permissionDecision: "deny"。文面の趣旨:「XX には YY が無い。これは意図的であって、設定漏れではない。 結論は返答本文に書け、フレームワークはそこから拾う —— 別の書き方で回避しようとするな。」
• 止めるのはこのセッションの main thread だけで、subagent は通す —— subagent のツールは AgentDefinition.tools が決める。
• 止めるべきツールが 1 つも無い場合は None を返す(たとえば worker() のようなフルセットの役割)。呼び出し側はこれを見て登録するかを決める。

なぜこれが必須か:allowed_tools は承認不要リストであって、排他的ホワイトリストではない。実測での証拠が 2 つ —— 目標を設定した judge が Bash を 11 回実行した;$0.1 のプローブでは allowed_tools=["Read"] の agent が普通に Write/Bash を呼べた。 したがって clarify() / judge() の「書き込みツールが無い」状態はこの hook によるものであって、ホワイトリスト自体によるものではない。

利点は allowed_tools から派生している点で、judge(can_run=True) は自動的に Bash を残しつつ Write/Edit は止め続ける —— 追加のスイッチは要らない。

delegate_guard()¶

def delegate_guard(*, tools: str = HANDS_ON, allow_glance: bool = False) -> HookMatcher

main thread が自分で手を動かす → 拒否し、行き先を示す。

• PreToolUse に付く。matcher はそのまま tools。
• main thread がこの 4 つのツールを呼ぶ → deny。理由の中で次に何をすべきかを示す:Agent ツールで subagent を派遣し、 タスクに目標と受け入れ基準を明記し、長い成果物は .flower/artifacts/ に書き、返答にはパスと結論だけを載せるよう要求する。
• subagent は一律に通す。

whitelist_guard との違いは言い回しである:止めるツールは同じだが、こちらは「人を派遣しろ」と言うので、より筋が通る。 だから delegate_only=True の役割にはこちらだけを登録する。両方登録するとモデルは矛盾した 2 つの指示を受け取る。

allow_glance=True の許可判定と「結果が刈られるかどうか」は同じ関数(is_ephemeral)である —— 許可集合は期限切れ集合と一致していなければならず、片方を変えたらもう片方も変えなければならない。

spill_guard()¶

def spill_guard(
    workbench: Workbench,
    *,
    threshold: int = 4000,
    tools: str = BULKY,
    main_only: bool = False,
) -> HookMatcher

閾値を超えたツール結果をその場でspill し、コンテキストには 1 行のポインタだけを残す —— コンテキストが埋まってから遡って compact するのではない。

• PostToolUse に付き、 {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <刈った後>}} を返す。
• spill ファイル名は内容の sha256 先頭 16 桁 + .txt。コンテキストでは 1 行のポインタ + 先頭 400 文字に置き換わる。
• updatedToolOutput は元のツールの出力構造を保たなければならないので、dict 内の長すぎる文字列フィールドだけを差し替える。 list には一切触れない(中身が画像ブロックである可能性がある)。構造が合わないと拒否される(原文のまま、エラーは出ない)。
• spill ファイル自体を読むときは必ず通す —— でないと「Read でそれを読む」は空文句になる:読み戻した全文がまた spill され、無限ループになる。 実測でぶつかった。モデルは 5 通りの書き方で回避を試みた。

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

workbench に何かを書いたら INDEX.md を更新し、次の agent が開始時点でその存在を知れるようにする。

PostToolUse に付き、matcher は "Write|Edit"。tool_input["file_path"] を resolve した結果が workbench.root 内なら workbench.refresh() を呼ぶ。常に {} を返す —— 何も書き換えず、副作用だけを持つ。

isolate_guard()¶

def isolate_guard(agents: dict[str, AgentDefinition], *, on_inject: Any = None) -> HookMatcher

役割ごとに subagent へ独立した git worktree を割り当て、isolation を実現する。

PreToolUse に付き、matcher は "Agent"。3 つの条件が同時に成り立つときだけ注入する:tool_name == "Agent"、 tool_input に cwd も isolation も無い、subagent_type に対応する役割が isolated() でマークされている。 成立すれば permissionDecision: "allow" + updatedInput(isolation を "worktree" に設定)を返す。

isolation と cwd は Agent ツールにおいて排他である —— モデル自身が cwd を指定したならそれを尊重する。 「隔離するかどうか」は役割の属性であって、グローバルスイッチでも、派遣のたびに判断するものでもない。隔離が不要な役割には 1 バイトも追加されない。

隔離を有効にするなら workbench をリポジトリの外に出すこと。 隔離された agent は共有 checkout に書き込めないので、 workbench は home= でリポジトリ外を指す必要がある。starter_flow(isolate=True) は <ws>.parent/.flower-<ws.name> を、Runtime(workbench=True) は <run_dir>/workbench を使う —— どちらもリポジトリ外だが、同じディレクトリではない。混ぜて使わないこと。

isolated() / wants_isolation()¶

def isolated(agent: AgentDefinition, flag: bool = True) -> AgentDefinition
def wants_isolation(agent: AgentDefinition | None) -> bool

subagent 定義に「独立した作業領域が必要」というマークを付け、またそれを読み戻す。

マークは object.__setattr__ で付けた Python 側の属性 _flower_isolate であって、dataclass のフィールドではない —— SDK は asdict() でシリアライズし、宣言済みフィールドしか見ないので、このマークが CLI 側に漏れることはない(実測済み)。

代償:AgentDefinition に dataclasses.replace() をかけるとこのマークが失われ、隔離が黙って無効になる。

worker(isolate=True) は内部で isolated() を使っている。

workbench_hooks()¶

def workbench_hooks(
    workbench: Workbench,
    *,
    delegate_only: bool = True,
    spill_threshold: int | None = 4000,
    agents: dict[str, AgentDefinition] | None = None,
    allow_glance: bool = False,
) -> dict[str, list[HookMatcher]]

workbench に必要な hook をまとめて登録する。Runtime._attempt が呼ぶのはこれ。

出力:

• PreToolUse:delegate_only=True → [delegate_guard(allow_glance=allow_glance)]; マーク済みの役割があれば isolate_guard(agents) を追加。
• PostToolUse:常に [index_guard(workbench)];spill_threshold が真なら spill_guard(workbench, threshold=spill_threshold) を追加。
• 空リストになるイベントキーは削られ、空 list は返らない。

merge_hooks()¶

def merge_hooks(*groups: dict[str, list[Any]] | None) -> dict[str, list[Any]]

イベント名ごとに複数の hook 設定を連結する。

extend を使い、重複除去はしない —— 同じ guard を 2 回渡せば 2 回登録される。Runtime はこれで spec.hooks、 workbench_hooks(...)、whitelist_guard を 1 つにまとめる。

workbench¶

ソース:flower/core/workbench.py

Workbench¶

@dataclass
class Workbench:
    workspace: Path
    dirname: str = ".flower"
    max_index_entries: int = 40
    home: Path | None = None

spill 用の作業ディレクトリ。サブディレクトリ 3 つ + インデックス 1 つ。インデックスは system prompt に注入されるので、 agent は毎ターン手元に何があるかを知っている。

スクリプトの自己記述フォーマット:先頭 8 行以内の # desc: 一文(// と -- のコメント記号も認識)。 無ければ最初の非空コメント行、または docstring の先頭行にフォールバック(100 文字で切る)。

prompt_block() が注入する 3 つのルール:

1. 2 回目も実行するスクリプトは scripts/ に書き、先頭行に # desc: を付ける。
2. 2000 文字を超える成果物は artifacts/ に書き、会話にはパスと結論だけを載せる。
3. 重要な決定は notes/ に書き、1 決定 1 ファイルにする。

external=True のとき、prompt_block() は「絶対パスでアクセスすること」の一文を追加で挿入する。

インデックスは subagent には継承されない。 これはセッションレベルの system_prompt.append を通っており、subagent は自分の system prompt を持つ(実測 $0.2461)。したがって「長い成果物は artifacts/ へ、workbench はどこにあるか」の 2 点は coordinator がtask brief の中で伝え直すしかない —— それが唯一の経路であって、冗長ではない。 WORKER_RULES には意図的に書いていない:実際のパスは Workbench が生成するもので、ハードコードすれば間違う。

session store¶

ソース:sqlite.py · trim.py · prune.py

3 層の継承:SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore。 Runtime は常に最外層を使い、3 層それぞれの方針はコンストラクタ引数で制御する。

3 層はそれぞれ 1 つのことを担う:永続化、サイズと価値によるtrim、「エラーかどうか」によるprune。 trim も prune も load()(つまり resume が履歴をモデルに戻す)瞬間に起きるだけで、SQLite 内の生レコードは 1 バイトも変わらない。

SqliteSessionStore¶

class SqliteSessionStore(SessionStore):
    def __init__(self, path: str | Path) -> None

SDK の SessionStore プロトコルを実装する。テーブルは 3 つ、entries / meta / summaries。 store key は project_key/session_id[/subpath] —— サブ agent の transcript は subpath で区別する。

内部の _next_mtime は厳密な単調性を保証する —— list_sessions と summary sidecar がこの時計を共有しており、 そうでないと SDK の staleness 高速パスが誤判定する。

TrimPolicy¶

@dataclass
class TrimPolicy:
    keep_recent: int = 20
    min_chars: int = 2000
    spill_dirname: str = ".flower/spill"
    enabled: bool = True

2 つの spill ディレクトリは同じものではない。 spill_guard は <workbench.root>/spill/(ワークスペース外でもよい)に落とし、 TrimPolicy.spill_dirname は <workspace>/.flower/spill/(ワークスペース内でなければならない)に落とす。 それぞれ「その場で刈る」と「resume 時に刈る」に対応しており、ディレクトリが違うのは意図的だ。1 つにまとめないこと。

EphemeralPolicy¶

@dataclass
class EphemeralPolicy:
    enabled: bool = True
    keep_recent: int = 6
    max_chars: int = 2000
    text: str = "[{cmd} 的结果已过期(第 {age} 轮前),当前状态可能已变。需要请重新执行]"

ephemeral command の結果に対する期限切れ方針。

Bash ツールの結果にのみ作用し、かつコマンドが ephemeral command のホワイトリストに一致する必要がある。Read は対象外 —— ファイルの内容は、時間の経過によって誤解を招くほど劣化するものではない。期限切れの内容はspill せず、そのまま捨てる。

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

Bash コマンドが ephemeral command かどうかを判定する。 delegate_guard の許可判定と trim の期限切れ判定は、この 1 つの関数を共有している —— coordinator が自分で実行できるコマンドの集合は、 結果が期限切れとマークされる集合と一致しなければならない。許可するのに trim しなければ、期限切れの git status が永久にコンテキストを占め、しかも誤導する。 trim するのに許可しなければ、coordinator は ls 1 本のために subagent を派遣し、4.3k の起動コストで数十文字を得ることになる。

判定の順序:

1. 空 / 全部空白 → False。
2. コマンド置換($(、バッククォート、<(、>()、または「状態を変える」書き方にヒット → False。
3. 安全なリダイレクト(2>&1、&> /dev/null など)を取り除いてもなお > または < を含む → False。
4. && / || / ; / | を取り除いてもなお単独の & が残る(バックグラウンド実行)→ False。
5. && / || / ; / | で分割し、各セグメントがすべてホワイトリストにヒットしなければならない。

ホワイトリストの動詞の大分類:読み取り専用の git サブコマンド(status diff log show branch rev-parse など)、 ディレクトリとシステム情報(ls pwd df du date whoami env など)、プロセスとコンテナ (ps top lsof docker ps kubectl get など)、ファイル閲覧(cat head tail wc stat find tree)、 パス検索(which whereis command -v type)、テキスト処理(grep rg sort uniq awk sed jq diff など)。

動詞がホワイトリストにあっても、次の書き方は止められる:xargs、exec、eval、source、tee、 find -delete / -ok / -fprint、sed -i、sort -o、awk 内の system( と print >、 git branch -D/-d/-m、git * --force/--hard/--prune。

初版は複合コマンドを一律に拒否していたが、実測で glance が完全に機能しなくなった(coordinator の 3 回の試行がすべて止められた)ので、 セグメントごとの判定に変えた。

TrimmingSessionStore¶

class TrimmingSessionStore(SqliteSessionStore):
    def __init__(
        self,
        path: str | Path,
        workspace: str | Path,
        policy: TrimPolicy | None = None,
        ephemeral: EphemeralPolicy | None = None,
    ) -> None

公開属性:workspace、policy、ephemeral、last_report: dict[str, int]。

load() の順序:super().load() → last_report をクリア → ephemeral.enabled なら expire() → policy.enabled なら trim()。enabled=False のときはそのステップを丸ごとスキップする。

刈るのはプレーンテキストだけ:image / document ブロックはそのまま残す。

構造上の 2 つのレッドライン:tool_result ブロック自体は必ず残す。差し替えてよいのは content だけ(1 つでも欠ければ 「Missing Tool Result Block」になる);isCompactSummary のエントリには触れない。

trim_report()¶