Python API¶

Эта страница исчерпывающе описывает 62 публичных символа верхнего уровня __all__ в flower: сигнатуры, параметры, значения по умолчанию, семантику, публичные атрибуты и методы. После прочтения не нужно лезть в исходники за параметрами.

Материал организован по тому, что вас интересует, а не по файлам модулей — хотите знать «как не дать координатору делать всё самому», идите в слой hook; хотите знать «как результат предыдущего шага попадает в следующий», идите в workflow. Терминология — строго по глоссарию.

Версия 0.1.0, зависимость claude-agent-sdk>=0.2.152. Все сигнатуры дословно соответствуют исходникам.

from flower import Runtime, Workflow, Step, coordinator, worker   # один импорт верхнего уровня

Что на этой странице¶

Шесть значений по умолчанию, которые кусаются¶

Эти шесть — не мелочи, а шесть самых частых аварий. Каждая полностью разобрана в соответствующем разделе.

Runtime¶

Исходник: flower/core/runtime.py

Runtime — ядро исполнения. Он держит рабочую область, хранилище сессий, workbench, политику устойчивости и политику handoff, а наружу отдаёт один глагол: run один шаг. Повторы, продолжение после прерывания, handoff при переполнении контекста — всё делается внутри этого одного вызова.

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 обязателен.

Хранилище сессий зашито жёстко: это всегда PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...)). Параметры конструктора не дают точки входа для смены бэкенда — если нужно сменить, соберите AgentSpec + build_options(session_store=...) сами либо перезапишите rt.store после конструирования.

Последние два шага конструктора — 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). Дальше идёт цикл с четырьмя выходами:

1. Успех → выход из цикла.
2. Прерывание человеком (result.error == INTERRUPTED) → не ограничено max_attempts, сети не ждёт. 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-документ → resume=None, fork=False (совершенно новая сессия) → 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 семантика добавления: при каждой записи файл перечитывается с диска и дедуплицируется по полю run (своя строка обновляется, чужие остаются), поэтому два параллельных flower в одном run_dir — безопасно, при условии что run_id не совпадут.

Три точки наблюдения за handoff (все — Event("handoff"), различаются по payload["phase"]): near (приближение к warn_at, отправляется один раз на поколение), writing (идёт запись handoff, это больше десятка секунд), done (в payload есть degraded / path / sections). Раунд записи handoff запускается с replace(spec, max_budget_usd=None) — handoff обязан быть написан и не должен упереться в бюджет; и on_event=None, этот раунд в UI не выводится.

Handoff пишется на диск в <workbench.notes>/交接-<步骤名>.md; без workbench на диск ничего не пишется, документ всё равно передаётся преемнику через 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

Весь учёт по одному завершённому шагу.

Workflow¶

Исходники: flower/workflow/

Workflow — это набор шагов, выстроенных по порядку, плюс правила передачи состояния между шагами и условия досрочного выхода. Фреймворк не даёт готовых workflow, 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, ...). Первые три поля — позиционные аргументы, Step("取词", terse, "读 seed.txt …") — допустимая запись.

Три способа подключить сессию (в пределах одного run):

on_reject меняет семантику повтора:

• Не задан → следующая попытка стартует с нуля (тот же prompt, тот же resume_from).
• Задан → следующая попытка продолжает ту самую сессию, которую только что забраковали, prompt заменяется на возвращённое значение, fork принудительно False.
• Возвращена пустая строка → отбоя нет, деградирует до старта с нуля.
• result.session_id равен None → тоже деградирует до старта с нуля.

Три значения on_fail:

Независимо от того, пройден gate или нет, 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 (все начинаются с _, с именами шагов не сталкиваются):

Payload события Event("step"): {"index": i, "total": len(steps), "resumed": bool, "woke": int}.

Метки повторов: на нулевой попытке — step.name; далее при наличии on_reject — f"{name}#round{attempt+1}", без него — f"{name}#retry{attempt}". В manifest сразу видно, как этот шаг доехал до конца. Имена с суффиксом в межпроцессный lineage не попадают — Lineage.remember использует исходное имя.

runtime.on_session оборачивает ровно один вызов runtime.run, а try/finally гарантирует, что до gate он снят. prompt_cur / resume_cur / fork_cur — локальные переменные, обратно в step они не пишутся: один и тот же объект Step может быть запущен второй раз.

StepAbort¶

class StepAbort(Exception): ...

