Python API¶

This page exhaustively covers the 62 public symbols in flower's top-level __all__: signatures, parameters, defaults, semantics, public attributes and methods. After reading it you should never need to open the source to look up a parameter.

The organization follows what you care about, not module files — if you want to know "how do I stop the coordinator from doing the work itself", go to the hook layer; if you want to know "how does one step's result reach the next", go to workflow. Terminology follows the glossary throughout.

Version 0.1.0, requires claude-agent-sdk>=0.2.152. Every signature corresponds verbatim to the source.

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

What's on this page¶

Six defaults that will bite you¶

These six are not trivia; they are the six most common ways to crash. Each has a full explanation in its own section.

Runtime¶

Source: flower/core/runtime.py

Runtime is the execution core. It holds the workspace, the session store, the workbench, the resilience policy and the handoff policy, and exposes exactly one verb: run a step. Retries, resuming after an interrupt, and handing off when context fills up all happen inside that single call.

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

Constructor parameters are all keyword-only (* comes first), and workspace is required.

The session store is hard-wired: always PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...)). The constructor offers no hook for swapping the backend — to swap it, build your own AgentSpec + build_options(session_store=...), or overwrite rt.store after construction.

The last two steps of construction are load_dotenv() and check_credentials(), and the latter raises RuntimeError on failure. With no credentials you blow up at construction time, not at run().

Public attributes¶

