Python API¶

Diese Seite arbeitet die 62 öffentlichen Symbole von flowers Top-Level-__all__ erschöpfend durch: Signaturen, Parameter, Defaults, Semantik, öffentliche Attribute und Methoden. Nach dem Lesen musst du für Parameter nicht mehr in den Quellcode schauen.

Gegliedert ist sie nach dem, was dich interessiert, nicht nach Moduldateien — willst du wissen, "wie halte ich den Coordinator davon ab, selbst Hand anzulegen", geh zur Hook-Schicht; willst du wissen, "wie kommt das Ergebnis des vorigen Schritts in den nächsten", geh zum Workflow. Terminologie durchgehend nach dem Glossar.

Version 0.1.0, Abhängigkeit claude-agent-sdk>=0.2.152. Alle Signaturen entsprechen wörtlich dem Quellcode.

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

Was auf dieser Seite steht¶

Sechs Defaults, die beißen¶

Diese sechs Punkte sind kein Randmaterial, sondern die sechs häufigsten Unfälle. Jeder ist im zugehörigen Abschnitt vollständig erklärt.

Runtime¶

Quellcode: flower/core/runtime.py

Runtime ist der Ausführungskern. Er hält Workspace, Session Store, Workbench, Resilience-Strategie und Handoff-Strategie und hat nach außen nur ein Verb: einen Schritt run. Retry, Weiterlaufen nach einer Unterbrechung, Handoff bei vollem Kontext — alles passiert innerhalb dieses einen Aufrufs.

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

Die Konstruktorparameter sind allesamt keyword-only (* ganz vorne), workspace ist Pflicht.

Der Session Store ist fest verdrahtet: immer PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...)). Die Konstruktorparameter bieten keinen Einstiegspunkt zum Wechseln des Backends — wer wechseln will, baut sich selbst AgentSpec + build_options(session_store=...) oder überschreibt nach dem Konstruieren rt.store.

Die letzten beiden Schritte des Konstruktors sind load_dotenv() und check_credentials(), letzteres wirft bei Fehler RuntimeError. Ohne Credentials fliegt es also schon beim Konstruieren, nicht erst bei run().

Öffentliche Attribute¶

Klassenkonstanten: INTERRUPTED = "interrupted-by-human", HANDOFF_DUE = "context-full-handoff", INTERRUPT_NOTE (ein Absatz, der beim Weiterlaufen nach einer Unterbrechung an die Worte des Menschen angehängt wird und erklärt, dass "ein damals fliegender Tool-Aufruf, der interrupted zurückgibt, eine normale Nebenwirkung der Unterbrechung ist, kein Umgebungsfehler").

Öffentliche Methoden¶

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

Zu Beginn jedes Schritts wird der Kontextpegel auf null gesetzt (self._ctx, self._warned = 0, False). Danach folgt eine Schleife mit vier Ausgängen:

1. Erfolg → raus.
2. Mensch unterbricht (result.error == INTERRUPTED) → nicht durch max_attempts begrenzt, wartet nicht aufs Netz. Mit den Worten des Menschen wird dieselbe Session resumet, attempt -= 1 (eine Unterbrechung zählt nicht als fehlgeschlagener Versuch), Prompt = Worte des Menschen + INTERRUPT_NOTE. Ohne session_id bleibt nur anhalten.
3. Kontext voll (result.error == HANDOFF_DUE, oder handoff.enabled und eine session_id liegt vor und is_overflow(...) greift) → ebenfalls nicht durch max_attempts begrenzt. Zuerst wird len(result.retired) >= handoff.max_generations geprüft, bei Überschreitung wird der Fehler durch eine Diagnosemeldung ersetzt und die Schleife verlassen; sonst Handoff-Dokument schreiben → resume=None, fork=False (ganz neue Session) → Prompt wird zu h.prompt_block() → Pegel auf null → attempt -= 1.
4. Wiederholbarer Fehler → bei not resilience.enabled or attempt >= max_attempts raus; ergibt classify(error), dass nicht wiederholt werden soll, ebenfalls raus; sonst Event("retry") senden, mit wait_online() aufs Netz warten, sleep(delay_for(attempt)); lag jemals eine session_id vor, wird mit resume fortgesetzt (Prompt wird zu resilience.resume_prompt) und result.resumed auf True gesetzt.

Zum Abschluss: ended_at schreiben, an self.results anhängen, manifest.json schreiben.

manifest.json hat Append-Semantik: bei jedem Schreiben wird die Platte neu gelesen und nach dem Feld run dedupliziert (die eigene Zeile wird ersetzt, fremde bleiben stehen), deshalb ist es sicher, zwei flower parallel im selben run_dir laufen zu lassen — vorausgesetzt, die run_ids kollidieren nicht.

Die drei Beobachtungspunkte des Handoffs (alle Event("handoff"), unterschieden durch payload["phase"]): near (nähert sich warn_at, wird pro Generation nur einmal gesendet), writing (das Handoff-Dokument wird geschrieben, dauert gut zehn Sekunden), done (Payload enthält degraded / path / sections). Die Runde, die das Handoff-Dokument schreibt, läuft mit replace(spec, max_budget_usd=None) — das Handoff muss geschrieben werden können und darf nicht am Budget hängenbleiben; außerdem on_event=None, diese Runde erscheint nicht im UI.

