API Python¶

Esta página esgota os 62 símbolos públicos do __all__ de topo de flower: assinaturas, parâmetros, valores padrão, semântica, atributos e métodos públicos. Depois de ler, não é preciso abrir o código-fonte para procurar parâmetros.

A organização segue aquilo com que você se importa, não os arquivos de módulo — se quer saber "como impedir que o coordenador faça o trabalho por conta própria", vá para a camada de hooks; se quer saber "como o resultado do passo anterior chega ao próximo", vá para workflow. A terminologia segue estritamente o glossário.

Versão 0.1.0, depende de claude-agent-sdk>=0.2.152. Todas as assinaturas correspondem literalmente ao código-fonte.

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

O que tem nesta página¶

Seis padrões que mordem¶

Estes seis não são detalhes marginais; são os seis acidentes mais comuns. Cada um tem explicação completa na seção correspondente.

Runtime¶

Código-fonte: flower/core/runtime.py

Runtime é o núcleo de execução. Ele detém o workspace, o session store, a workbench, a política de resiliência e a política de handoff; para fora, tem apenas um verbo: run um passo. Retentativa, continuação após interrupção, handoff quando o contexto enche — tudo acontece dentro dessa única chamada.

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,
)

Os parâmetros do construtor são todos keyword-only (* logo no início), e workspace é obrigatório.

O session store é fixo no código: sempre PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...)). Os parâmetros do construtor não oferecem ponto de entrada para trocar de backend — para trocar, construa seu próprio AgentSpec + build_options(session_store=...), ou sobrescreva rt.store depois de construir.

Os dois últimos passos da construção são load_dotenv() e check_credentials(), e este último faz raise RuntimeError em caso de erro. Sem credenciais, a explosão acontece já na construção, não ao chegar em run().

Atributos públicos¶

Constantes de classe: INTERRUPTED = "interrupted-by-human", HANDOFF_DUE = "context-full-handoff", INTERRUPT_NOTE (o trecho anexado após a fala da pessoa ao continuar depois de uma interrupção, explicando que "chamadas de ferramenta em voo retornando interrupted são efeito colateral normal da interrupção, não falha de ambiente").

Métodos públicos¶

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

No início de cada passo, o nível de contexto é zerado (self._ctx, self._warned = 0, False). Depois vem um laço com quatro saídas:

1. Sucesso → sai do laço.
2. Interrupção humana (result.error == INTERRUPTED) → não está sujeita a max_attempts, não espera a rede. Faz resume na mesma sessão levando a fala da pessoa, attempt -= 1 (interrupção não conta como tentativa falha), prompt = fala da pessoa + INTERRUPT_NOTE. Se não obteve session_id, só resta parar.
3. Contexto cheio (result.error == HANDOFF_DUE, ou handoff.enabled com session_id obtido e is_overflow(...) acionado) → também não está sujeito a max_attempts. Primeiro verifica len(result.retired) >= handoff.max_generations; se passou, troca o error por uma linha de diagnóstico e sai do laço; caso contrário escreve o documento de handoff → resume=None, fork=False (sessão totalmente nova) → prompt vira h.prompt_block() → nível zerado → attempt -= 1.
4. Falha retentável → sai se not resilience.enabled or attempt >= max_attempts; sai também se classify(error) decidir que não se deve retentar; caso contrário emite Event("retry"), fica pendurado em wait_online() esperando a rede, sleep(delay_for(attempt)); se algum session_id já foi obtido, continua com resume (o prompt vira resilience.resume_prompt) e define result.resumed como True.

Encerramento: escreve ended_at, anexa a self.results, escreve o manifest.json.

O manifest.json tem semântica de append: cada escrita relê o disco e deduplica pelo campo run (a própria linha é substituída, as dos outros permanecem), então rodar dois flowers em paralelo no mesmo run_dir é seguro — desde que os run_id não colidam.

Os três pontos de observação do handoff (todos Event("handoff"), diferenciados por payload["phase"]): near (aproximando-se de warn_at, emitido uma vez por geração), writing (escrevendo o handoff, leva dezenas de segundos), done (payload traz degraded / path / sections). O turno que escreve o handoff roda com replace(spec, max_budget_usd=None) — o handoff precisa conseguir ser escrito, não pode travar no orçamento; além disso on_event=None, esse turno não atualiza a UI.

O handoff é gravado em <workbench.notes>/交接-<步骤名>.md; sem workbench não há gravação, o documento ainda é entregue a quem assume via prompt, apenas não fica consultável depois. Handoffs antigos são movidos para 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

Toda a contabilidade de um passo concluído.

Workflow¶