Class constants: INTERRUPTED = "interrupted-by-human", HANDOFF_DUE = "context-full-handoff", INTERRUPT_NOTE (a passage appended after the human's words when resuming from an interrupt, explaining that "a tool call in flight returning interrupted is a normal side effect of the interruption, not an environment failure").

Public methods¶

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

Each step starts by zeroing the context watermark (self._ctx, self._warned = 0, False). What follows is a loop with four exits:

1. Success → break out.
2. Human interrupt (result.error == INTERRUPTED) → not bound by max_attempts, does not wait for the network. resume the same session carrying the human's words, attempt -= 1 (an interrupt is not a failed attempt), prompt = the human's words + INTERRUPT_NOTE. Without a session_id there is nothing to do but stop.
3. Context full (result.error == HANDOFF_DUE, or handoff.enabled and a session_id was obtained and is_overflow(...) fires) → also not bound by max_attempts. First check len(result.retired) >= handoff.max_generations; if exceeded, replace the error with a diagnostic line and break out; otherwise write the handoff document → resume=None, fork=False (a brand new session) → prompt becomes h.prompt_block() → zero the watermark → attempt -= 1.
4. Retryable failure → break out if not resilience.enabled or attempt >= max_attempts; break out if classify(error) says it should not be retried; otherwise emit Event("retry"), hang on wait_online() waiting for the network, sleep(delay_for(attempt)); if a session_id was ever obtained, resume it (prompt becomes resilience.resume_prompt), and set result.resumed to True.

Wrap-up: write ended_at, append to self.results, write manifest.json.

manifest.json has append semantics: every write re-reads the disk and deduplicates on the run field (your own rows are replaced, other runs' rows are kept), so running two flowers in parallel under the same run_dir is safe — provided the run_ids do not collide.

The three observation points of a handoff (all Event("handoff"), distinguished by payload["phase"]): near (approaching warn_at, emitted once per generation), writing (the handoff is being written, takes a dozen-odd seconds), done (payload carries degraded / path / sections). The turn that writes the handoff runs with replace(spec, max_budget_usd=None) — the handoff must be writable, it cannot get stuck on budget; and on_event=None, so that turn does not stream to the UI.

The handoff spills to <workbench.notes>/交接-<步骤名>.md; with no workbench there is no spill, the document still reaches the successor through the prompt, you just cannot look it up afterwards. Old handoffs are moved to 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

The complete books for one finished step.

Workflow¶

Source: flower/workflow/

A workflow is a set of steps strung together in order, plus how state flows between steps and when to bail out early. The framework ships no ready-made workflow — you write the workflow yourself; starter_flow is just a working sample.

Type alias Ctx = dict[str, Any] (flower.workflow.base.Ctx, listed in flower.workflow.__all__, not in the top-level __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

The declaration of a step. Step itself is not a function — what actually executes is Runtime.run(step.spec, prompt, ...). The first three fields are positional, so Step("取词", terse, "读 seed.txt …") is legal.

Three ways to wire sessions (within one run):

on_reject changes retry semantics:

• Omitted → the next attempt starts over (same prompt, same resume_from).
• Provided → the next attempt resumes the session that was just rejected, with the prompt replaced by its return value, and fork forced to False.
• Returning an empty string → no push-back, degrades to starting over.
• result.session_id is None → also degrades to starting over.

The three values of on_fail:

Pass or fail, ctx["_results"][name] = result is always written; when result.session_id is non-empty it is also written into ctx["_sessions"] and lineage.remember(...) is called.

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

Runs a list of Steps in order and returns the final ctx. steps is positional, so Workflow([...]) is legal.

The private keys run() writes into ctx (all prefixed with _, so they never collide with step names):

The payload of Event("step"): {"index": i, "total": len(steps), "resumed": bool, "woke": int}.

Retry labels: attempt 0 uses step.name; afterwards, with on_reject it is f"{name}#round{attempt+1}", without it f"{name}#retry{attempt}". The manifest then shows at a glance how the step got through. Suffixed names do not enter the cross-process lineage — Lineage.remember uses the original name.

runtime.on_session covers only the runtime.run call, with try/finally guaranteeing it is detached before the gate. prompt_cur / resume_cur / fork_cur are local variables and are not written back to step — the same Step object may be run a second time.

StepAbort¶

class StepAbort(Exception): ...

Raised by a gate = stop now, do not retry. The difference from returning False: False means "not this time, go another round"; StepAbort means "another round will not help".

After it is raised: ctx["_aborted"] = str(exc), passed = False, break out of the retry loop (the remaining retries are not consumed), then follow on_fail as an ordinary failure ("stop" by default).

with_goal raises it in two places: when it cannot get ctx["_runtime"], and when the verdict is unreachable and nobody answers.

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

Produces a Step that performs clarify: ask until the requirement is clear → parse into a Brief → freeze and spill to disk once all four sections are present.

Here is how the fields of the produced Step are filled in:

• resume_prompt = CLARIFY_RESUME.
• when: always_ask=True → always True; otherwise, if Brief.load(brief_path) is complete it is poured into ctx and then False (skip) is returned — it must be poured even when skipping, otherwise downstream gets no requirement.
• gate: Brief.parse(result.text); incomplete → write ctx[MISSING_KEY] and return False; complete → freeze with b.write(brief_path), pour into ctx, return True.
• reduce: returns ctx[BRIEF_KEY].prompt_block(), not the model's raw text — the raw text may contain extra material it wrote.
• resume_from keeps the default None: the next step is a new session that gets only the brief, not the Q&A. The clarify Q&A never entered the coordinator's context; it was not trimmed away after entering.

The three places poured into 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

Produces a goal-setting Step: have the judge read the brief, write the goal plus a verdict checklist, parse it into a Goal, then freeze and spill it. Same shape as clarify_step.

There is no can_run parameter — to let the goal-setting judge run commands, pass can_run=True through **spec_kw. Without it, it does not get Bash, and the "first look carefully at what environment you are in" rule in JUDGE_RULES cannot be carried out.

Beyond parsing and freezing, the gate does one extra thing: when the goal contains [此环境无法验证:…] entries, it emits an Event("task", payload={"unverifiable", "total", "path"}) on the spot through ctx["_on_event"] as a warning — the fate of those entries is decided at goal-setting time, and by verdict time you have already paid for a whole round of work.

No resume_prompt is set — goal setting should resend the full brief anyway.

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

Wraps an existing Step in a goal guard: after each round the judge rules independently, and if the goal is not met the work is pushed back to continue.

The output is replace(step, retries=max(0, rounds - 1), gate=<new gate>, on_reject=<new on_reject>) — using dataclasses.replace rather than rebuilding field by field, because one rebuild once dropped resume_prompt, without raising anything; it just resent the entire brief on every resume.

The gate is async, and goes:

1. ctx["_runtime"] missing → raise StepAbort ("拿不到 Runtime,无法判定目标"). Do not fake a pass.
2. ctx[ROUND_KEY] += 1.
3. Get the goal: prefer a complete Goal in ctx[GOAL_KEY], else Goal.load(goal_path), else an empty Goal().
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...). The judge is an independent Runtime.run, with resume always None — always a new session; step_name carries the round number, so it does not enter the cross-process lineage.
5. Verdict.parse(vr.text) is written into ctx[VERDICT_KEY].
6. v.achieved → return True.
7. Not unreachable (including the ambiguous case where v.ok=False) → fill in a default reason when ambiguous, return False. Ambiguity always counts as not achieved — a "looks fine" must not be allowed to close out the work.
8. unreachable → await channel.ask(...) to ask the human, with three options:
9. Nobody answers (a.state != "answered") → raise StepAbort. Spinning on is the most expensive choice.
10. "Accept this result and move on" → return True.
11. "Amend the goal" → ask once more for the new goal, g.amend(...).write(goal_path), update ctx[GOAL_KEY], return False.
12. Anything else (including a free-form answer typed by the human) → treat it as "you ruled wrong", record what the human said into v.reason, return False.

on_reject is synchronous: it returns ctx[VERDICT_KEY].feedback(), or "" when there is no Verdict (degrading to starting over).

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

Assembles a three-step workflow that runs out of the box: clarify → set goal → do the work (with a goal guard). This is what the flower command line uses.

Fixed assembly:

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)

Behavioural branches:

• isolate=True and the workspace is not a git repository → raise ValueError, rather than finding out when the Agent tool errors out (by which point the money is spent).
• Wake detection: it counts as a wake when Brief.load(brief_path) exists and is complete(). Not a wake and ask is empty → raise ValueError("要给一句诉求,例如 flower '帮我做一个 X'").
• On a wake that line lands in three places at once, and missing any one of them fails silently: appended into the brief (ch.amend(said, label="唤醒时追加"), not written twice if already in the file); goal_step(always_set=True) re-derives the checklist (without re-deriving, the judge still reads the old goal); and it is delivered straight to the coordinator (whose context holds the old goal — without this it would work to the old standard and then be judged by the new one).

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

A read-only probe before the run starts; it writes not one byte. Used to tell the human, before anything really runs, whether this continues last time or starts from scratch.

The returned dict:

The workbench location is defined exactly once, here and in starter_flow: isolate=True → <ws>.parent/.flower-<ws.name> (outside the repository); otherwise <ws>/.flower. A driver program that wants to know where the brief is goes through this function too — assembling the path yourself raises nothing when you get it wrong, it just fails silently.

Role factory¶

Source: flower/core/roles.py

All five roles are factory functions. Each role = a block of injected rule text + a set of tools + a set of hooks. worker() produces an SDK AgentDefinition (to be handed to a subagent); the other four produce an AgentSpec (they start a session of their own).

The roles themselves attach no hooks — intercepting tools is done by Runtime._attempt, which wires them up automatically according to spec.delegate_only; see the hook layer.

Internal tool-group constants (not exported, but they determine the defaults):

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

Builds the coordinator that sits on the main thread: breaks down the task, delegates, reads reports, makes decisions, but never does the work itself. The first three parameters are positional.

Three fields are fixed in the resulting AgentSpec: delegate_only=True, agents=workers, and workbench keeps AgentSpec's default of True.

The workers' web tools get merged in¶

After building the list, coordinator() walks every AgentDefinition.tools; anything that falls in WEB_TOOLS (WebFetch, WebSearch, roles.py:33) also gets added to the coordinator's own allowed_tools (roles.py:523-526).

Reason: allowed_tools, like disallowed_tools, is session-scoped. This is the hardest piece of evidence in the whole document on that point — it does not only affect the main thread. A tool not on this session-level list also requires permission approval when a subagent calls it; unattended, nobody approves, and the harness answers Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected), while the model retries the same call over and over. This has bitten us in practice: WebFetch/WebSearch were added to the worker but written only into AgentDefinition.tools, and that novel run produced twenty-odd user-rejected events and not a single word of text (roles.py:513-518).