Брошено из gate = немедленно стоп, повторов не будет. Отличие от «вернуть False»: False — это «в этот раз не вышло, ещё круг»; StepAbort — «ещё круг не поможет».

После броска: ctx["_aborted"] = str(exc), passed = False, выход из цикла повторов (остаток retries не тратится), далее — обычная обработка провала по on_fail (по умолчанию "stop").

with_goal бросает его в двух местах: когда не удалось получить ctx["_runtime"], и когда verdict — 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, выполняющий clarify: выяснить требования → разобрать в Brief → если все четыре раздела на месте, заморозить и записать на диск.

Поля получившегося 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: следующий шаг — новая сессия, ему достаётся только brief, но не тот диалог. Вопросы и ответы clarify никогда не попадали в контекст координатора — их не вырезали оттуда, их там и не было.

Три места, куда заливается ctx: 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, задающий цель: judge читает brief, пишет цель + список проверок, результат разбирается в Goal, замораживается и пишется на диск. Форма та же, что у clarify_step.

Формального параметра can_run нет — чтобы judge, задающий цель, мог запускать команды, can_run=True передаётся только через **spec_kw. Без этого у него не будет Bash, и пункт «сначала разберись, в какой ты среде» из JUDGE_RULES выполнить нечем.

Помимо разбора и заморозки gate делает ещё одно: если в цели есть пункты [此环境无法验证:…], он тут же шлёт через ctx["_on_event"] событие Event("task", payload={"unverifiable", "total", "path"}) как предупреждение — судьба этих пунктов решается именно в момент постановки цели, а к моменту вынесения verdict деньги за целый круг работы уже потрачены.

resume_prompt не задан — при постановке цели brief и должен пересылаться целиком.

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 goal guard: после каждого круга judge независимо выносит verdict, и если цель не достигнута — отбивает работу обратно.

Результат — replace(step, retries=max(0, rounds - 1), gate=<новый gate>, on_reject=<новый on_reject>): используется dataclasses.replace, а не пересборка по полям, потому что при пересборке однажды потеряли resume_prompt, и ошибки при этом не было, просто при продолжении заново отправлялся весь brief целиком.

gate — async, порядок действий:

1. Нет ctx["_runtime"] → бросить StepAbort («не удалось получить Runtime, verdict по цели вынести нельзя»). Не притворяйтесь, что прошло.
2. ctx[ROUND_KEY] += 1.
3. Взять цель: сначала полный Goal из ctx[GOAL_KEY], иначе Goal.load(goal_path), иначе пустой Goal().
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...). Judge — это отдельный вызов Runtime.run, resume всегда None — это всегда новая сессия; step_name содержит номер круга, поэтому в межпроцессный lineage он не попадает.
5. Verdict.parse(vr.text) записывается в ctx[VERDICT_KEY].
6. v.achieved → вернуть True.
7. Не unreachable (включая размытые случаи с v.ok=False) → при размытости дописать дефолтную reason, вернуть False. Размытое всегда считается недостигнутым — нельзя, чтобы фраза «вроде бы работает» закрывала работу.
8. unreachable → await channel.ask(...), спросить человека, три варианта:
9. Никто не ответил (a.state != "answered") → бросить StepAbort. Крутиться вхолостую — самый дорогой выбор.
10. «Принять этот результат и идти дальше» → вернуть True.
11. «Изменить цель» → спросить новую цель, g.amend(...).write(goal_path), обновить ctx[GOAL_KEY], вернуть False.
12. Всё остальное (включая свободный ответ человека) → трактуется как «ты вынес неверный verdict», слова человека записываются в 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

Собирает готовый к запуску workflow из трёх шагов: уточнить требования → задать цель → работать (с goal guard). Именно его использует команда 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'").
• При пробуждении эта фраза уходит в три места сразу, и потеря любого из них тихо ломает всё: дописывается в brief (ch.amend(said, label="唤醒时追加"), если уже есть в файле — повторно не пишется); заставляет goal_step(always_set=True) переписать список проверок (без этого judge читает старую цель); и подаётся напрямую координатору (в его контексте лежит старая цель, и без этой фразы он будет работать по старым критериям, а оцениваться по новым).

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

Разведка только на чтение до старта, не пишет ни байта. Нужна, чтобы до реального запуска сказать человеку: «это продолжение прошлого раза» или «начинаем с нуля».

Возвращаемый dict:

Положение workbench определяется ровно один раз здесь и в starter_flow: isolate=True → <ws>.parent/.flower-<ws.name> (вне репозитория); иначе <ws>/.flower. Драйвер, желающий узнать, где лежит brief, тоже идёт через эту функцию — если склеить путь самому и ошибиться, ошибки не будет, всё просто тихо перестанет работать.

Фабрика ролей¶

Исходный код: flower/core/roles.py

Все пять ролей — фабричные функции. Каждая роль = фрагмент внедряемого текста правил + набор инструментов + набор hook'ов. worker() возвращает AgentDefinition из SDK (для передачи subagent'у), остальные четыре — AgentSpec (каждая поднимает собственную session).

Сами роли 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

Создаёт coordinator, живущий в main thread: разбирает задачу, раздаёт работу, читает отчёты, принимает решения, но сам руками ничего не делает. Первые три параметра — позиционные.

Три поля в полученном AgentSpec фиксированы: delegate_only=True, agents=workers, а workbench остаётся со значением по умолчанию True из AgentSpec.

Web-инструменты из workers вливаются сюда же¶

После сборки списка coordinator() проходит по AgentDefinition.tools каждого worker'а, и всё, что попадает в WEB_TOOLS (WebFetch, WebSearch, roles.py:33), добавляется ещё и в собственный allowed_tools координатора (roles.py:523-526).

Причина: allowed_tools, как и disallowed_tools, действует на уровне session. Это самое жёсткое во всём тексте доказательство данного факта — влияние не ограничивается main thread. Инструмент, которого нет в этом списке уровня session, при вызове из subagent'а тоже пойдёт через запрос разрешения; в автономном режиме подтверждать некому, и harness отвечает Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected), а модель раз за разом повторяет тот же вызов. На этом реально спотыкались: worker'у дали WebFetch/WebSearch, но записали их только в AgentDefinition.tools — тот прогон novel получил больше двадцати user-rejected и не написал ни слова (roles.py:513-518).

Оба поля одинаково действуют на уровне session, но симптомы разные: disallowed_tools даёт ошибку сразу, allowed_tools — молчаливые повторы до самой смерти. Второе искать труднее, потому что на экране ничто не похоже на ошибку.

Вливаются только read-only инструменты без побочных эффектов. Write/Edit/Bash намеренно не вливаются: как только main thread получит их без подтверждения, стена «координатор руками не работает», выстроенная delegate_guard, теряет смысл; а Bash/Write у subagent'а и так проходят (замерено 462 пропуска, roles.py:520-522).

В исходниках прямо написано: не реализуйте «только координировать, руками не работать» через disallowed_tools — это уровень session, и запрет заодно отключит Bash/Write у subagent'ов, см. предупреждение в AgentSpec. Правильный способ — как здесь: delegate_only=True + не давать allowed_tools, а delegate_guard по agent_id перехватит только main thread.

Если задан channel, приходят сразу два инструмента, это не опция: как только подключён 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, который действительно делает работу. Первые два параметра — позиционные. Возвращается AgentDefinition из SDK, его сразу можно класть в coordinator(workers={...}).

isolate=True требует, чтобы workspace был git-репозиторием, иначе инструмент Agent сразу выдаёт "not in a git repository", без молчаливой деградации. Кроме того, метка — это атрибут Python: dataclasses.replace() над AgentDefinition её потеряет, и изоляция молча перестанет работать.

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

Создаёт clarifier: прежде чем что-то делать, выясняет требования; ничего не делает, только спрашивает, а в конце выдаёт ровно четыре раздела.

Полученный AgentSpec: allowed_tools = [channel.tool_name] + (те пять при can_read), 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

Создаёт judge: он либо задаёт цель перед стартом, либо выносит вердикт после каждого раунда.

Полученный AgentSpec: allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (при can_run) ["Bash"], workbench=False, остальное как у clarify(). Нет ни Write / Edit / Agent, ни inbox.

Компромисс: при can_run=True вердикт жёстче (можно реально запустить приёмочные команды), ценой того, что judge получает возможность менять workspace — 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

Создаёт oracle: пока run ещё идёт, его можно спросить «где мы сейчас», и он посмотрит на последние events и workbench, а потом ответит. Сказанное им не попадает в контекст этого run.

Полученный AgentSpec: allowed_tools = ["Read", "Glob", "Grep"] (без channel — он не спрашивает, а только отвечает), workbench=True (единственный из пяти, кто не координатор, но с включённым workbench — ему как раз нужно читать эти артефакты и заметки).