Código-fonte: flower/workflow/

Um workflow é um conjunto de steps encadeados em ordem, mais como o estado passa de um step para o outro e quando se sai mais cedo. O framework não fornece workflows prontos; o workflow é você que escreve — starter_flow é apenas um modelo que roda.

Alias de tipo Ctx = dict[str, Any] (flower.workflow.base.Ctx, está em flower.workflow.__all__, não no __all__ de topo).

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

A declaração de um step. Step em si não é uma função — quem executa de fato é Runtime.run(step.spec, prompt, ...). Os três primeiros campos são argumentos posicionais; Step("取词", terse, "读 seed.txt …") é uma escrita válida.

Três formas de conectar sessões (dentro de uma mesma run):

on_reject muda a semântica do retry:

• Não fornecido → a próxima tentativa recomeça do zero (mesmo prompt, mesmo resume_from).
• Fornecido → a próxima tentativa continua a sessão que acabou de ser rejeitada, com o prompt trocado pelo valor de retorno dele, e fork forçado para False.
• Retornar string vazia → não devolve nada; degrada para recomeçar do zero.
• result.session_id sendo None → também degrada para recomeçar do zero.

Os três valores de on_fail:

Passando ou não, ctx["_results"][name] = result sempre é escrito; quando result.session_id não é vazio, também é escrito em ctx["_sessions"] e chamado 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

Roda uma sequência de Step em ordem e retorna o ctx final. steps é argumento posicional, então Workflow([...]) é válido.

As chaves privadas que run() escreve no ctx (todas começam com _, não colidem com nomes de step):

Payload de Event("step"): {"index": i, "total": len(steps), "resumed": bool, "woke": int}.

Rótulos de retry: a tentativa 0 usa step.name; depois disso, com on_reject usa f"{name}#round{attempt+1}", sem ele usa f"{name}#retry{attempt}". No manifest dá para ver de relance como aquele step chegou ao fim. Nomes com sufixo não entram na lineage entre processos — Lineage.remember usa o nome original.

runtime.on_session cobre apenas a linha runtime.run, com try/finally garantindo que seja removido antes do gate. prompt_cur / resume_cur / fork_cur são variáveis locais e não são escritas de volta no step — o mesmo objeto Step pode ser rodado uma segunda vez.

StepAbort¶

class StepAbort(Exception): ...

Lançada pelo gate = pare já, não tente de novo. A diferença para "retornar False": False é "desta vez não deu, mais uma rodada"; StepAbort é "tentar de novo não adianta".

Depois de lançada: ctx["_aborted"] = str(exc), passed = False, sai do laço de retry (sem consumir os retries restantes) e então segue o on_fail como uma falha comum (padrão "stop").

with_goal a lança em dois pontos: quando não consegue obter ctx["_runtime"], e quando o verdict é unreachable e ninguém responde.

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

Produz um Step que faz o clarify: esclarecer a demanda → parsear para Brief → congelar e gravar em disco quando as quatro seções estiverem completas.

No Step produzido, os campos ficam assim:

• resume_prompt = CLARIFY_RESUME.
• when: always_ask=True → sempre True; caso contrário, se Brief.load(brief_path) estiver completo, injeta-o no ctx e então retorna False (pula) — mesmo pulando é preciso injetar, senão quem vem depois não recebe a demanda.
• gate: Brief.parse(result.text); incompleto → escreve ctx[MISSING_KEY] e retorna False; completo → b.write(brief_path) para congelar, injeta no ctx, retorna True.
• reduce: retorna ctx[BRIEF_KEY].prompt_block(), não o texto original do modelo — no texto original pode vir junto coisa que ele escreveu a mais.
• resume_from fica no padrão None: o próximo step é uma sessão nova, que recebe só o brief e não a rodada de perguntas e respostas. As perguntas do clarify nunca entraram no contexto do coordinator; não é que entraram e foram podadas depois.

Os três pontos injetados no 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

Produz um Step que define o objetivo: faz o judge ler o brief, escrever o objetivo + o checklist de verdict, e depois de parsear para Goal congelar e gravar em disco. Tem o mesmo formato de clarify_step.

Não existe o parâmetro can_run — para que o judge que define o objetivo possa rodar comandos, só passando can_run=True via **spec_kw. Sem isso ele não recebe Bash, e a regra "primeiro veja bem em que ambiente você está" de JUDGE_RULES não pode ser executada.

Além de parsear e congelar, o gate faz mais uma coisa: quando há itens [此环境无法验证:…] no objetivo, ele emite na hora, via ctx["_on_event"], um Event("task", payload={"unverifiable", "total", "path"}) como aviso — o destino desses itens já está selado no momento em que o objetivo é definido, e quando chega a hora do verdict o dinheiro de uma rodada inteira de trabalho já foi gasto.