The two fields have the same session scope but different symptoms: disallowed_tools errors out on the spot, allowed_tools retries silently until death. The latter is harder to diagnose, because nothing on screen looks like an error.

Only the read-only, side-effect-free ones get merged. Write/Edit/Bash are deliberately not merged: the moment the main thread is approval-free for them, delegate_guard's "the coordinator does not do the work" wall becomes meaningless; and a subagent's Bash/Write already goes through fine anyway (measured: 462 allowed calls, roles.py:520-522).

The source says explicitly do not use disallowed_tools to implement "coordinate only, don't touch" — that is session-scoped and would disable subagents' Bash/Write along with it; see the warning under AgentSpec. The correct approach is the one used here: delegate_only=True + not granting allowed_tools, then letting delegate_guard block only the main thread by agent_id.

If channel is given you get both tools together, not optionally: once the MCP server is mounted both are there, and allowed_tools is not exclusive, so they are callable whether listed or not. Unattended, every ask blocks for the full timeout_s — for that situation 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

Builds the definition of the subagent that actually does the work. The first two parameters are positional. It returns an SDK AgentDefinition, which goes straight into coordinator(workers={...}).

isolate=True requires the workspace to be a git repository; otherwise the Agent tool errors out with "not in a git repository" — it does not silently degrade. And the marker is a Python attribute — running dataclasses.replace() on an AgentDefinition drops it, and isolation silently stops working.

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