Das Handoff-Dokument landet unter <workbench.notes>/交接-<步骤名>.md; ohne Workbench wird nichts geschrieben, das Dokument geht trotzdem per Prompt an den Nachfolger, es lässt sich hinterher nur nicht mehr nachschlagen. Alte Handoffs werden nach notes/archive/交接/<名>-<时间戳>.md verschoben.

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

Die komplette Abrechnung eines gelaufenen Schritts.

Workflow¶

Quellcode: flower/workflow/

Ein Workflow ist eine Reihe hintereinandergehängter Schritte, plus die Regeln, wie Zustand zwischen den Schritten weitergereicht wird und wann vorzeitig abgebrochen wird. Das Framework liefert keine fertigen Workflows, den Workflow schreibst du — starter_flow ist nur eine lauffähige Vorlage.

Typ-Alias Ctx = dict[str, Any] (flower.workflow.base.Ctx, steht in flower.workflow.__all__, nicht im obersten __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

Die Deklaration eines Schritts. Step selbst ist keine Funktion — ausgeführt wird tatsächlich Runtime.run(step.spec, prompt, ...). Die ersten drei Felder sind positional, Step("取词", terse, "读 seed.txt …") ist eine gültige Schreibweise.

Drei Arten, Sessions zu verbinden (innerhalb desselben Runs):

on_reject ändert die Retry-Semantik:

• Nicht gesetzt → der nächste Versuch läuft von vorn (gleicher Prompt, gleiches resume_from).
• Gesetzt → der nächste Versuch setzt genau die eben abgelehnte Session fort, der Prompt wird durch den Rückgabewert ersetzt, fork wird auf False gezwungen.
• Rückgabe leerer String → keine Rückweisung, degradiert zum Lauf von vorn.
• result.session_id ist None → degradiert ebenfalls zum Lauf von vorn.

Die drei Werte von on_fail:

Ob bestanden oder nicht: ctx["_results"][name] = result wird immer geschrieben; ist result.session_id nicht leer, wird zusätzlich ctx["_sessions"] geschrieben und lineage.remember(...) aufgerufen.

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

Führt eine Reihe von Step der Reihe nach aus und gibt das finale ctx zurück. steps ist positional, Workflow([...]) ist gültig.

Die privaten Schlüssel, die run() in ctx schreibt (alle mit _ beginnend, kollidieren also nicht mit Schrittnamen):

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

Retry-Labels: Beim 0. Versuch step.name; danach mit on_reject f"{name}#round{attempt+1}", ohne on_reject f"{name}#retry{attempt}". Im Manifest sieht man auf einen Blick, wie dieser Schritt zu Ende gegangen ist. Namen mit Suffix gehen nicht in die prozessübergreifende Lineage — Lineage.remember verwendet den Originalnamen.

runtime.on_session umschließt nur die eine Zeile runtime.run; try/finally stellt sicher, dass es vor dem gate garantiert wieder abgenommen wird. prompt_cur / resume_cur / fork_cur sind lokale Variablen und werden nicht nach step zurückgeschrieben — dasselbe Step-Objekt kann ein zweites Mal laufen.

StepAbort¶

class StepAbort(Exception): ...

Aus dem gate geworfen = sofort stoppen, nicht weiter versuchen. Unterschied zu „False zurückgeben": False heißt „diesmal nicht, noch eine Runde"; StepAbort heißt „noch eine Runde bringt auch nichts".

Nach dem Wurf: ctx["_aborted"] = str(exc), passed = False, Ausstieg aus der Retry-Schleife (die restlichen retries werden nicht verbraucht), danach geht es wie bei einem normalen Fehlschlag über on_fail weiter (Default "stop").

with_goal wirft es an zwei Stellen: wenn ctx["_runtime"] nicht zu bekommen ist, und wenn das Verdict unreachable lautet und niemand antwortet.

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

Erzeugt einen Step, der Clarify durchführt: Bedarf klären → in einen Brief parsen → wenn alle vier Abschnitte vollständig sind, einfrieren und auf Platte schreiben.

So sind die Felder im erzeugten Step gefüllt:

• resume_prompt = CLARIFY_RESUME.
• when: bei always_ask=True → immer True; sonst wird Brief.load(brief_path) bei Vollständigkeit in ctx eingespielt und dann False zurückgegeben (übersprungen) — auch beim Überspringen muss eingespielt werden, sonst bekommt die nachgelagerte Stufe den Bedarf nicht.
• gate: Brief.parse(result.text), unvollständig → ctx[MISSING_KEY] schreiben und False zurückgeben; vollständig → b.write(brief_path) einfrieren, in ctx einspielen, True zurückgeben.
• reduce: gibt ctx[BRIEF_KEY].prompt_block() zurück, nicht den Originaltext des Modells — im Originaltext kann Zusatzgeschriebenes stecken.
• resume_from bleibt beim Default None: Der nächste Schritt ist eine neue Session, bekommt nur den Brief, nicht das Frage-Antwort-Protokoll. Die Fragen und Antworten des Clarify waren nie im Kontext des Koordinators, sie wurden nicht nachträglich herausgeschnitten.

Drei Stellen, an denen in ctx eingespielt wird: 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

Erzeugt einen Step, der das Ziel setzt: Der Judge liest den Brief, schreibt Ziel + Prüfliste, das Ganze wird in ein Goal geparst, eingefroren und auf Platte geschrieben. Gleiche Form wie clarify_step.

Es gibt keinen can_run-Parameter — wer will, dass der zielsetzende Judge Kommandos ausführen kann, muss can_run=True über **spec_kw durchreichen. Ohne das bekommt er kein Bash, und die Regel aus JUDGE_RULES „verschaffe dir zuerst Klarheit, in welcher Umgebung du bist" ist nicht ausführbar.

Das gate tut neben Parsen und Einfrieren noch eines: Enthält das Ziel Einträge [此环境无法验证:…], wird sofort über ctx["_on_event"] ein Event("task", payload={"unverifiable", "total", "path"}) als Hinweis gesendet — das Schicksal dieser Einträge entscheidet sich genau in diesem Moment des Zielsetzens; bis zum Verdict ist bereits das Geld einer ganzen Arbeitsrunde ausgegeben.

Es ist kein resume_prompt gesetzt — beim Zielsetzen soll der Brief ohnehin im Volltext neu geschickt werden.

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

Legt einem vorhandenen Step den Goal Guard um: Nach jeder Runde beurteilt der Judge unabhängig; ist das Ziel nicht erreicht, wird zurückgewiesen und weitergearbeitet.

Das Ergebnis ist replace(step, retries=max(0, rounds - 1), gate=<neues gate>, on_reject=<neues on_reject>) — mit dataclasses.replace statt feldweisem Neubau; beim Neubau ging einmal resume_prompt verloren, und zwar ohne Fehlermeldung, es wurde bei der Continuity nur der ganze Brief noch einmal verschickt.

Das gate ist async, Ablauf:

1. ctx["_runtime"] fehlt → StepAbort werfen („Runtime nicht verfügbar, Ziel kann nicht beurteilt werden"). Nicht so tun, als sei es bestanden.
2. ctx[ROUND_KEY] += 1.
3. Ziel holen: bevorzugt ein vollständiges Goal aus ctx[GOAL_KEY], sonst Goal.load(goal_path), sonst ein leeres Goal().
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...). Der Judge ist ein eigenständiges Runtime.run, resume ist immer None — es ist immer eine neue Session; step_name trägt die Rundennummer und geht damit nicht in die prozessübergreifende Lineage.
5. Verdict.parse(vr.text) wird nach ctx[VERDICT_KEY] geschrieben.
6. v.achieved → True zurückgeben.
7. Nicht unreachable (einschließlich unklarer Fälle mit v.ok=False) → bei Unklarheit wird ein Default-Reason ergänzt, Rückgabe False. Unklarheit zählt ausnahmslos als nicht erreicht — ein „sieht ok aus" darf die Arbeit nicht abschließen.
8. unreachable → await channel.ask(...) fragt den Menschen, drei Optionen:
9. Niemand antwortet (a.state != "answered") → StepAbort werfen. Weiter leerzudrehen ist die teuerste Option.
10. „Ergebnis akzeptieren und so weitermachen" → True zurückgeben.
11. „Ziel ändern" → noch einmal nach dem neuen Ziel fragen, g.amend(...).write(goal_path), ctx[GOAL_KEY] aktualisieren, False zurückgeben.
12. Alles andere (einschließlich frei getippter Antworten) → gilt als „du hast falsch beurteilt", die Aussage des Menschen wird in v.reason festgehalten, Rückgabe False.

on_reject ist synchron: gibt ctx[VERDICT_KEY].feedback() zurück, ohne Verdict "" (degradiert zum Lauf von vorn).

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

Baut einen sofort lauffähigen Drei-Schritt-Workflow zusammen: Bedarf klären → Ziel setzen → arbeiten (mit Goal Guard). Genau das benutzt die Kommandozeile flower.

Feste Verdrahtung:

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)