Não define resume_prompt — definir o objetivo deve mesmo reenviar o brief por inteiro.

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

Envolve um Step existente com o goal guard: ao fim de cada rodada o judge emite um verdict independente; se o objetivo não foi alcançado, devolve para continuar trabalhando.

O resultado é replace(step, retries=max(0, rounds - 1), gate=<novo gate>, on_reject=<novo on_reject>) — usa dataclasses.replace em vez de reconstruir campo a campo; uma reconstrução já deixou o resume_prompt escapar, e sem erro nenhum, apenas reenviando o brief inteiro de novo na continuidade.

O gate é async, e o fluxo é:

1. ctx["_runtime"] ausente → lança StepAbort ("拿不到 Runtime,无法判定目标"). Não finja que passou.
2. ctx[ROUND_KEY] += 1.
3. Pega o objetivo: primeiro um Goal completo em ctx[GOAL_KEY], senão Goal.load(goal_path), senão um Goal() vazio.
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...). O judge é um Runtime.run independente, com resume sempre None — é sempre uma sessão nova; o step_name carrega o número da rodada, então não entra na lineage entre processos.
5. Verdict.parse(vr.text) é escrito em ctx[VERDICT_KEY].
6. v.achieved → retorna True.
7. Não sendo unreachable (inclusive os casos ambíguos com v.ok=False) → nos casos ambíguos acrescenta um reason padrão e retorna False. Ambiguidade é sempre tratada como não alcançado — não dá para deixar um "parece que dá" encerrar o trabalho.
8. unreachable → await channel.ask(...) pergunta a uma pessoa, com três opções:
9. Ninguém responde (a.state != "answered") → lança StepAbort. Continuar girando em falso é a escolha mais cara.
10. 「接受这个结果,就这样往下走」 → retorna True.
11. 「修改目标」 → pergunta de novo o novo objetivo, faz g.amend(...).write(goal_path), atualiza ctx[GOAL_KEY] e retorna False.
12. Qualquer outra coisa (inclusive resposta livre digitada pela pessoa) → é tratada como "você julgou errado", registra a fala da pessoa em v.reason e retorna False.

O on_reject é síncrono: retorna ctx[VERDICT_KEY].feedback(); sem Verdict, retorna "" (degrada para recomeçar do zero).

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

Monta um workflow de três steps que roda de imediato: confirmar requisitos → definir objetivo → trabalhar (com goal guard). É o que a linha de comando flower usa.

Montagem fixa:

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)

Ramificações de comportamento:

• isolate=True com o workspace não sendo um repositório git → lança ValueError, em vez de só descobrir quando a ferramenta Agent reclama (aí o dinheiro já foi gasto).
• Detecção de wake: se Brief.load(brief_path) existe e está complete(), é um wake. Se não é wake e ask está vazio → lança ValueError("要给一句诉求,例如 flower '帮我做一个 X'").
• No wake, essa frase cai em três lugares ao mesmo tempo; faltar um deles causa falha silenciosa: anexada ao brief (ch.amend(said, label="唤醒时追加"), sem reescrever se já estiver no arquivo); fazendo goal_step(always_set=True) refazer o checklist (sem isso, o judge continua lendo o objetivo antigo); e entregue direto ao coordinator (no contexto dele está o objetivo antigo, e sem isso ele trabalha pelo critério antigo e é julgado pelo critério novo).

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

Uma sondagem somente-leitura antes da largada, que não escreve um único byte. Serve para dizer à pessoa, antes de começar de verdade, "isto é continuação da vez passada ou começo do zero".

O dict retornado:

A posição do workbench é definida uma única vez, aqui e em starter_flow: isolate=True → <ws>.parent/.flower-<ws.name> (fora do repositório); caso contrário <ws>/.flower. Um driver que queira saber onde está o brief também passa por esta função — montar o caminho por conta própria e errar não dá erro, apenas falha em silêncio.

Fábrica de papéis¶

Código-fonte: flower/core/roles.py

Os cinco papéis são funções de fábrica. Cada papel = um trecho de texto de regras injetado + um conjunto de ferramentas + um conjunto de hooks. worker() produz o AgentDefinition do SDK (para ser despachado a um subagent); os outros quatro produzem AgentSpec (abrem a própria sessão).

Os papéis em si não montam hooks — interceptar ferramentas é trabalho que Runtime._attempt instala automaticamente conforme spec.delegate_only, veja camada de hooks.