Builds the clarifier: pin down the requirements before any work starts, do nothing but ask, and finally output exactly four sections.

The resulting AgentSpec: allowed_tools = [channel.tool_name] + (those five when readable), mcp_servers = channel.mcp_servers(), workbench=False (it has no write tools, so the index is meaningless to it), and permission_mode inherits AgentSpec's default of "default". No Write / Edit / Bash / Agent, and no inbox (unlike the coordinator).

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

Builds the judge: either set the goal before the run starts, or deliver a verdict on each round after it ends.

The resulting AgentSpec: allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (when can_run) ["Bash"], workbench=False, everything else as in clarify(). No Write / Edit / Agent, and no inbox.

Trade-off: can_run=True makes the verdict harder (it can actually run the acceptance commands), at the cost that the judge can then modify the workspace — Bash alone can write files. If you want an absolutely neutral verdict, leave it off.

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

Builds the oracle: while a run is still going, ask it "where are we now", and it glances at recent events and the workbench before answering. Nothing it says enters that run's context.

The resulting AgentSpec: allowed_tools = ["Read", "Glob", "Grep"] (no channel — it does not ask, it only answers), workbench=True (the only non-coordinator among the five roles with the workbench on — reading those outputs and notes is exactly its job).

The five rule texts¶

All five constants are in __all__; you can import them to read, splice, or modify.

Agent definition¶

Source: flower/core/agent.py

AgentSpec is the complete declaration of one specialized agent, and build_options compiles it into the SDK's ClaudeAgentOptions. What the role factory produces is exactly an AgentSpec — when you need a combination outside the factories, construct it directly.

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

Compiles an AgentSpec into the SDK's ClaudeAgentOptions. This is what Runtime._attempt calls internally; it is also the entry point when you drive the SDK yourself (without Runtime).

The mapping:

PLUGIN_DIR is the repo root's plugin/ (three levels up from flower/core/agent.py). After a pip install this directory may not exist, so the code checks with is_dir().

CompactPolicy¶

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

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

The control panel for auto-compaction; its product is a set of environment variables to inject into the subprocess. The compaction algorithm itself lives in the harness binary and cannot be changed; all you can change is whether it triggers.

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)

The policy object for "write a handoff document and start a new session" instead of compacting when the context is nearly full.

default_window()¶

def default_window() -> int

Guesses the context window from the model name string in the ANTHROPIC_MODEL or ANTHROPIC_DEFAULT_OPUS_MODEL environment variable:

The default is the aggressive value. Guessing too high is not a hard error: the API rejects with prompt is too long, and Runtime recognizes that signal (internally is_overflow) and hands off on the spot — but that generation's handoff is a degraded one.

Documents¶

Source: brief.py · handoff.py · goal.py

Four dataclasses, all of them "parse a block of model output into a fixed set of sections, then spill it to disk". The shared shape: parse() parses, missing() / complete() check completeness, to_markdown() is for humans, prompt_block() is for downstream models, write() / load() spill and read back.

Brief¶

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

The brief, exactly four sections, in the fixed order goal → accept → bounds → unknowns; the Chinese section names are 「目标」「验收标准」「边界」「未知与假设」.

Parsing rules (where the mistakes cluster):