Verhaltensverzweigungen:

• isolate=True und der Workspace ist kein Git-Repo → ValueError werfen, nicht erst warten, bis das Agent-Tool einen Fehler meldet (dann ist das Geld schon ausgegeben).
• Wake-Erkennung: Existiert Brief.load(brief_path) und ist complete(), gilt es als Wake. Kein Wake und ask leer → ValueError("要给一句诉求,例如 flower '帮我做一个 X'") werfen.
• Beim Wake landet dieser Satz gleichzeitig an drei Stellen, fehlt eine davon, verpufft er still: angehängt an den Brief (ch.amend(said, label="唤醒时追加"), steht er schon in der Datei, wird er nicht doppelt geschrieben), goal_step(always_set=True) leitet die Prüfliste neu her (ohne Neuherleitung liest der Judge weiter das alte Ziel), und er geht direkt an den Koordinator (in dessen Kontext steht das alte Ziel; ohne diesen Schritt arbeitet er nach altem Maßstab und wird nach neuem beurteilt).

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

Eine nur lesende Sondierung vor dem Start, es wird kein einziges Byte geschrieben. Damit man dem Menschen vor dem eigentlichen Lauf sagen kann: „Das ist eine Fortsetzung" oder „das fängt von vorn an".

Das zurückgegebene dict:

Die Lage der Workbench wird nur hier und in starter_flow je einmal definiert: isolate=True → <ws>.parent/.flower-<ws.name> (außerhalb des Repos); sonst <ws>/.flower. Auch ein Treiberprogramm, das wissen will, wo der Brief liegt, geht über diese Funktion — wer den Pfad selbst zusammensetzt und sich vertut, bekommt keinen Fehler, es verpufft nur still.

Rollen-Factories¶

Quelle: flower/core/roles.py