Пять текстов правил¶

Все пять констант есть в __all__, их можно напрямую import, читать, склеивать и менять.

Определение agent¶

Исходный код: flower/core/agent.py

AgentSpec — полное объявление специализированного agent, а build_options компилирует его в ClaudeAgentOptions из SDK. Фабрика ролей как раз и возвращает 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 в ClaudeAgentOptions из SDK. Именно её вызывает Runtime._attempt внутри; если вы сами управляете SDK (без Runtime), вход тоже здесь.

Соответствие:

PLUGIN_DIR — это каталог plugin/ в корне репозитория (на три уровня вверх относительно flower/core/agent.py). После установки через 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)

Объект политики: когда контекст почти заполнен, «написать handoff document и перейти в новую session» вместо compact.

default_window()¶

def default_window() -> int

Угадывает окно контекста по строке имени модели из переменных окружения ANTHROPIC_MODEL или ANTHROPIC_DEFAULT_OPUS_MODEL:

По умолчанию берётся оптимистичное значение. Переоценка — не жёсткая ошибка: API вернёт prompt is too long, Runtime распознаёт этот сигнал (внутренняя is_overflow) и делает handoff на месте — но handoff этого поколения будет деградированной версией.

Документы¶

Исходный код: brief.py · handoff.py · goal.py

Четыре 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)

Brief, ровно четыре раздела, фиксированный порядок goal → accept → bounds → unknowns, китайские названия разделов соответственно «目标», «验收标准», «边界», «未知与假设».

Правила разбора (здесь сконцентрированы все ловушки):