• When stripping fences, **an unclosed `` or_~causes everything from that point on to be discarded** — in practice the clarifier pastes entire source files into its reply. When the model's output is truncated, all later sections fail to parse, socomplete()isFalse` and the gate sends it back.
• The heading regex tolerates ## 目标 / **目标** / 目标: / 3. 边界, and also tolerates body text directly after the heading.
• The alias table is compiled longest-first, otherwise "未知" would swallow "未知与假设".
• When the same section appears more than once, the first one with content wins.
• If you hand-edit the brief and copy to_markdown()'s "(未填)" placeholder text verbatim, that section still counts as missing.

Handoff¶

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

The handoff document written at handoff time, five sections.

Only doing and next are required — insisting that "paths that don't work" be non-empty would force the model to invent them.

Three members in the same module that are not exported but semantically critical: is_overflow(*texts) matches prompt is too long, context length exceeded, maximum context length, too many total text bytes, input length and max_tokens exceed and others, turning a "hard error" into an immediate handoff; HANDOFF_PROMPT is the prompt that makes the current session itself write the handoff (it contains the two placeholders {used} and {window}, and is not a new role — only that session has that context); degraded(step, prompt, *, why="") mechanically assembles one when the handoff cannot be written, stuffing the first 1200 characters of the original task into scene.

Goal¶

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

The goal plus check list used by the goal guard.

Verdict¶

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

The result of one round of judging by the judge, exactly three sections: conclusion / reason / not passed.

parse's recognition order:

1. First take the 「结论」/「判定」 section by heading.
2. With no heading section, strip the whole text and fullmatch(r"1|true") → achieved; fullmatch(r"0|false") → not yet.
3. Otherwise, search the conclusion text against the state-word table (longer words first) for the first hit. 「无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable」 all map to unreachable — this has bitten us: target platform macOS, running in a Linux container, and the judge passed it after reading a source-code branch.
4. Still nothing → look for a standalone \b1\b → achieved, \b0\b → not yet.
5. Nothing matches at all → state="", ok=False.

unreachable and not_yet are two different conclusions: the former takes the "stop and ask a human" path, not "run another round".

Hook layer¶

Source: flower/core/guard.py

This layer is flower's execution boundary: which tools the main thread may not touch, how oversized results get trimmed, which role goes into its own worktree — all enforced by SDK hooks, not by prompts. The reason is direct: a prompt is a suggestion, and the model may ignore it. Measured: even with the system prompt explicitly saying "do not use worktree", isolate_guard's injection still took effect (the model passed None, what landed was 'worktree').

Nine exports: five guard factories returning a HookMatcher (whitelist_guard may return None), one assembler, one merger, two isolation-marking functions. You don't need to wire them up by hand — Runtime assembles them automatically from the AgentSpec. Manual wiring is only needed when you drive the SDK yourself (bypassing Runtime).

Main-thread detection goes through a single function: _is_main_thread(data) = not data.get("agent_id") — a subagent's tool-lifecycle hook data carries agent_id, the main thread does not. Every "main-thread only" guard rests on this one line.

Tool-group constants (module level, not exported, but they determine the default matchers):

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

Quick reference: which guard hangs on which SDK event¶

The key inference to read off this table: with Runtime(workbench=False), workbench_hooks is not installed at all; and for a delegate_only=True coordinator, whitelist_guard is skipped too — the main thread has no wall at all. See the warning under Runtime.

whitelist_guard()¶

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

Makes allowed_tools genuinely exclusive for the four hands-on tools.

• Hangs on PreToolUse, matcher is "|".join(banned), where banned = those of Bash Write Edit NotebookEdit not in allowed.
• A hit is permissionDecision: "deny", with text along the lines of: "XX does not have YY. This is intentional, not a missing config. Write the conclusion in the body of your reply, the framework picks it up from there — don't try other phrasings to get around it."
• Intercepts only this session's main thread, subagents pass — a subagent's tools are determined by AgentDefinition.tools.
• Returns None when there is nothing to intercept (e.g. a role like worker() with the full tool set), so the caller can decide whether to install it.

Why it has to exist: allowed_tools is an approval-free list, not an exclusive whitelist. Two pieces of measured evidence — a judge with a target ran Bash 11 times; in the $0.1 probe, an agent with allowed_tools=["Read"] could still call Write/Bash. So the "no write tools" property of clarify() / judge() comes from this hook, not from the whitelist itself.