Alle fünf Rollen sind Factory-Funktionen. Jede Rolle = ein Stück injizierter Regeltext + ein Satz Tools + ein Satz Hooks. worker() liefert eine AgentDefinition des SDK (für Subagents gedacht), die anderen vier liefern AgentSpec (startet jeweils eine eigene Session).

Die Rollen selbst hängen keine Hooks ein — das Abfangen von Tools montiert Runtime._attempt automatisch anhand von spec.delegate_only, siehe Hook-Schicht.

Interne Tool-Gruppenkonstanten (nicht exportiert, bestimmen aber die 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

Baut den Koordinator auf dem Haupt-Thread: Aufgabe zerlegen, delegieren, Berichte lesen, entscheiden, aber nicht selbst Hand anlegen. Die ersten drei Parameter sind Positionsargumente.

Drei Dinge stehen in der erzeugten AgentSpec fest: delegate_only=True, agents=workers, und workbench behält den AgentSpec-Default True.

Die Web-Tools aus workers werden mit eingemischt¶

Nachdem die Liste gebaut ist, geht coordinator() jede AgentDefinition.tools durch; alles, was in WEB_TOOLS (WebFetch, WebSearch, roles.py:33) liegt, wird zusätzlich in die eigenen allowed_tools des Koordinators aufgenommen (roles.py:523-526).

Grund: allowed_tools ist genau wie disallowed_tools session-weit. Das ist der härteste Beleg im ganzen Text für diesen Punkt — es betrifft nicht nur den Haupt-Thread. Ein Tool, das nicht in dieser session-weiten Liste steht, muss auch beim Aufruf durch einen Subagent durch die Berechtigungsfreigabe; im unbeaufsichtigten Betrieb gibt es niemanden, der freigibt, und die Harness antwortet mit Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected), während das Modell denselben Aufruf immer wieder wiederholt. In der Praxis reingefallen: dem Worker WebFetch/WebSearch gegeben, aber nur in AgentDefinition.tools geschrieben — in jenem Novel-Run gab es über zwanzig user-rejected und kein einziges geschriebenes Wort (roles.py:513-518).

Beide Felder sind gleichermaßen session-weit, die Symptome unterscheiden sich: disallowed_tools wirft sofort einen Fehler, allowed_tools führt zu stillem Wiederholen bis zum Ende. Letzteres ist schwerer zu finden, weil auf dem Bildschirm nichts nach Fehler aussieht.

Eingemischt wird nur, was lesend und nebenwirkungsfrei ist. Write/Edit/Bash werden absichtlich nicht eingemischt: Sobald der Haupt-Thread dafür freigabefrei ist, ist die Mauer „der Koordinator legt nicht selbst Hand an" von delegate_guard sinnlos; und Bash/Write im Subagent funktionieren ohnehin (gemessen 462-mal durchgelassen, roles.py:520-522).

Im Quellcode steht ausdrücklich, dass man „nur koordinieren, nicht selbst Hand anlegen" nicht über disallowed_tools umsetzen soll — das ist session-weit und sperrt Bash/Write der Subagents gleich mit, siehe die Warnung bei AgentSpec. Der richtige Weg ist genau der hier: delegate_only=True + nichts in allowed_tools, und delegate_guard blockt anhand der agent_id nur den Haupt-Thread.

Ist channel gesetzt, kommen beide Tools zusammen, das ist nicht optional: Hängt der MCP-Server dran, sind beide da, und allowed_tools ist nicht exklusiv — aufrufbar sind sie so oder so. Im unbeaufsichtigten Betrieb blockiert jedes ask die volle timeout_s — für solche Fälle HumanChannel(timeout_s=0) verwenden.

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

Baut die Definition des Subagents, der die eigentliche Arbeit macht. Die ersten beiden Parameter sind Positionsargumente. Zurück kommt eine AgentDefinition des SDK, die direkt in coordinator(workers={...}) gesteckt wird.

isolate=True verlangt, dass der Workspace ein Git-Repository ist, sonst meldet das Agent-Tool direkt "not in a git repository" — es degradiert nicht still. Und die Markierung ist ein Python-Attribut — ein dataclasses.replace() auf der AgentDefinition verliert sie, die Isolation fällt still aus.

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

Baut den Clarifier: vor dem Loslegen die Anforderungen klären, nichts tun, nur fragen, am Ende genau vier Abschnitte ausgeben.

Die erzeugte AgentSpec: allowed_tools = [channel.tool_name] + (die fünf, wenn lesend), mcp_servers = channel.mcp_servers(), workbench=False (er hat keine Schreib-Tools, der Index bringt ihm nichts), permission_mode erbt den AgentSpec-Default "default". Kein Write / Edit / Bash / Agent, und auch kein inbox (anders als beim Koordinator).

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

Baut den Judge: entweder vor dem Start das Ziel festlegen oder nach jeder Runde diese Runde beurteilen.

Die erzeugte AgentSpec: allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (bei can_run) ["Bash"], workbench=False, der Rest wie bei clarify(). Kein Write / Edit / Agent, und auch kein inbox.

Abwägung: can_run=True macht die Beurteilung härter (Abnahmebefehle können wirklich laufen), der Preis ist, dass der Judge damit den Arbeitsbereich verändern kann — Bash allein kann schon Dateien schreiben. Wer absolut neutrale Beurteilung will, lässt es aus.

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

Baut den Oracle: Während der Run noch läuft, fragt man ihn „wo stehen wir gerade", er sieht sich die letzten Events und die Workbench an und antwortet dann. Was er sagt, gelangt nicht in den Kontext dieses Runs.