• При снятии ограждений **незакрытый `` или_~приводит к отбрасыванию всего от него и дальше** — на практике clarifier вставлял в ответ весь код целиком. Когда вывод модели обрезан, все последующие разделы не разбираются,complete()даётFalse`, и gate отправляет на повтор.
• Регулярка заголовка допускает ## 目标 / **目标** / 目标: / 3. 边界, а также текст сразу после заголовка.
• Таблица псевдонимов компилируется в порядке убывания длины, иначе «未知» съест «未知与假设» раньше.
• При повторе одноимённых разделов берётся первый непустой.
• Если при ручном редактировании brief вы скопировали заглушку "(未填)" из 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)

Handoff document, который пишется при handoff, пять разделов.

Обязательны только два раздела — doing и next: жёсткое требование непустых «тупиковых путей» вынудило бы модель их выдумывать.

Три члена того же модуля, не экспортируемые, но семантически ключевые: 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»; HANDOFF_PROMPT — тот промпт, который заставляет текущую session саму написать handoff (содержит два плейсхолдера {used} и {window}, это не новая роль — только у неё самой есть этот контекст); degraded(step, prompt, *, why="") механически собирает handoff, когда его не удалось написать, запихивая в scene первые 1200 символов исходной задачи.

Goal¶

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

Цель и список проверок в goal guard.

Verdict¶

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

Результат одного раунда работы judge, ровно три раздела: вывод / обоснование / не пройдено.

Порядок распознавания в parse:

1. Сначала берётся раздел по заголовку «结论» / «判定».
2. Если разделов с заголовками нет, весь текст после strip: fullmatch(r"1|true") → достигнуто; fullmatch(r"0|false") → не достигнуто.
3. Иначе в тексте вывода ищется первое совпадение по таблице слов состояния (длинные слова первыми). «无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable» — всё это относится к unreachable — на этом реально спотыкались: целевая платформа macOS, запуск в Linux-контейнере, а judge посмотрел на ветку в исходниках и признал пройденным.
4. Всё ещё нет — ищется отдельно стоящее \b1\b → достигнуто, \b0\b → не достигнуто.
5. Ничего не подошло → state="", ok=False.

unreachable и not_yet — два разных вывода: первый ведёт по ветке «остановиться и спросить человека», а не «ещё один раунд».

Слой hook¶

Исходник: flower/core/guard.py

Этот слой — граница исполнения flower: каких инструментов не должен касаться главный поток, как подрезать сверхдлинные результаты, какая роль уходит в отдельный worktree — всё это принудительно задаётся hook'ами SDK, а не промптом. Причина прямая: промпт — это рекомендация, модель может её проигнорировать. Замерено: даже когда в системном промпте прямо написано «не используй worktree», инъекция isolate_guard всё равно срабатывает (модель передаёт None, а на диск ложится 'worktree').

Девять экспортов: пять фабрик guard'ов, возвращающих HookMatcher (whitelist_guard может вернуть None), один сборщик, один объединитель, две функции маркировки изоляции. Вручную их вешать не нужно — Runtime собирает всё автоматически по AgentSpec. Ручная установка нужна только если вы сами управляете SDK (минуя Runtime).

Определение главного потока идёт через одну функцию: _is_main_thread(data) = not data.get("agent_id") — в данных tool-lifecycle hook'а у subagent есть agent_id, у главного потока его нет. На этом держатся все guard'ы, которые «перехватывают только главный поток».

Константы групп инструментов (уровня модуля, не экспортируются, но определяют 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 пропускается ещё и whitelist_guard — у главного потока не остаётся ни одной стены. Подробности — в предупреждении в разделе Runtime.

whitelist_guard()¶

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

Делает allowed_tools действительно исключающим для тех четырёх инструментов, что работают руками.

• Висит на PreToolUse, matcher — "|".join(banned), где banned = те из Bash Write Edit NotebookEdit, которых нет в allowed.
• Срабатывание = permissionDecision: "deny", смысл текста: «У XX нет YY. Это сделано намеренно, а не забыто в конфиге. Впиши вывод в тело своего ответа, фреймворк возьмёт его оттуда — не пытайся обойти это другой формулировкой».
• Перехватывает только главный поток текущей сессии, subagent'ы проходят — их инструменты определяются AgentDefinition.tools.
• Если перехватывать нечего, возвращает None (например, для роли вроде worker() с полным набором инструментов), и вызывающая сторона по этому решает, ставить его или нет.

Почему он обязан существовать: allowed_tools — это список без запроса подтверждения, а не исключающий белый список. Два замера: судья с заданной целью выполнил Bash 11 раз; в пробнике за $0.1 агент с allowed_tools=["Read"] спокойно вызывал 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

Главный поток работает руками → отказ с указанием, что делать дальше.

• Висит на PreToolUse, matcher — это и есть tools.
• Главный поток вызывает один из этих четырёх инструментов → deny, и в причине сказано, как поступить дальше: отправить subagent через инструмент Agent, прописав в задании цель и критерии приёмки, и потребовать, чтобы длинный вывод он писал в .flower/artifacts/, а в ответе давал только путь и вывод.
• Subagent'ы пропускаются всегда.

Отличие от whitelist_guard — в формулировке: оба перехватывают один и тот же набор инструментов, но этот говорит «делегируй», что здесь уместнее. Поэтому роли с delegate_only=True получают только его — при дублировании модель получит две противоречащие инструкции.

Критерий пропуска при allow_glance=True и критерий «будет ли результат подрезан» — это одна и та же функция (is_ephemeral): множество пропускаемого обязано совпадать с множеством устаревающего, меняешь одно — меняй и другое.

spill_guard()¶

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

Результаты инструментов сверх порога сразу сбрасываются на диск, в контексте остаётся одна строка-указатель — а не ждём, пока контекст заполнится, чтобы потом его сжимать.

• Висит на PostToolUse, возвращает {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <подрезанное>}}.
• Имя файла сброса — первые 16 знаков sha256 содержимого + .txt, в контексте вместо содержимого остаётся строка-указатель + первые 400 символов.
• updatedToolOutput обязан сохранить структуру вывода исходного инструмента, поэтому заменяются только слишком длинные строковые поля словаря; списки не трогаются вообще (внутри могут быть блоки изображений). Неверная структура будет отвергнута (оригинал останется как есть, без ошибки).
• Чтение самих сброшенных файлов обязано пропускаться — иначе «прочитай его через Read» пустые слова: прочитанный текст снова сбрасывается, бесконечный цикл. Напоролись на практике, модель перепробовала пять формулировок обхода.

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

Записали что-то на верстак — обновляем INDEX.md, и следующий агент с самого начала знает, что это есть.

Висит на 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 по роли, реализуя изоляцию.

Висит на PreToolUse, matcher "Agent". Инъекция происходит только при одновременном выполнении трёх условий: tool_name == "Agent", в tool_input нет ни cwd, ни isolation, и роль, соответствующая subagent_type, помечена через isolated(). Тогда возвращается permissionDecision: "allow" + updatedInput (в котором isolation выставлен в "worktree").

isolation и cwd взаимоисключающи в инструменте Agent — если модель сама указала cwd, это уважается. «Нужна ли изоляция» — свойство роли, а не глобальный переключатель и не решение на каждую выдачу задачи; ролям, которым изоляция не нужна, не добавляется ни байта.

Включили изоляцию — выносите верстак из репозитория. Изолированный агент не может писать в общий checkout, поэтому верстак обязан указывать через 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 (проверено).

Цена: dataclasses.replace() над AgentDefinition теряет эту метку, и изоляция молча отключается.

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]]

Ставит за один раз все 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 дважды, он и установится дважды. Runtime через него объединяет spec.hooks, workbench_hooks(...) и whitelist_guard.

Верстак¶

Исходник: flower/core/workbench.py

Workbench¶

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

Рабочий каталог для сброса на диск: три подкаталога + один индекс. Индекс инъектируется в system prompt, поэтому агент на каждом ходу знает, что у него под рукой.

Формат самоописания скрипта: # desc: 一句话 в пределах первых 8 строк (распознаются также комментарии // и --), с откатом на первый непустой комментарий или первую строку docstring (обрезается до 100 символов).

Три правила, которые инъектирует prompt_block():

1. Скрипты, которые понадобится запустить второй раз, писать в scripts/, первой строкой # desc:.
2. Результаты длиннее 2000 символов писать в artifacts/, в диалог давать только путь и вывод.
3. Ключевые решения писать в notes/, одно решение — один файл.

При external=True prompt_block() дополнительно вставляет фразу «обращайся к нему по абсолютному пути».

Индекс subagent'ам не наследуется. Он идёт через system_prompt.append уровня сессии, а у subagent'а свой system prompt (замерено на $0.2461). Поэтому два пункта — «длинные результаты в artifacts/» и «где находится верстак» — обязан пересказать координатор в задании: это единственный канал, а не избыточность. В WORKER_RULES этого намеренно нет: реальные пути генерирует Workbench, зашитые в текст они будут неверными.

Хранилище сессий¶

Исходник: sqlite.py · trim.py · prune.py

Три уровня наследования: SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore. Runtime всегда использует самый внешний, а политики всех трёх уровней задаются параметрами конструктора.

Каждый уровень отвечает за своё: запись на диск, подрезка по объёму и ценности, отсечение по признаку «это ошибка или нет». И подрезка, и отсечение происходят в момент load() (то есть когда resume скармливает историю модели обратно), исходные записи в SQLite не меняются ни на байт.

SqliteSessionStore¶

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

Реализация протокола SessionStore из SDK, три таблицы entries / meta / summaries. Ключ хранилища — project_key/session_id[/subpath]: транскрипты дочерних агентов различаются по subpath.

Внутренний _next_mtime гарантирует строгую монотонность — list_sessions и summary sidecar используют эти часы совместно, иначе быстрый путь проверки staleness в SDK ошибётся.

TrimPolicy¶

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

Два каталога сброса — не один и тот же. spill_guard пишет в <workbench.root>/spill/ (может быть вне рабочей области); TrimPolicy.spill_dirname пишет в <workspace>/.flower/spill/ (обязан быть внутри рабочей области). Они соответствуют «подрезке на месте» и «подрезке при resume»; разные каталоги — это намеренно, не сводите их в один.

EphemeralPolicy¶

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

Политика устаревания результатов одноразовых команд.

Действует только на результаты инструмента Bash, и команда должна попадать в белый список одноразовых команд. Read сюда не входит — содержимое файла со временем не искажается настолько, чтобы вводить в заблуждение. Устаревшее содержимое на диск не сбрасывается, а просто выбрасывается.

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

Определяет, является ли Bash-команда одноразовой командой. Проверка пропуска в delegate_guard и проверка устаревания при подрезке используют одну и ту же функцию: множество команд, которые координатор может выполнить сам, обязано совпадать с множеством результатов, помечаемых как устаревшие. Пропускать без подрезки — устаревший git status навсегда займёт контекст да ещё и введёт в заблуждение; подрезать без пропуска — координатор ради одного ls отправит 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, system( и print > внутри awk, git branch -D/-d/-m, git * --force/--hard/--prune.

Первая версия рубила все составные команды подряд, и на практике glance перестал работать вообще (все три попытки координатора были перехвачены), поэтому проверку переделали на посегментную.

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 остаются как есть.

Две структурные красные линии: сам блок tool_result обязан остаться, менять можно только content (пропадёт один — получите «Missing Tool Result Block»); записи isCompactSummary трогать нельзя.