The upside is that it derives from allowed_tools, so judge(can_run=True) automatically keeps Bash while still blocking Write/Edit — no extra switch needed.

delegate_guard()¶

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

Main thread doing the work itself → denied, with directions.

• Hangs on PreToolUse, matcher is exactly tools.
• Main thread calling one of these four tools → deny, and the reason says what to do next: dispatch a subagent with the Agent tool, state the goal and acceptance criteria in the task, and require it to write long output into .flower/artifacts/ and reply with only the path and the conclusion.
• Subagents always pass.

The difference from whitelist_guard is wording: both intercept the same set of tools, but this one says "go dispatch someone", which is the right message here. So a delegate_only=True role installs only this one; installing both would give the model two contradictory instructions.

The pass criterion for allow_glance=True and "will this result get trimmed" are the same function (is_ephemeral) — the pass set must equal the expiry set; change one and you must change the other.

spill_guard()¶

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

Tool results over the threshold are spilled on the spot, leaving one pointer line in context — not compacted after the fact once the context is full.

• Hangs on PostToolUse, returns {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <trimmed>}}.
• The spill file is named with the first 16 characters of the content's sha256 + .txt; in context it becomes one pointer line + the first 400 characters.
• updatedToolOutput must preserve the original tool's output structure, so it only replaces overly long string fields in the dict; lists are never touched (they may contain image blocks). A wrong structure is rejected (original kept, no error).
• Reading a spill file itself must be let through — otherwise "read it with Read" is an empty promise: the full text read back gets spilled again, an infinite loop. Measured: hit it in practice, the model tried five phrasings in a row to get around it.

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

Anything written to the workbench refreshes INDEX.md, so the next agent knows at the start that it exists.

Hangs on PostToolUse, matcher "Write|Edit". If tool_input["file_path"], after resolve, falls inside workbench.root, it calls workbench.refresh(). Always returns {} — it changes nothing, it only has a side effect.

isolate_guard()¶

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

Assigns subagents their own git worktree by role, implementing isolation.

Hangs on PreToolUse, matcher "Agent". Injection happens only when all three conditions hold: tool_name == "Agent", tool_input has neither cwd nor isolation, and the role behind subagent_type is marked by isolated(). If so it returns permissionDecision: "allow" + updatedInput (setting isolation to "worktree").

isolation and cwd are mutually exclusive in the Agent tool — if the model specified cwd itself, that is respected. "Whether to isolate" is a property of the role, not a global switch, and not something decided per dispatch; a role that doesn't need isolation gets not one byte added.

Turning on isolation means moving the workbench out of the repo. An isolated agent cannot write into the shared checkout, so the workbench must be pointed outside the repo with home=. starter_flow(isolate=True) uses <ws>.parent/.flower-<ws.name>, Runtime(workbench=True) uses <run_dir>/workbench — both outside the repo, but not the same directory; don't mix them.

isolated() / wants_isolation()¶

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

Marks a subagent definition as "needs its own workspace", and reads that mark back.

The mark is a Python-side attribute _flower_isolate set via object.__setattr__, not a dataclass field — the SDK serializes with asdict(), which only sees declared fields, so this mark never leaks over to the CLI side (measured).

The cost: dataclasses.replace() on an AgentDefinition drops the mark, and isolation silently stops working.

worker(isolate=True) internally goes through 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]]

Installs the hooks a workbench needs in one call. This is what Runtime._attempt calls.

Output:

• PreToolUse: delegate_only=True → [delegate_guard(allow_glance=allow_glance)]; if some role is marked → append isolate_guard(agents).
• PostToolUse: always [index_guard(workbench)]; if spill_threshold is truthy → append spill_guard(workbench, threshold=spill_threshold).
• Event keys with empty lists are dropped, no empty list is returned.

merge_hooks()¶

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

Concatenates several hook configs by event name.

Uses extend, no dedup — pass the same guard twice and it gets installed twice. Runtime uses it to merge spec.hooks, workbench_hooks(...) and whitelist_guard.

Workbench¶

Source: flower/core/workbench.py

Workbench¶

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