Die erzeugte AgentSpec: allowed_tools = ["Read", "Glob", "Grep"] (kein Channel — er fragt nicht, er antwortet nur), workbench=True (der einzige der fünf Rollen, der kein Koordinator ist und trotzdem die Workbench anhat — er soll ja genau die Ergebnisse und Notizen lesen).

Die fünf Regeltexte¶

Alle fünf Konstanten stehen in __all__, man kann sie direkt importieren, lesen, zusammensetzen und ändern.

Agent-Definition¶

Quelle: flower/core/agent.py

AgentSpec ist die vollständige Deklaration eines spezialisierten Agents, build_options kompiliert sie zu den ClaudeAgentOptions des SDK. Die Rollen-Factories liefern genau eine AgentSpec — wer eine Kombination außerhalb der Factories braucht, konstruiert sie direkt.

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

Kompiliert eine AgentSpec zu den ClaudeAgentOptions des SDK. Genau das ruft Runtime._attempt intern auf; wer das SDK selbst steuert (ohne Runtime), steigt ebenfalls hier ein.

Zuordnung:

PLUGIN_DIR ist das plugin/ in der Repo-Wurzel (relativ zu flower/core/agent.py drei Ebenen nach oben). Nach einer pip-Installation existiert dieses Verzeichnis nicht zwingend, der Code prüft es mit is_dir().

CompactPolicy¶

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

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

Das Schaltbrett für den Auto-Compact, Ergebnis ist ein Satz Umgebungsvariablen für den Subprozess. Der Compact-Algorithmus selbst steckt im Harness-Binary und ist nicht änderbar, änderbar ist nur, „ob er auslöst".

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)

Das Policy-Objekt für „bei vollem Kontext ein Handoff-Dokument schreiben und eine neue Session starten" statt zu komprimieren.

default_window()¶

def default_window() -> int

Schätzt das Kontextfenster anhand des Modellnamen-Strings in den Umgebungsvariablen ANTHROPIC_MODEL oder ANTHROPIC_DEFAULT_OPUS_MODEL:

Der Default ist aggressiv. Zu groß geschätzt ist kein harter Fehler: Die API weist mit prompt is too long zurück, Runtime erkennt dieses Signal (intern is_overflow) und macht auf der Stelle einen Handoff — aber der Handoff dieser Generation ist die degradierte Fassung.

Dokumente¶

Quelle: brief.py · handoff.py · goal.py

Vier Dataclasses, alle nach dem Muster „eine Antwort des Modells in fest definierte Abschnitte parsen und auf Platte schreiben". Gemeinsame Form: parse() parst, missing() / complete() prüft auf Vollständigkeit, to_markdown() ist für Menschen, prompt_block() für nachgelagerte Modelle, write() / load() schreibt auf Platte und liest zurück.

Brief¶

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

Das Brief, genau vier Abschnitte, feste Reihenfolge goal → accept → bounds → unknowns, die chinesischen Abschnittsnamen lauten 「目标」「验收标准」「边界」「未知与假设」.

Parse-Regeln (hier ballen sich die Fehlerquellen):