Constantes internas de grupos de ferramentas (não exportadas, mas determinam os padrões):

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

Constrói o coordenador que fica na thread principal: decompõe a tarefa, despacha trabalho, lê relatórios, decide, mas não põe a mão na massa. Os três primeiros parâmetros são posicionais.

Três itens fixos no AgentSpec produzido: delegate_only=True, agents=workers, e workbench mantém o padrão True do AgentSpec.

As ferramentas web dos workers são mescladas aqui¶

Depois de montar a lista, coordinator() percorre cada AgentDefinition.tools; toda ferramenta que caia em WEB_TOOLS (WebFetch, WebSearch, roles.py:33) também é adicionada ao próprio allowed_tools do coordenador (roles.py:523-526).

Motivo: allowed_tools, assim como disallowed_tools, é de escopo de sessão. Esta é a evidência mais dura do texto inteiro sobre esse ponto — ela não afeta só a thread principal. Uma ferramenta que não esteja nessa lista de escopo de sessão também exige aprovação de permissão quando um subagent a chama; sem supervisão não há quem aprove, e o harness responde Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected), enquanto o modelo repete a mesma chamada indefinidamente. Já tropeçamos nisso na prática: acrescentamos WebFetch/WebSearch ao worker, mas só escrevemos em AgentDefinition.tools; aquela run de novel teve mais de vinte user-rejected e não escreveu uma única palavra (roles.py:513-518).

Os dois campos têm a mesma natureza de escopo de sessão, mas sintomas diferentes: disallowed_tools dá erro na hora, allowed_tools fica repetindo em silêncio até morrer. O segundo é mais difícil de diagnosticar, porque nada na tela parece um erro.

Só as somente-leitura, sem efeitos colaterais, são mescladas. Write/Edit/Bash são deliberadamente deixadas de fora: se a thread principal ficar isenta de aprovação para elas, o muro do delegate_guard — "o coordenador não põe a mão na massa" — perde o sentido; e Bash/Write do subagent já funcionam de qualquer maneira (462 liberações medidas, roles.py:520-522).

O código-fonte diz explicitamente para não usar disallowed_tools para implementar "só coordena, não executa" — aquilo é de escopo de sessão, e desabilitaria junto o Bash/Write dos subagents; veja o aviso em AgentSpec. O jeito correto é o daqui: delegate_only=True + não conceder em allowed_tools, e deixar o delegate_guard interceptar apenas a thread principal por agent_id.

Fornecer channel traz as duas ferramentas juntas, não é opcional: montado o MCP server, as duas estão lá, e como allowed_tools não é exclusivo, ambas podem ser chamadas estejam listadas ou não. Sem supervisão, cada ask trava até estourar o timeout_s — nesses cenários, use 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

Constrói a definição do subagent que de fato trabalha. Os dois primeiros parâmetros são posicionais. Retorna o AgentDefinition do SDK, para ser colocado diretamente em coordinator(workers={...}).

isolate=True exige que o workspace seja um repositório git; caso contrário, a ferramenta Agent reporta direto "not in a git repository", não há degradação silenciosa. E como a marca é um atributo Python, fazer dataclasses.replace() sobre o AgentDefinition a perde, e o isolamento deixa de valer silenciosamente.

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

Constrói o clarificador: esclarece o requisito antes de agir, não faz nada além de perguntar, e no fim produz exatamente quatro seções.

O AgentSpec produzido: allowed_tools = [channel.tool_name] + (os cinco acima, quando pode ler), mcp_servers = channel.mcp_servers(), workbench=False (ele não tem ferramenta de escrita, o índice não lhe serve de nada), permission_mode herda o padrão "default" do AgentSpec. Sem Write / Edit / Bash / Agent, e sem inbox (diferente do coordenador).

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

Constrói o juiz: ou define o objetivo antes de começar, ou emite o veredito ao fim de cada rodada.

O AgentSpec produzido: allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (quando can_run) ["Bash"], workbench=False, o resto igual a clarify(). Sem Write / Edit / Agent, e sem inbox.

Trade-off: can_run=True dá um veredito mais duro (consegue de fato rodar os comandos de aceitação), ao custo de o juiz passar a poder alterar o workspace — Bash por si só já escreve arquivos. Se você quer um veredito absolutamente neutro, não ligue.

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

Constrói o oracle: com a run ainda em andamento, você pergunta "onde estamos agora" e ele dá uma olhada nos eventos recentes e na bancada antes de responder. O que ele diz não entra no contexto daquela run.