The on-disk working directory: three subdirectories plus an index. The index is injected into the system prompt, so the agent knows every turn what it has on hand.

Script self-description format: # desc: one line within the first 8 lines (also accepts // and -- comment markers), falling back to the first non-empty comment line or the first line of the docstring (truncated at 100 characters).

The three rules prompt_block() injects:

1. Scripts meant to be run a second time go in scripts/, with # desc: on the first line.
2. Output over 2000 characters goes in artifacts/, with only the path and conclusion in conversation.
3. Key decisions go in notes/, one file per decision.

When external=True, prompt_block() adds a sentence saying "access it by absolute path".

Subagents do not inherit the index. It goes through the session-level system_prompt.append, and a subagent has its own system prompt (measured at $0.2461). So "write long output to artifacts/" and "where the workbench is" must be relayed by the coordinator in the task brief — that is the only channel, not redundancy. WORKER_RULES deliberately omits this: the real paths are generated by Workbench, hard-coding them would be wrong.

Session store¶

Source: sqlite.py · trim.py · prune.py

Three layers of inheritance: SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore. Runtime always uses the outermost one; the policies of all three layers are controlled by constructor arguments.

Each layer does one thing: persist, trim by size and value, prune by "is it an error". Both trimming and pruning happen at load() (i.e. the moment resume feeds history back to the model); not a byte of the raw record in SQLite is touched.

SqliteSessionStore¶

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

Implements the SDK's SessionStore protocol with three tables entries / meta / summaries. The store key is project_key/session_id[/subpath] — subagent transcripts are distinguished by subpath.

The internal _next_mtime guarantees strict monotonicity — list_sessions and the summary sidecar share this clock, otherwise the SDK's staleness fast path misjudges.

TrimPolicy¶

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

The two spill directories are not the same one. spill_guard lands in <workbench.root>/spill/ (which may be outside the workspace); TrimPolicy.spill_dirname lands in <workspace>/.flower/spill/ (must be inside the workspace). They correspond to "trim on the spot" and "trim at resume" respectively; the different directories are intentional, don't merge them.

EphemeralPolicy¶

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

The expiry policy for ephemeral command results.

Applies only to Bash tool results, and the command must match the ephemeral-command whitelist. Read is not included — file contents don't decay with time to the point of being misleading. Expired content is not spilled, it is simply thrown away.

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

Decides whether a Bash command is an ephemeral command. delegate_guard's pass decision and trimming's expiry decision share this one function — the set of commands the coordinator may run itself must equal the set whose results get marked expired. Pass without trim, and an expired git status occupies context forever and misleads; trim without pass, and the coordinator dispatches a subagent for one ls, trading 4.3k of startup cost for a few dozen characters.

Decision order:

1. empty / all whitespace → False.
2. command substitution ($(, backticks, <(, >() or a state-changing form matched → False.
3. after stripping safe redirections (2>&1, &> /dev/null and the like), still contains > or < → False.
4. after stripping && / || / ; / |, a lone & remains (background execution) → False.
5. split on && / || / ; / |, and every segment must hit the whitelist.

Whitelisted verb categories: read-only git subcommands (status diff log show branch rev-parse etc.), directory and system info (ls pwd df du date whoami env etc.), processes and containers (ps top lsof docker ps kubectl get etc.), viewing files (cat head tail wc stat find tree), locating paths (which whereis command -v type), text processing (grep rg sort uniq awk sed jq diff etc.).

Even with a whitelisted verb, these forms are blocked: xargs, exec, eval, source, tee, find -delete / -ok / -fprint, sed -i, sort -o, system( and print > inside awk, git branch -D/-d/-m, git * --force/--hard/--prune.

The first version rejected all compound commands outright, which measurably broke glance entirely (all three of the coordinator's attempts were blocked), so it was changed to per-segment judgement.

TrimmingSessionStore¶

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

Public attributes: workspace, policy, ephemeral, last_report: dict[str, int].

Order in load(): super().load() → clear last_report → if ephemeral.enabled, expire() → if policy.enabled, trim(). When enabled=False the step is skipped entirely.

Only plain text is trimmed: image / document blocks are left as they are.

Two structural red lines: the tool_result block itself must remain, only the content may be replaced (one missing and you get "Missing Tool Result Block"); isCompactSummary entries must not be touched.

trim_report()¶