• Beim Entfernen der Umzäunung wird **ab einem nicht geschlossenen `` oder_~alles Folgende verworfen** — gemessen: der Clarifier klebt ganzen Code in die Antwort. Ist die Modellausgabe abgeschnitten, lassen sich die folgenden Abschnitte gar nicht parsen, also istcomplete()gleichFalse` und das Gate schickt es zurück.
• Die Überschriften-Regex toleriert ## 目标 / **目标** / 目标: / 3. 边界 und auch, dass direkt hinter der Überschrift der Fließtext beginnt.
• Die Alias-Tabelle wird nach Länge absteigend kompiliert, sonst schluckt „未知" zuerst „未知与假设".
• Kommt derselbe Abschnittsname mehrfach vor, wird der erste mit Inhalt genommen.
• Wer das Brief von Hand bearbeitet und dabei den Platzhaltertext "(未填)" aus to_markdown() mitkopiert, hat diesen Abschnitt weiterhin als fehlend.

Handoff¶

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

Das beim Handoff geschriebene Handoff-Dokument, fünf Abschnitte.

Pflicht sind nur die zwei Abschnitte doing und next — hart zu fordern, dass „Wege, die nicht funktionieren" nicht leer ist, würde das Modell zum Erfinden zwingen.

Drei nicht exportierte, aber semantisch entscheidende Member im selben Modul: is_overflow(*texts) matcht prompt is too long, context length exceeded, maximum context length, too many total text bytes, input length and max_tokens exceed und andere und verwandelt einen „harten Fehler" in ein „sofort Handoff"; HANDOFF_PROMPT ist der Prompt, mit dem die aktuelle Session selbst ihren Handoff schreibt (enthält die beiden Platzhalter {used} und {window}, das ist keine neue Rolle — nur sie selbst hat diesen Kontext); degraded(step, prompt, *, why="") setzt mechanisch einen Handoff zusammen, wenn keiner geschrieben werden konnte, und stopft in scene die ersten 1200 Zeichen der ursprünglichen Aufgabe.

Goal¶

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

Das Ziel plus Prüfliste im Goal Guard.

Verdict¶

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

Das Ergebnis einer Beurteilungsrunde des Judge, genau drei Abschnitte: 结论 / 理由 / 未通过.

Erkennungsreihenfolge von parse:

1. Zuerst über die Überschriftenabschnitte 「结论」/「判定」 holen.
2. Gibt es keine Überschriftenabschnitte, wird der ganze Text nach strip mit fullmatch(r"1|true") → erreicht; fullmatch(r"0|false") → noch nicht.
3. Sonst im Ergebnistext anhand der Statuswortliste (längere Wörter zuerst) das erste Treffer-Wort suchen. 「无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable」 fallen alle unter unreachable — in der Praxis reingefallen: Zielplattform macOS, gelaufen im Linux-Container, der Judge sah sich den Quellcode-Zweig an und wertete es als bestanden.
4. Weiterhin nichts → ein isoliertes \b1\b suchen → erreicht, \b0\b → noch nicht.
5. Nichts davon trifft → state="", ok=False.

unreachable und not_yet sind zwei verschiedene Schlüsse: Ersteres geht den Weg „anhalten und den Menschen fragen", nicht „noch eine Runde".

hook-Ebene¶

Quelle: flower/core/guard.py

Diese Ebene ist die Ausführungsgrenze von flower: welche Tools der Main Thread nicht anfassen darf, wie überlange Ergebnisse gekürzt werden, welche Rolle in ein eigenes worktree geht — alles wird per SDK-hook erzwungen, nicht über den prompt. Der Grund ist schlicht: ein prompt ist ein Vorschlag, das Modell kann ihn ignorieren. Gemessen: selbst wenn im System-prompt ausdrücklich steht „kein worktree benutzen", greift die Injektion von isolate_guard trotzdem (das Modell übergibt None, gelandet ist 'worktree').

Neun Exporte: fünf guard-Fabriken, die einen HookMatcher zurückgeben (whitelist_guard kann None liefern), ein Zusammenbauer, ein Merger, zwei Isolations-Markierungsfunktionen. Man muss sie nicht von Hand einhängen — Runtime montiert sie anhand der AgentSpec automatisch. Von Hand einhängen ist nur nötig, wenn man das SDK selbst treibt (ohne Runtime).

Die Erkennung des Main Threads läuft einheitlich über eine Funktion: _is_main_thread(data) = not data.get("agent_id") — in den tool-lifecycle-hook-Daten eines subagent steckt agent_id, beim Main Thread nicht. Alle guards, die „nur den Main Thread abfangen", hängen an dieser einen Bedingung.

Tool-Gruppen-Konstanten (Modulebene, nicht exportiert, bestimmen aber die Default-matcher):

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

Schnellübersicht: welcher guard hängt an welchem SDK-Event¶

Die zentrale Folgerung aus dieser Tabelle: bei Runtime(workbench=False) wird workbench_hooks komplett nicht installiert; und bei einem Coordinator mit delegate_only=True wird auch whitelist_guard übersprungen — der Main Thread hat dann keine einzige Mauer. Siehe die Warnung unter Runtime.

whitelist_guard()¶

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

Macht allowed_tools für die vier handanlegenden Tools tatsächlich exklusiv.

• Hängt an PreToolUse, matcher ist "|".join(banned), banned = diejenigen aus Bash Write Edit NotebookEdit, die nicht in allowed stehen.
• Treffer heißt permissionDecision: "deny", Text sinngemäß: „XX hat kein YY. Das ist Absicht, keine vergessene Konfiguration. Schreib das Ergebnis in den Fließtext deiner Antwort, das Framework holt es dort ab — versuch nicht, es über eine andere Schreibweise zu umgehen."
• Fängt nur den Main Thread dieser session ab, subagents gehen durch — deren Tools bestimmt AgentDefinition.tools.
• Gibt None zurück, wenn es nichts abzufangen gibt (etwa bei einer Rolle wie worker() mit vollem Tool-Satz); der Aufrufer entscheidet danach, ob er installiert.

Warum es das geben muss: allowed_tools ist eine Freigabeliste ohne Nachfrage, keine exklusive Whitelist. Zwei gemessene Belege — der Judge im Ziel-Setzen-Lauf hat 11-mal Bash ausgeführt; in der $0.1-Sonde konnte ein agent mit allowed_tools=["Read"] trotzdem Write/Bash aufrufen. Dass clarify() / judge() „keine Schreibtools haben", liegt also an diesem hook, nicht an der Whitelist selbst.

Der Vorteil: er leitet sich aus allowed_tools ab, deshalb behält judge(can_run=True) automatisch Bash und blockt weiterhin Write/Edit — kein zusätzlicher Schalter nötig.

delegate_guard()¶

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

Main Thread legt selbst Hand an → Ablehnung, mit Wegweiser.

• Hängt an PreToolUse, matcher ist genau tools.
• Ruft der Main Thread eines dieser vier Tools auf → deny, und die Begründung sagt, was als Nächstes zu tun ist: mit dem Agent-Tool einen subagent schicken, in der Aufgabe Ziel und Abnahmekriterien klar benennen und verlangen, dass lange Ergebnisse nach .flower/artifacts/ geschrieben werden und die Antwort nur Pfad und Schlussfolgerung enthält.
• subagents gehen ausnahmslos durch.

Der Unterschied zu whitelist_guard ist die Formulierung: beide fangen dieselben Tools ab, aber dieser hier sagt „delegier das", was hier passender ist. Deshalb bekommt eine Rolle mit delegate_only=True nur diesen einen; beide zusammen würden dem Modell zwei widersprüchliche Anweisungen liefern.

Das Durchlasskriterium von allow_glance=True und die Frage „wird das Ergebnis später gekürzt" sind dieselbe Funktion (is_ephemeral) — die durchgelassene Menge muss der veralteten Menge entsprechen; ändert man die eine Seite, muss man die andere mitändern.

spill_guard()¶

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

Tool-Ergebnisse über dem Schwellwert werden sofort gespillt, im Kontext bleibt nur eine Zeiger-Zeile — nicht erst compacten, wenn der Kontext voll ist.

• Hängt an PostToolUse, liefert {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <gekürzt>}}.
• Der spill-Dateiname sind die ersten 16 Stellen des sha256 des Inhalts + .txt; im Kontext steht stattdessen eine Zeiger-Zeile + die ersten 400 Zeichen.
• updatedToolOutput muss die Ausgabestruktur des Originaltools beibehalten, deshalb werden nur die zu langen String-Felder im dict ersetzt; Listen werden nie angefasst (darin können Bildblöcke stecken). Eine falsche Struktur wird abgelehnt (Original bleibt, kein Fehler).
• Das Lesen der spill-Datei selbst muss durchgelassen werden — sonst ist „lies sie mit Read" eine leere Aussage: der zurückgelesene Volltext würde erneut gespillt, Endlosschleife. Gemessen aufgetreten; das Modell hat fünf Schreibweisen durchprobiert, um es zu umgehen.

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

Wird etwas in den Workbench geschrieben, wird INDEX.md aufgefrischt; der nächste agent weiß gleich zu Beginn, dass es das gibt.

Hängt an PostToolUse, matcher "Write|Edit". Liegt tool_input["file_path"] nach resolve innerhalb von workbench.root, wird workbench.refresh() aufgerufen. Gibt immer {} zurück — er ändert nichts, er hat nur einen Seiteneffekt.

isolate_guard()¶

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

Weist subagents je nach Rolle ein eigenes git-worktree zu und realisiert damit die Isolation.

Hängt an PreToolUse, matcher "Agent". Injiziert wird nur, wenn drei Bedingungen gleichzeitig gelten: tool_name == "Agent", tool_input hat weder cwd noch isolation, und die zu subagent_type gehörende Rolle wurde mit isolated() markiert. Dann kommt permissionDecision: "allow" + updatedInput zurück (isolation wird auf "worktree" gesetzt).

isolation und cwd schließen sich im Agent-Tool gegenseitig aus — hat das Modell selbst ein cwd angegeben, wird das respektiert. „Isolieren oder nicht" ist eine Eigenschaft der Rolle, kein globaler Schalter und keine Entscheidung pro Delegation; einer Rolle, die keine Isolation braucht, wird kein einziges Byte hinzugefügt.

Wer Isolation einschaltet, muss den Workbench aus dem Repo herausziehen. Ein isolierter agent kann nicht in den geteilten checkout schreiben, also muss der Workbench per home= außerhalb des Repos liegen. starter_flow(isolate=True) nimmt <ws>.parent/.flower-<ws.name>, Runtime(workbench=True) nimmt <run_dir>/workbench — beide liegen außerhalb des Repos, sind aber nicht dasselbe Verzeichnis; nicht mischen.

isolated() / wants_isolation()¶

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

Markiert eine subagent-Definition als „braucht eigenen Arbeitsbereich" und liest die Markierung wieder aus.

Die Markierung ist ein per object.__setattr__ gesetztes Python-Attribut _flower_isolate, kein dataclass-Feld — das SDK serialisiert mit asdict() und kennt nur deklarierte Felder, die Markierung leckt also nicht zur CLI durch (gemessen).

Der Preis: ein dataclasses.replace() auf AgentDefinition verliert diese Markierung, die Isolation fällt still aus.

worker(isolate=True) geht intern genau über 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]]

Installiert in einem Zug die hooks, die ein Workbench braucht. Runtime._attempt ruft genau das auf.

Ergebnis:

• PreToolUse: delegate_only=True → [delegate_guard(allow_glance=allow_glance)]; gibt es markierte Rollen → zusätzlich isolate_guard(agents).
• PostToolUse: immer [index_guard(workbench)]; ist spill_threshold truthy → zusätzlich spill_guard(workbench, threshold=spill_threshold).
• Event-Schlüssel mit leerer Liste werden entfernt, es wird keine leere list zurückgegeben.

merge_hooks()¶

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

Fügt mehrere hook-Konfigurationen pro Event-Name aneinander.

Nutzt extend und dedupliziert nicht — derselbe guard zweimal übergeben wird zweimal installiert. Runtime fügt damit spec.hooks, workbench_hooks(...) und whitelist_guard zusammen.

Workbench¶

Quelle: flower/core/workbench.py

Workbench¶

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

Das Arbeitsverzeichnis fürs Spillen: drei Unterverzeichnisse + ein Index. Der Index wird in den System-prompt injiziert, der agent weiß also in jeder Runde, was er zur Hand hat.

Selbstbeschreibungsformat der Skripte: # desc: ein Satz innerhalb der ersten 8 Zeilen (auch // und -- als Kommentarzeichen), Rückfall auf den ersten nicht-leeren Kommentar oder die erste Zeile des docstring (auf 100 Zeichen gekürzt).

Die drei Regeln, die prompt_block() injiziert:

1. Skripte, die ein zweites Mal laufen sollen, kommen nach scripts/, erste Zeile # desc:.
2. Ergebnisse über 2000 Zeichen kommen nach artifacts/, im Dialog stehen nur Pfad und Schlussfolgerung.
3. Zentrale Entscheidungen kommen nach notes/, eine Datei pro Entscheidung.

Bei external=True ergänzt prompt_block() zusätzlich den Satz „greif mit absolutem Pfad darauf zu".

Den Index erben subagents nicht. Er läuft über das session-weite system_prompt.append, subagents haben aber ihren eigenen System-prompt (gemessen $0.2461). Die beiden Punkte „lange Ergebnisse nach artifacts/" und „wo der Workbench liegt" muss der Coordinator deshalb im Task Brief weitergeben — das ist der einzige Kanal, keine Redundanz. In WORKER_RULES steht das absichtlich nicht: der echte Pfad wird von Workbench erzeugt, fest verdrahtet wäre er falsch.

Session Store¶

Quelle: sqlite.py · trim.py · prune.py

Drei Vererbungsstufen: SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore. Runtime benutzt immer die äußerste, die Strategien aller drei Schichten werden über Konstruktorparameter gesteuert.

Jede Schicht macht eine Sache: Persistieren, Trimmen nach Größe und Wert, Prunen nach „ist das ein Fehler". Trimmen und Prunen passieren beide im Moment von load() (also wenn resume die Historie zurück ins Modell füttert); an den Rohdaten in SQLite ändert sich kein Byte.

SqliteSessionStore¶

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

Implementiert das SessionStore-Protokoll des SDK, drei Tabellen entries / meta / summaries. Der store key ist project_key/session_id[/subpath] — die transcripts von sub-agents werden über subpath unterschieden.

Das interne _next_mtime garantiert strenge Monotonie — list_sessions und die summary-sidecar teilen sich diese Uhr, sonst urteilt der staleness-Schnellpfad des SDK falsch.

TrimPolicy¶

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

Die beiden spill-Verzeichnisse sind nicht dasselbe. spill_guard landet in <workbench.root>/spill/ (darf außerhalb des Arbeitsbereichs liegen); TrimPolicy.spill_dirname landet in <workspace>/.flower/spill/ (muss innerhalb des Arbeitsbereichs liegen). Die beiden entsprechen „sofort kürzen" und „beim resume kürzen"; die Trennung ist Absicht, nicht zusammenlegen.

EphemeralPolicy¶

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

Die Verfallsstrategie für Ergebnisse von ephemeral commands.

Wirkt nur auf Ergebnisse des Bash-Tools, und das Kommando muss der Whitelist der ephemeral commands entsprechen. Read gehört nicht dazu — Dateiinhalte veralten nicht in einem Maß, das in die Irre führt. Veralteter Inhalt wird nicht gespillt, sondern direkt weggeworfen.

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

Entscheidet, ob ein Bash-Kommando ein ephemeral command ist. Die Durchlassentscheidung von delegate_guard und die Verfallsentscheidung beim Trimmen teilen sich diese eine Funktion — die Menge der Kommandos, die der Coordinator selbst ausführen darf, muss der Menge entsprechen, deren Ergebnisse als veraltet markiert werden. Durchlassen ohne Trimmen: ein veraltetes git status besetzt dauerhaft Kontext und führt zusätzlich in die Irre; Trimmen ohne Durchlassen: der Coordinator schickt für ein ls einen subagent, 4.3k Startkosten für ein paar Dutzend Zeichen.

Prüfreihenfolge:

1. leer / nur Whitespace → False.
2. Kommandosubstitution ($(, Backticks, <(, >() oder eine zustandsändernde Schreibweise trifft zu → False.
3. nach Abzug sicherer Umleitungen (2>&1, &> /dev/null u. ä.) immer noch > oder < enthalten → False.
4. nach Abzug von && / || / ; / | bleibt ein einzelnes & übrig (Hintergrundausführung) → False.
5. an && / || / ; / | zerlegen, jedes Segment muss die Whitelist treffen.

Die Verbgruppen der Whitelist: nur lesende git-Subkommandos (status diff log show branch rev-parse usw.), Verzeichnis- und Systeminfos (ls pwd df du date whoami env usw.), Prozesse und Container (ps top lsof docker ps kubectl get usw.), Dateien ansehen (cat head tail wc stat find tree), Pfade suchen (which whereis command -v type), Textverarbeitung (grep rg sort uniq awk sed jq diff usw.).

Auch wenn das Verb auf der Whitelist steht, werden diese Schreibweisen abgefangen: xargs, exec, eval, source, tee, find -delete / -ok / -fprint, sed -i, sort -o, system( und print > innerhalb von awk, git branch -D/-d/-m, git * --force/--hard/--prune.

Die erste Version hat pauschal alle zusammengesetzten Kommandos abgelehnt, damit fiel glance in der Messung komplett aus (alle drei Versuche des Coordinators wurden geblockt), deshalb die segmentweise Prüfung.

TrimmingSessionStore¶

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

Öffentliche Attribute: workspace, policy, ephemeral, last_report: dict[str, int].

Reihenfolge in load(): super().load() → last_report leeren → bei ephemeral.enabled expire() → bei policy.enabled trim(). Bei enabled=False entfällt der jeweilige Schritt komplett.

Getrimmt wird nur reiner Text: image- / document-Blöcke bleiben unverändert stehen.

Zwei strukturelle rote Linien: der tool_result-Block selbst muss bleiben, nur der content darf ersetzt werden (fehlt einer, gibt es „Missing Tool Result Block"); isCompactSummary-Einträge dürfen nicht angefasst werden.

trim_report()¶