O AgentSpec produzido: allowed_tools = ["Read", "Glob", "Grep"] (sem channel — ele não pergunta, só responde), workbench=True (o único dos cinco papéis que não é coordenador e ainda assim abre a bancada — é exatamente lá que ele vai ler os artefatos e as notas).

Os cinco textos de regras¶

As cinco constantes estão em __all__; dá para import direto e ler, concatenar, alterar.

Definição de agent¶

Código-fonte: flower/core/agent.py

AgentSpec é a declaração completa de um agent especializado, e build_options a compila no ClaudeAgentOptions do SDK. O que a fábrica de papéis produz é justamente um AgentSpec — quando você precisa de uma combinação fora da fábrica, construa-o diretamente.

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

Compila o AgentSpec no ClaudeAgentOptions do SDK. É o que Runtime._attempt chama internamente; quando você mesmo dirige o SDK (sem Runtime), a entrada também é por aqui.

Mapeamento:

PLUGIN_DIR é o plugin/ na raiz do repositório (três níveis acima de flower/core/agent.py). Após instalar por pip esse diretório pode não existir; o código verifica com is_dir().

CompactPolicy¶

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

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

O painel de chaves do auto-compact; o produto é um conjunto de variáveis de ambiente a injetar no subprocesso. O algoritmo de compact em si está no binário do harness e não pode ser mudado; o que dá para mudar é apenas "dispara ou não".

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)

O objeto de política que, quando o contexto está quase cheio, "escreve o documento de handoff e troca de sessão" em vez de fazer compact.

default_window()¶

def default_window() -> int

Adivinha a janela de contexto a partir da string do nome do modelo nas variáveis de ambiente ANTHROPIC_MODEL ou ANTHROPIC_DEFAULT_OPUS_MODEL:

O padrão é o valor agressivo. Estimar alto não é um erro fatal: a API rejeita com prompt is too long, o Runtime reconhece esse sinal (o is_overflow interno) e faz handoff na hora — mas o handoff dessa geração sai em versão degradada.

Documentos¶

Código-fonte: brief.py · handoff.py · goal.py

Quatro dataclasses, todas "parseiam uma resposta do modelo em um número fixo de seções e gravam em disco". Formato comum: parse() parseia, missing() / complete() verificam completude, to_markdown() é para humanos, prompt_block() é para os modelos a jusante, write() / load() gravam e leem de volta.

Brief¶

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

O brief, exatamente quatro seções, na ordem fixa goal → accept → bounds → unknowns; os nomes em chinês são «目标», «验收标准», «边界», «未知与假设».

Regras de parsing (onde os erros se concentram):

• Ao remover as cercas, **um `` ou_~não fechado faz descartar tudo a partir dele** — na prática o clarificador cola o código inteiro na resposta. Quando a saída do modelo é truncada, todas as seções seguintes deixam de ser parseadas,complete()ficaFalse` e o gate manda refazer.
• A regex de título tolera ## 目标 / **目标** / 目标: / 3. 边界, e também tolera texto logo após o título.
• A tabela de aliases é compilada em ordem decrescente de comprimento; senão "未知" engoliria "未知与假设" primeiro.
• Quando a mesma seção aparece repetida, prevalece a primeira com conteúdo.
• Se, ao editar o brief à mão, você copiar o placeholder "(未填)" de to_markdown(), aquela seção continua contando como faltante.

Handoff¶

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

O documento de handoff escrito na hora do handoff, cinco seções.

Obrigatórias são só doing e next — exigir rigidamente que "caminhos que não deram certo" seja não vazio forçaria o modelo a inventar.

Três membros do mesmo módulo não exportados, mas semanticamente cruciais: is_overflow(*texts) casa com prompt is too long, context length exceeded, maximum context length, too many total text bytes, input length and max_tokens exceed etc., convertendo um "erro fatal" em "handoff imediato"; HANDOFF_PROMPT é o prompt que faz a própria sessão atual escrever o handoff (contém os dois placeholders {used} e {window}; não é um novo papel — só ela tem aquele contexto); degraded(step, prompt, *, why="") monta mecanicamente um handoff quando ele não pôde ser escrito, enfiando em scene os primeiros 1200 caracteres da tarefa original.

Goal¶

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

O objetivo + a lista de verificação usados pelo goal guard.

Verdict¶

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

O resultado de uma rodada de julgamento do juiz, exatamente três seções: conclusão / motivo / não aprovados.

Ordem de reconhecimento do parse:

1. Primeiro busca as seções tituladas «结论» / «判定».
2. Sem seções tituladas, faz strip do texto inteiro e fullmatch(r"1|true") → atingido; fullmatch(r"0|false") → ainda não.
3. Caso contrário, procura no texto da conclusão a primeira ocorrência de uma palavra da tabela de estados (palavras longas primeiro). «无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable» são todas mapeadas para unreachable — já tropeçamos nisso: plataforma-alvo macOS, execução em container Linux, e o juiz olhou o branch no código-fonte e deu aprovado.
4. Ainda nada → procura um \b1\b isolado → atingido, \b0\b → ainda não.
5. Nenhum caso → state="", ok=False.

unreachable e not_yet são conclusões diferentes: a primeira leva ao caminho de "parar e perguntar a alguém", não ao de "mais uma rodada".

Camada de hooks¶

Código-fonte: flower/core/guard.py

Esta camada é a fronteira de execução do flower: quais ferramentas a thread principal não pode tocar, como recortar resultados gigantes, qual papel vai para um worktree independente — tudo é imposto por hooks do SDK, não por prompt. O motivo é direto: prompt é sugestão, o modelo pode ignorar. Já foi medido na prática que, mesmo com o system prompt dizendo explicitamente "não use worktree", a injeção do isolate_guard funcionou do mesmo jeito (o modelo passou None, o que chegou no disco foi 'worktree').

Nove exports: cinco fábricas de guard que retornam HookMatcher (whitelist_guard pode retornar None), um montador, um combinador, duas funções de marcação de isolamento. Não é preciso instalá-los à mão — o Runtime monta tudo automaticamente a partir do AgentSpec. Instalação manual só é necessária quando você mesmo dirige o SDK (sem passar pelo Runtime).

A detecção da thread principal passa por uma única função: _is_main_thread(data) = not data.get("agent_id") — os dados de tool-lifecycle hook de um subagent trazem agent_id, os da thread principal não. Todos os guards que "só barram a thread principal" se apoiam nessa única regra.

Constantes de grupos de ferramentas (nível de módulo, não exportadas, mas definem os matchers padrão):

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

Tabela de referência: qual guard entra em qual evento do SDK¶

A conclusão importante que se lê nessa tabela: com Runtime(workbench=False), os workbench_hooks inteiros não são instalados; e para um coordenador com delegate_only=True, o whitelist_guard também é pulado — a thread principal fica sem nenhuma barreira. Veja o aviso em Runtime.

whitelist_guard()¶

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

Faz allowed_tools ser de fato exclusivo para as quatro ferramentas de ação.

• Entra em PreToolUse, com matcher "|".join(banned), onde banned = as ferramentas entre Bash Write Edit NotebookEdit que não estão em allowed.
• Ao acertar, permissionDecision: "deny", com texto no espírito de: "XX não tem YY. Isso é intencional, não é configuração faltando. Escreva a conclusão no corpo da sua resposta, o framework a lê de lá — não tente outra formulação para contornar."
• Só barra a thread principal desta sessão, subagents passam — as ferramentas de um subagent são decididas por AgentDefinition.tools.
• Quando não há nada a barrar, retorna None (por exemplo, um papel com o kit completo como worker()); quem chama decide se instala ou não.

Por que ele precisa existir: allowed_tools é uma lista de dispensa de aprovação, não uma whitelist exclusiva. Duas evidências medidas na prática: um judge com meta definida rodou Bash 11 vezes; na sonda de $0.1, um agent com allowed_tools=["Read"] chamou Write/Bash sem problema. Ou seja, o "não tem ferramenta de escrita" de clarify() / judge() vem deste hook, não da whitelist em si.

A vantagem é que ele deriva de allowed_tools, então judge(can_run=True) mantém Bash automaticamente e continua barrando Write/Edit — sem precisar de flag extra.

delegate_guard()¶

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

Thread principal pondo a mão → recusa, com indicação do caminho.

• Entra em PreToolUse, matcher é o próprio tools.
• Thread principal chamando uma dessas quatro ferramentas → deny, e o motivo diz qual é o próximo passo: usar a ferramenta Agent para despachar um subagent, escrever no encargo o objetivo e os critérios de aceitação, e exigir que ele escreva produções longas em .flower/artifacts/ e devolva só caminho e conclusão.
• Subagents sempre passam.

A diferença para o whitelist_guard é a formulação: ambos barram o mesmo conjunto de ferramentas, mas este diz "vá delegar", que é o mais adequado ali. Por isso um papel com delegate_only=True instala só este; instalar os dois faria o modelo receber duas orientações contraditórias.

O critério de liberação de allow_glance=True e o critério de "o resultado vai ser recortado" são a mesma função (is_ephemeral) — o conjunto liberado tem que ser igual ao conjunto que expira; mexeu em um lado, tem que mexer no outro.

spill_guard()¶

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

Resultados de ferramenta acima do limiar vão para o disco na hora (spill), deixando no contexto só uma linha de ponteiro — não é esperar o contexto encher para depois compactar.

• Entra em PostToolUse, retorna {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <recortado>}}.
• O nome do arquivo é os 16 primeiros caracteres do sha256 do conteúdo + .txt; no contexto entra uma linha de ponteiro + os 400 primeiros caracteres.
• O updatedToolOutput precisa preservar a estrutura de saída da ferramenta original, então só os campos string longos demais do dict são substituídos; listas nunca são tocadas (podem conter blocos de imagem). Estrutura errada é rejeitada (fica o original, sem erro).
• Ler o próprio arquivo de spill tem que ser liberado — senão "use Read para lê-lo" é conversa fiada: o texto lido de volta sofre spill de novo, em loop infinito. Já aconteceu na prática; o modelo tentou cinco formulações seguidas para contornar.

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

Escreveu algo no workbench, o INDEX.md é atualizado, e o próximo agent já começa sabendo que aquilo existe.

Entra em PostToolUse, matcher "Write|Edit". Se tool_input["file_path"], depois de resolvido, cai dentro de workbench.root, chama workbench.refresh(). Sempre retorna {} — não altera nada, só tem efeito colateral.

isolate_guard()¶

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

Dá a cada subagent, conforme o papel, um git worktree independente — é assim que o isolamento acontece.

Entra em PreToolUse, matcher "Agent". A injeção só ocorre com as três condições simultâneas: tool_name == "Agent", tool_input sem cwd e sem isolation, e o papel correspondente ao subagent_type marcado por isolated(). Nesse caso retorna permissionDecision: "allow" + updatedInput (define isolation como "worktree").

isolation e cwd são mutuamente exclusivos na ferramenta Agent — se o modelo especificou um cwd, isso é respeitado. "Isolar ou não" é uma propriedade do papel, não um botão global nem uma decisão tomada a cada despacho; papéis que não precisam de isolamento não ganham um byte a mais.

Ligar isolamento exige tirar o workbench de dentro do repositório. Um agent isolado não consegue escrever no checkout compartilhado, então o workbench precisa apontar para fora do repositório via home=. starter_flow(isolate=True) usa <ws>.parent/.flower-<ws.name>, Runtime(workbench=True) usa <run_dir>/workbench — os dois ficam fora do repositório, mas não são o mesmo diretório, não misture.

isolated() / wants_isolation()¶

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

Marca uma definição de subagent como "precisa de workspace próprio", e lê essa marca de volta.

A marca é um atributo _flower_isolate do lado Python, gravado com object.__setattr__, não é um campo de dataclass — o SDK serializa com asdict() e só reconhece campos declarados, então essa marca não vaza para o lado da CLI (medido na prática).

Custo: fazer dataclasses.replace() sobre um AgentDefinition perde essa marca, e o isolamento falha silenciosamente.

Por dentro, worker(isolate=True) passa por 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]]

Instala de uma vez os hooks de que o workbench precisa. É o que Runtime._attempt chama.

Saída:

• PreToolUse: delegate_only=True → [delegate_guard(allow_glance=allow_glance)]; havendo papel marcado → acrescenta isolate_guard(agents).
• PostToolUse: sempre [index_guard(workbench)]; spill_threshold verdadeiro → acrescenta spill_guard(workbench, threshold=spill_threshold).
• Chaves de evento com lista vazia são removidas, não se retorna lista vazia.

merge_hooks()¶

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

Concatena vários grupos de configuração de hook por nome de evento.

Usa extend, sem deduplicação — passar o mesmo guard duas vezes o instala duas vezes. O Runtime o usa para juntar spec.hooks, workbench_hooks(...) e whitelist_guard.

Workbench¶

Código-fonte: flower/core/workbench.py

Workbench¶

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

O diretório de trabalho em disco: três subdiretórios + um índice. O índice é injetado no system prompt, então o agent sabe a cada turno o que tem em mãos.

Formato de autodescrição de script: # desc: uma frase dentro das 8 primeiras linhas (também reconhece os comentadores // e --), com fallback para o primeiro comentário não vazio ou a primeira linha da docstring (cortada em 100 caracteres).

As três regras que prompt_block() injeta:

1. Scripts que serão rodados uma segunda vez vão em scripts/, com # desc: na primeira linha.
2. Produções acima de 2000 caracteres vão em artifacts/; na conversa, só caminho e conclusão.
3. Decisões-chave vão em notes/, um arquivo por decisão.

Quando external=True, prompt_block() acrescenta uma frase dizendo "acesse por caminho absoluto".

O índice não é herdado pelos subagents. Ele vai pelo system_prompt.append no nível da sessão, e subagents têm system prompt próprio (medido: $0.2461). Por isso as duas informações — "produção longa vai em artifacts/" e "onde fica o workbench" — precisam ser repassadas pelo coordenador dentro do task brief: esse é o único canal, não é redundância. Em WORKER_RULES isso foi deliberadamente omitido: o caminho real é gerado pelo Workbench, escrevê-lo fixo daria errado.

Session store¶

Código-fonte: sqlite.py · trim.py · prune.py

Três níveis de herança: SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore. O Runtime usa sempre o mais externo; as políticas dos três níveis são controladas por parâmetros de construção.

Cada nível cuida de uma coisa: persistir, fazer trim por volume e valor, e fazer prune pelo critério "isso é um erro?". Tanto o trim quanto o prune acontecem no momento do load() (ou seja, quando o resume devolve o histórico ao modelo); os registros originais no SQLite não mudam um byte.

SqliteSessionStore¶

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

Implementa o protocolo SessionStore do SDK, com três tabelas: entries / meta / summaries. A chave do store é project_key/session_id[/subpath] — o transcript de um subagent se distingue pelo subpath.

O _next_mtime interno garante monotonicidade estrita — list_sessions e o sidecar de summary compartilham esse relógio; sem isso, o caminho rápido de staleness do SDK erra o julgamento.

TrimPolicy¶

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

Os dois diretórios de spill não são o mesmo. O spill_guard grava em <workbench.root>/spill/ (pode estar fora do workspace); o TrimPolicy.spill_dirname grava em <workspace>/.flower/spill/ (tem que estar dentro do workspace). Um corresponde ao "corte na hora" e o outro ao "corte no resume"; os diretórios serem diferentes é intencional, não unifique.

EphemeralPolicy¶

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

Política de expiração dos resultados de comandos efêmeros.

Só age sobre resultados da ferramenta Bash, e o comando precisa bater com a whitelist de comandos efêmeros. Read não entra nisso — o conteúdo de um arquivo não se distorce com o tempo a ponto de enganar. Conteúdo expirado não sofre spill, é descartado direto.

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

Decide se um comando Bash é um comando efêmero. A liberação do delegate_guard e a checagem de expiração do trim usam esta mesma função — o conjunto de comandos que o coordenador pode rodar sozinho tem que ser igual ao conjunto de resultados que serão marcados como expirados. Liberar sem recortar faz um git status expirado ocupar contexto para sempre e ainda enganar; recortar sem liberar faz o coordenador despachar um subagent para um ls, trocando 4.3k de custo de arranque por algumas dezenas de caracteres.

Ordem de decisão:

1. Vazio / só espaços → False.
2. Substituição de comando ($(, crase, <(, >() ou alguma forma que "muda estado" → False.
3. Depois de remover redirecionamentos seguros (2>&1, &> /dev/null e afins), ainda contém > ou < → False.
4. Depois de remover && / || / ; / |, ainda sobra um & isolado (execução em background) → False.
5. Quebra por && / || / ; / |, e cada segmento tem que bater com a whitelist.

Grandes categorias de verbos na whitelist: subcomandos git somente-leitura (status diff log show branch rev-parse etc.), informação de diretório e sistema (ls pwd df du date whoami env etc.), processos e containers (ps top lsof docker ps kubectl get etc.), leitura de arquivos (cat head tail wc stat find tree), busca de caminhos (which whereis command -v type), processamento de texto (grep rg sort uniq awk sed jq diff etc.).

Mesmo com o verbo na whitelist, estas formas são barradas: xargs, exec, eval, source, tee, find -delete / -ok / -fprint, sed -i, sort -o, system( e print > dentro de awk, git branch -D/-d/-m, git * --force/--hard/--prune.

A primeira versão recusava todo comando composto de forma indiscriminada, e na prática isso matou o glance por completo (as três tentativas do coordenador foram todas barradas), então passou a avaliar segmento por segmento.

TrimmingSessionStore¶

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

Atributos públicos: workspace, policy, ephemeral, last_report: dict[str, int].

Ordem no load(): super().load() → limpa last_report → se ephemeral.enabled, expire() → se policy.enabled, trim(). Com enabled=False, o passo inteiro é pulado.

Só recorta texto puro: blocos image / document são deixados como estão.

Duas linhas vermelhas estruturais: o bloco tool_result tem que continuar existindo, só se pode trocar o content (faltar um já é "Missing Tool Result Block"); entradas isCompactSummary não podem ser tocadas.

trim_report()¶