Python API¶

이 페이지는 flower 최상위 __all__의 공개 심볼 62개를 남김없이 다룬다: 시그니처, 파라미터, 기본값, 의미, 공개 속성과 메서드. 다 읽고 나면 파라미터를 찾으려고 소스를 열 일이 없다.

구성은 모듈 파일이 아니라 관심사 기준이다 —— "협조자가 직접 손대는 걸 어떻게 막나"가 궁금하면 hook 층으로, "이전 단계의 결과를 다음 단계에 어떻게 넘기나"가 궁금하면 워크플로로 가면 된다. 용어는 전부 용어집을 따른다.

버전 0.1.0, 의존성 claude-agent-sdk>=0.2.152. 모든 시그니처는 소스와 글자 그대로 일치한다.

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

이 페이지에 있는 것¶

물릴 수 있는 기본값 여섯 개¶

이 여섯 줄은 자잘한 곁가지가 아니라 가장 흔한 여섯 번의 사고다. 각 항목은 해당 절에서 온전히 설명한다.

런타임¶

소스: flower/core/runtime.py

Runtime은 실행 코어다. 작업 공간, session store, workbench, resilience 정책, handoff 정책을 들고 있고, 밖으로 내주는 동사는 하나뿐이다: step 하나를 run. 재시도, 중단 후 이어 달리기, 컨텍스트가 꽉 차서 handoff 하기 —— 전부 이 호출 하나 안에서 끝난다.

Runtime¶

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

생성자 파라미터는 전부 keyword-only이고(*가 맨 앞), workspace는 필수다.

session store는 하드코딩되어 있다: 언제나 PruningSessionStore(run_dir/"sessions.db", workspace=..., policy=<TrimPolicy>, ephemeral=<EphemeralPolicy>, prune=PrunePolicy(keep_denials=...)). 생성자 파라미터는 백엔드를 갈아 끼우는 입구를 제공하지 않는다 —— 바꾸려면 직접 AgentSpec + build_options(session_store=...)를 구성하거나, 생성한 뒤 rt.store를 덮어써야 한다.

생성의 마지막 두 단계는 load_dotenv()와 check_credentials()이고, 후자는 문제가 있으면 raise RuntimeError 한다. 자격 증명이 없으면 run()까지 가지 않고 생성 단계에서 터진다.

공개 속성¶

클래스 상수: INTERRUPTED = "interrupted-by-human", HANDOFF_DUE = "context-full-handoff", INTERRUPT_NOTE(중단 후 이어 달릴 때 사람의 말 뒤에 붙는 문단으로, "당시 비행 중이던 도구 호출이 interrupted로 돌아오는 것은 중단의 정상적인 부작용이지 환경 장애가 아니다"라고 설명한다).

공개 메서드¶

Runtime.run(...)¶

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

step 시작 때마다 컨텍스트 수위를 0으로 되돌린다(self._ctx, self._warned = 0, False). 그다음은 루프이고, 출구는 네 개다:

1. 성공 → 빠져나온다.
2. 사람의 중단(result.error == INTERRUPTED)→ max_attempts의 제약을 받지 않고, 네트워크도 기다리지 않는다. 사람의 말을 들고 같은 session을 resume 하며, attempt -= 1(중단은 실패 시도로 치지 않는다), prompt = 사람의 말 + INTERRUPT_NOTE. session_id를 못 받았으면 멈추는 수밖에 없다.
3. 컨텍스트 포화(result.error == HANDOFF_DUE, 또는 handoff.enabled이고 session_id를 받았고 is_overflow(...)가 걸린 경우)→ 역시 max_attempts의 제약을 받지 않는다. 먼저 len(result.retired) >= handoff.max_generations를 확인해 초과하면 error를 진단 문장으로 바꾸고 빠져나온다; 아니면 handoff document를 쓰고 → resume=None, fork=False (완전히 새 session) → prompt를 h.prompt_block()으로 교체 → 수위 0으로 → attempt -= 1.
4. 재시도 가능한 장애 → not resilience.enabled or attempt >= max_attempts면 빠져나온다; classify(error)가 재시도할 게 아니라고 판정해도 빠져나온다; 아니면 Event("retry")를 내보내고, wait_online()으로 네트워크를 기다리고, sleep(delay_for(attempt)); session_id를 한 번이라도 받았으면 resume으로 이어 달리고(prompt는 resilience.resume_prompt로 교체), result.resumed를 True로 둔다.

마무리: ended_at을 쓰고, self.results에 append하고, manifest.json을 쓴다.

manifest.json은 append 의미론이다: 쓸 때마다 디스크를 다시 읽고 run 필드로 중복을 제거한다(자기 줄은 새것으로 교체, 남의 줄은 그대로 둔다). 그래서 같은 run_dir 아래에서 flower 두 개를 병렬로 돌려도 안전하다 —— run_id가 충돌하지 않는다는 전제에서.

handoff의 관측 지점 세 개(모두 Event("handoff")이고 payload["phase"]로 구분): near(warn_at에 근접, 세대당 한 번만 발생), writing(handoff document를 쓰는 중, 십몇 초 걸린다), done(payload에 degraded / path / sections 포함). handoff document를 쓰는 그 턴은 replace(spec, max_budget_usd=None)로 돌린다 —— handoff는 반드시 써 낼 수 있어야 하고, 예산에 걸려 멈추면 안 된다; 또한 on_event=None이라 이 턴은 UI로 흘리지 않는다.

handoff는 <workbench.notes>/交接-<步骤名>.md에 spill된다; workbench가 없으면 spill하지 않고, 문서는 그래도 prompt를 통해 인계받는 쪽에 전달되며, 다만 나중에 찾아볼 수 없을 뿐이다. 예전 handoff는 notes/archive/交接/<名>-<时间戳>.md로 옮겨진다.

StepResult¶

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

step 하나를 끝낸 전체 장부.

워크플로¶

소스: flower/workflow/

워크플로는 순서대로 이어붙인 스텝들의 묶음이고, 여기에 스텝 사이에서 상태를 어떻게 넘길지, 언제 조기 종료할지가 더해진다. 프레임워크는 완성된 워크플로를 제공하지 않는다. 워크플로는 당신이 쓰는 것이다 —— starter_flow는 그저 굴러가는 견본일 뿐이다.

타입 별칭 Ctx = dict[str, Any](flower.workflow.base.Ctx, flower.workflow.__all__에는 있지만 최상위 __all__에는 없다).

Step¶

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

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

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

한 스텝의 선언이다. Step 자체는 함수가 아니다 —— 실제로 실행하는 것은 Runtime.run(step.spec, prompt, ...)다. 앞의 세 필드는 위치 인자라서 Step("取词", terse, "读 seed.txt …") 같은 표기도 유효하다.

session을 잇는 세 가지 방식(같은 run 안에서):

on_reject는 재시도 의미를 바꾼다:

• 주지 않으면 → 다음 시도는 처음부터 다시 돌린다(같은 prompt, 같은 resume_from).
• 주면 → 다음 시도는 방금 기각된 그 session을 이어서 돌리고, prompt는 그 반환값으로 바뀌며, fork는 강제로 False가 된다.
• 빈 문자열을 반환하면 → 되돌려보내지 않고, 처음부터 다시 돌리는 쪽으로 퇴화한다.
• result.session_id가 None이면 → 이 역시 처음부터 다시 돌리는 쪽으로 퇴화한다.

on_fail의 세 가지 값:

통과했든 아니든 ctx["_results"][name] = result는 항상 기록된다. result.session_id가 비어 있지 않으면 ctx["_sessions"]에도 쓰고 lineage.remember(...)도 호출한다.

Workflow¶

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

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

Step들을 순서대로 돌리고 최종 ctx를 반환한다. steps는 위치 인자이므로 Workflow([...])도 유효하다.

run()이 ctx에 쓰는 비공개 키(전부 _로 시작하므로 스텝 이름과 충돌하지 않는다):

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

재시도 라벨: 0번째는 step.name을 쓴다. 그 이후로 on_reject가 있으면 f"{name}#round{attempt+1}", 없으면 f"{name}#retry{attempt}". manifest만 봐도 이 스텝이 어떻게 끝났는지 한눈에 보인다. 접미사가 붙은 이름은 프로세스 간 lineage에 들어가지 않는다 —— Lineage.remember는 원래 이름을 쓴다.

runtime.on_session은 runtime.run 그 한 줄만 감싸고, try/finally로 gate 이전에 반드시 떼어낸다. prompt_cur / resume_cur / fork_cur는 지역 변수이고 step에 되쓰지 않는다 —— 같은 Step 객체가 두 번 돌 수 있기 때문이다.

StepAbort¶

class StepAbort(Exception): ...

gate가 이걸 던지면 = 즉시 멈추고, 더 재시도하지 마라. "False 반환"과의 차이: False는 "이번엔 안 됐으니 한 번 더"이고, StepAbort는 "다시 해도 소용없다"이다.

던져진 뒤: ctx["_aborted"] = str(exc), passed = False, 재시도 루프를 빠져나온다(남은 retries를 소모하지 않는다). 그다음은 일반 실패와 똑같이 on_fail(기본 "stop")을 따른다.

with_goal은 두 곳에서 이걸 던진다: ctx["_runtime"]을 얻지 못할 때, 그리고 verdict가 unreachable인데 아무도 응답하지 않을 때.

clarify_step()¶

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

clarify를 수행하는 Step을 만든다: 요구를 캐물어 명확히 한다 → Brief로 파싱한다 → 네 항목이 다 갖춰지면 동결해 파일로 떨군다.

만들어진 Step의 각 필드는 이렇게 채워진다:

• resume_prompt = CLARIFY_RESUME.
• when: always_ask=True → 항상 True. 아니면 Brief.load(brief_path)가 온전하면 그것을 ctx에 넣고 그다음 False(건너뜀)를 반환한다 —— 건너뛰더라도 넣어야 한다. 아니면 하위 스텝이 요구를 받지 못한다.
• gate: Brief.parse(result.text), 온전하지 않으면 → ctx[MISSING_KEY]를 쓰고 False 반환. 온전하면 → b.write(brief_path)로 동결하고 ctx에 넣은 뒤 True 반환.
• reduce: ctx[BRIEF_KEY].prompt_block()을 반환한다. 모델 원문이 아니다 —— 원문에는 모델이 더 써 넣은 것이 섞여 있을 수 있다.
• resume_from은 기본값 None을 유지한다: 다음 스텝은 새 session이고, brief만 받을 뿐 그 문답은 받지 못한다. clarify의 문답은 coordinator의 컨텍스트에 애초에 들어간 적이 없다. 들어갔다가 잘려 나간 것이 아니다.

ctx에 넣는 곳은 세 군데다: ctx[BRIEF_KEY] = b, ctx[name] = b.prompt_block(), ctx.pop(MISSING_KEY, None).

goal_step()¶

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

목표를 설정하는 Step을 만든다: judge가 brief를 읽고 목표 + 판정 체크리스트를 쓰게 하고, Goal로 파싱한 뒤 동결해 파일로 떨군다. clarify_step과 같은 형태다.

can_run 형식 인자는 없다 —— 목표를 세우는 judge가 명령을 실행할 수 있게 하려면 **spec_kw로 can_run=True를 넘기는 수밖에 없다. 넘기지 않으면 Bash를 얻지 못하고, JUDGE_RULES의 "먼저 네가 어떤 환경에 있는지 똑똑히 봐라"라는 항목을 수행할 수 없다.

gate는 파싱과 동결 외에 한 가지를 더 한다: 목표 안에 [此环境无法验证:…] 항목이 있으면, 그 자리에서 ctx["_on_event"]를 통해 Event("task", payload={"unverifiable", "total", "path"})를 보내 알린다 —— 이 항목들의 운명은 목표를 세우는 이 순간에 이미 정해지고, 판정 시점까지 가면 이미 한 라운드 작업 비용을 다 써 버린 뒤다.

resume_prompt는 설정하지 않는다 —— 목표 설정은 애초에 brief 전문을 다시 보내야 하는 일이다.

with_goal()¶

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

기존 Step에 goal guard를 씌운다: 매 라운드가 끝나면 judge가 독립적으로 판정하고, 달성되지 않았으면 되돌려보내 계속 하게 한다.

결과물은 replace(step, retries=max(0, rounds - 1), gate=<새 gate>, on_reject=<새 on_reject>)다 —— 필드를 하나씩 재구성하지 않고 dataclasses.replace를 쓴다. 한 번 재구성했다가 resume_prompt를 빠뜨린 적이 있는데, 에러도 나지 않았다. 그저 이어서 갈 때 brief 전문을 통째로 다시 보냈을 뿐이다.

gate는 async이고, 흐름은 이렇다:

1. ctx["_runtime"]이 없으면 → StepAbort를 던진다("Runtime을 얻을 수 없어 목표를 판정할 수 없다"). 통과한 척하지 마라.
2. ctx[ROUND_KEY] += 1.
3. 목표를 가져온다: ctx[GOAL_KEY]에 온전한 Goal이 있으면 그것, 아니면 Goal.load(goal_path), 그것도 아니면 빈 Goal().
4. await rt.run(judger, VERIFY_PROMPT..., step_name=f"{label}#{轮次}", on_event=...). judge는 독립된 한 번의 Runtime.run이고, resume은 항상 None이다 —— 언제나 새 session이다. step_name에 라운드가 붙으므로 프로세스 간 lineage에는 들어가지 않는다.
5. Verdict.parse(vr.text)를 ctx[VERDICT_KEY]에 쓴다.
6. v.achieved면 → True 반환.
7. unreachable이 아니면(v.ok=False인 모호한 경우 포함) → 모호할 때는 기본 reason을 한 줄 채우고 False 반환. 모호하면 무조건 미달성으로 본다 —— "될 것 같다" 한마디로 일을 끝내게 둘 수는 없다.
8. unreachable이면 → await channel.ask(...)로 사람에게 묻는다. 선택지는 셋:
9. 아무도 응답하지 않으면(a.state != "answered") → StepAbort를 던진다. 계속 헛돌게 두는 것이 가장 비싼 선택이다.
10. 「이 결과를 받아들이고 이대로 계속 간다」 → True 반환.
11. 「목표를 수정한다」 → 새 목표를 한 번 더 묻고, g.amend(...).write(goal_path)로 쓰고 ctx[GOAL_KEY]를 갱신한 뒤 False 반환.
12. 그 외(사람이 직접 쓴 자유 답변 포함) → "네 판정이 틀렸다"로 보고, 사람의 말을 v.reason에 기록한 뒤 False 반환.

on_reject는 동기다: ctx[VERDICT_KEY].feedback()을 반환하고, Verdict가 없으면 ""를 반환한다(처음부터 다시 돌리는 쪽으로 퇴화한다).

starter_flow()¶

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

바로 굴러가는 세 스텝 워크플로를 조립한다: 요구 확인 → 목표 설정 → 작업(goal guard 포함). 커맨드라인 flower가 쓰는 것이 바로 이것이다.

고정된 조립:

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

동작 분기:

• isolate=True인데 workspace가 git 저장소가 아니면 → ValueError를 던진다. Agent 툴이 에러를 낼 때까지 기다리지 않는다(그때는 이미 돈을 썼다).
• wake 판정: Brief.load(brief_path)가 존재하고 complete()이면 wake로 본다. wake가 아닌데 ask가 비어 있으면 → ValueError("要给一句诉求,例如 flower '帮我做一个 X'")를 던진다.
• wake일 때 그 한마디는 동시에 세 곳에 떨어진다. 하나라도 빠지면 조용히 무효가 된다: brief에 덧붙이기(ch.amend(said, label="唤醒时追加"), 이미 파일에 있으면 중복해서 쓰지 않는다), goal_step(always_set=True)으로 체크리스트를 다시 도출하기(다시 도출하지 않으면 judge가 읽는 것은 여전히 옛 목표다), 그리고 coordinator 앞에 직접 보내기(coordinator의 컨텍스트에는 옛 목표가 들어 있어서, 주지 않으면 옛 기준으로 일하고 새 기준으로 판정받는다).

wake_state()¶

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

출발 전에 하는 읽기 전용 탐지로, 한 바이트도 쓰지 않는다. 실제로 돌리기 전에 "이번이 지난번을 잇는 것인지, 처음부터인지"를 사람에게 알려 주는 데 쓴다.

반환되는 dict:

workbench 위치는 여기와 starter_flow 두 곳에서만 정의된다: isolate=True → <ws>.parent/.flower-<ws.name>(저장소 바깥), 아니면 <ws>/.flower. 드라이버가 brief 위치를 알고 싶을 때도 이 함수를 거친다 —— 직접 경로를 조합하다 틀려도 에러가 나지 않고, 그저 조용히 무효가 된다.

역할 팩토리¶

소스: flower/core/roles.py

다섯 역할은 모두 팩토리 함수다. 각 역할 = 주입되는 규칙 텍스트 한 편 + 도구 한 묶음 + hook 한 묶음. worker()가 내놓는 것은 SDK의 AgentDefinition(subagent로 파견할 때 쓴다)이고, 나머지 넷은 AgentSpec을 내놓는다(자기 session을 따로 연다).

역할 자체는 hook을 달지 않는다 —— 도구를 가로채는 일은 Runtime._attempt가 spec.delegate_only에 따라 자동으로 붙인다. hook 층을 보라.

내부 도구 묶음 상수(export되지 않지만 기본값을 결정한다):

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

main thread 위의 그 coordinator를 만든다: 작업을 쪼개고, 일을 파견하고, 보고를 읽고, 결정을 내린다. 단 직접 손을 대지는 않는다. 앞의 세 인자는 위치 인자다.

내놓는 AgentSpec에서 고정된 세 항목: delegate_only=True, agents=workers, workbench는 AgentSpec의 기본값 True 유지.

workers의 web 도구는 병합된다¶

목록을 만든 뒤 coordinator()는 각 AgentDefinition.tools를 순회하면서 WEB_TOOLS(WebFetch, WebSearch, roles.py:33)에 속하는 것은 coordinator 자신의 allowed_tools에도 한 부 추가한다(roles.py:523-526).

이유: allowed_tools도 disallowed_tools와 마찬가지로 session 단위다. 이것이 이 점에 대한 문서 전체에서 가장 단단한 증거다 —— 이 필드는 main thread에만 영향을 주지 않는다. 이 session 단위 목록에 없는 도구는 subagent가 호출할 때도 권한 승인을 거쳐야 한다. 무인 실행 중에는 승인할 사람이 없으니 harness가 Claude requested permissions to use X, but you haven't granted it yet (toolDenialKind=user-rejected)라고 답하고, 모델은 같은 호출을 반복해서 재시도한다. 실제로 넘어진 적이 있다: worker에게 WebFetch/WebSearch를 줬는데 AgentDefinition.tools에만 적었더니, 그 novel run은 스무 번 넘게 user-rejected를 받고 한 글자도 쓰지 못했다(roles.py:513-518).

두 필드의 session 단위 성질은 같지만 증상이 다르다: disallowed_tools는 그 자리에서 에러를 내고, allowed_tools는 조용히 죽을 때까지 재시도한다. 후자가 더 찾기 어렵다. 화면상 아무것도 에러처럼 보이지 않기 때문이다.

읽기 전용, 부작용 없는 것만 병합한다. Write/Edit/Bash는 일부러 병합하지 않는다: main thread가 그것들에 대해 승인 면제를 받는 순간 delegate_guard의 "coordinator는 손대지 않는다"는 벽이 무의미해진다. 그리고 subagent의 Bash/Write는 원래부터 통과된다(실측 462회 허용, roles.py:520-522).

소스에는 disallowed_tools로 "조율만 하고 손대지 않기"를 구현하지 말라고 명시되어 있다 —— 그건 session 단위라 subagent의 Bash/Write까지 함께 금지한다. AgentSpec의 경고를 보라. 올바른 방법은 여기서 하듯 delegate_only=True + allowed_tools에 넣지 않기, 그리고 delegate_guard가 agent_id로 main thread만 가로막게 하는 것이다.

channel을 주면 두 도구가 함께 온다. 선택이 아니다: MCP server를 달면 둘 다 붙고, allowed_tools는 배타적이지 않으므로 목록에 넣든 안 넣든 호출된다. 무인 실행에서는 ask 한 번마다 timeout_s를 꽉 채워 멈춘다 —— 그런 상황에서는 HumanChannel(timeout_s=0)을 쓴다.

worker()¶

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

실제로 일하는 subagent 정의를 만든다. 앞의 두 인자는 위치 인자다. 반환값은 SDK의 AgentDefinition이며, 그대로 coordinator(workers={...})에 넣는다.

isolate=True는 workspace가 git 저장소일 것을 요구한다. 아니면 Agent 도구가 바로 "not in a git repository"를 뱉는다. 조용히 퇴화하지 않는다. 게다가 이 표시는 Python 속성이다 —— AgentDefinition에 dataclasses.replace()를 하면 표시가 사라져 isolation이 조용히 무효화된다.

clarify()¶

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

clarifier를 만든다: 손대기 전에 요구사항을 묻고 확실히 한다. 일은 하지 않고 질문만 하며, 마지막에 정확히 네 단락을 출력한다.

내놓는 AgentSpec: allowed_tools = [channel.tool_name] + (읽기 가능할 때 그 다섯 개), mcp_servers = channel.mcp_servers(), workbench=False(쓰기 도구가 없으니 인덱스가 의미 없다), permission_mode는 AgentSpec의 기본값 "default"를 그대로 쓴다. Write / Edit / Bash / Agent가 없고, inbox도 없다(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

judge를 만든다: 시작 전에 목표를 세우거나, 매 라운드가 끝난 뒤 그 라운드를 판정한다.

내놓는 AgentSpec: allowed_tools = [channel.tool_name, "Read", "Glob", "Grep"] + (can_run일 때) ["Bash"], workbench=False, 나머지는 clarify()와 같다. Write / Edit / Agent가 없고, inbox도 없다.

트레이드오프: can_run=True면 판정이 더 단단해지지만(검수 명령을 실제로 돌릴 수 있다), 대가로 judge가 워크스페이스를 변경할 수 있게 된다 —— Bash 자체로 파일을 쓸 수 있기 때문이다. 절대적으로 중립적인 판정을 원하면 켜지 마라.

oracle()¶

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

oracle을 만든다: run이 아직 돌고 있는 동안 "지금 어디까지 왔나"를 물으면, 최근 event와 workbench를 한 번 보고 답한다. 그가 하는 말은 그 run의 컨텍스트에 들어가지 않는다.

내놓는 AgentSpec: allowed_tools = ["Read", "Glob", "Grep"](channel 없음 —— 질문하지 않고 답만 한다), workbench=True(다섯 역할 중 coordinator가 아니면서 workbench를 켜는 유일한 역할 —— 산출물과 노트를 읽는 것이 목적이기 때문이다).

다섯 편의 규칙 텍스트¶

다섯 상수는 모두 __all__에 있어 그대로 import해서 읽고, 붙이고, 고칠 수 있다.

agent 정의¶

소스: flower/core/agent.py

AgentSpec은 전문 agent 하나의 완전한 선언이고, build_options는 그것을 SDK의 ClaudeAgentOptions로 컴파일한다. 역할 팩토리가 내놓는 것이 바로 AgentSpec이다 —— 팩토리 밖의 조합이 필요하면 직접 구성하면 된다.

AgentSpec¶

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

build_options()¶

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

AgentSpec을 SDK의 ClaudeAgentOptions로 컴파일한다. Runtime._attempt가 내부에서 호출하는 것이 바로 이것이고, 직접 SDK를 구동할 때(Runtime을 쓰지 않을 때)도 여기로 들어온다.

매핑 관계:

PLUGIN_DIR은 저장소 루트의 plugin/이다(flower/core/agent.py 기준 세 단계 위). pip으로 설치하면 이 디렉터리가 없을 수도 있어서 코드가 is_dir()로 확인한다.

CompactPolicy¶

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

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

auto-compact의 스위치 패널이고, 산출물은 자식 프로세스에 주입할 환경 변수 묶음이다. compact 알고리즘 자체는 harness 바이너리 안에 있어 고칠 수 없고, 고칠 수 있는 것은 "트리거할지 말지"뿐이다.

HandoffPolicy¶

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

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

컨텍스트가 거의 찼을 때 compact 대신 "handoff document를 쓰고 새 session으로 갈아타는" 정책 객체다.

default_window()¶

def default_window() -> int

환경 변수 ANTHROPIC_MODEL 또는 ANTHROPIC_DEFAULT_OPUS_MODEL의 모델 이름 문자열로 컨텍스트 윈도를 추정한다:

기본값은 공격적으로 잡는다. 크게 잡는 것은 치명적 오류가 아니다: API가 prompt is too long으로 되돌려주고, Runtime이 이 신호를 인식해 (내부의 is_overflow) 그 자리에서 handoff한다 —— 다만 그 세대의 handoff는 격하된 판본이다.

문서¶

소스: brief.py · handoff.py · goal.py

네 개의 dataclass이며, 모두 "모델의 회신 한 편을 고정된 몇 단락으로 파싱한 뒤 디스크에 쓰는" 것이다. 공통된 형태: parse()로 파싱, missing() / complete()로 빠짐 여부 확인, to_markdown()은 사람이 볼 것, prompt_block()은 하위 모델이 볼 것, write() / load()는 저장과 되읽기.

Brief¶

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

brief는 정확히 네 단락이며, 순서는 goal → accept → bounds → unknowns로 고정이고, 한국어 단락명은 각각 「목표」「검수 기준」「경계」「미지와 가정」이다.

파싱 규칙(실수하기 쉬운 지점 모음):

• 펜스를 벗길 때 **닫히지 않은 `` 또는_~를 만나면 그 지점부터 전부 버린다** —— 실측상 clarifier가 코드 전체를 회신에 붙여 넣는다. 모델 출력이 잘리면 뒤쪽 단락이 전부 파싱되지 않으므로complete()가False`가 되고, gate가 되돌려 보낸다.
• 제목 정규식은 ## 目标 / **目标** / 目标: / 3. 边界를 허용하고, 제목 바로 뒤에 본문이 오는 것도 허용한다.
• 별칭 표는 길이 내림차순으로 컴파일한다. 그러지 않으면 "未知"가 "未知与假设"를 먼저 먹어버린다.
• 같은 이름의 단락이 여러 번 나오면 내용이 있는 첫 번째 것을 취한다.
• brief를 손으로 편집할 때 to_markdown()의 "(未填)" 자리표시자 문구를 그대로 베껴 넣었다면 그 단락은 여전히 빠진 것으로 친다.

Handoff¶

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

handoff 때 쓰는 handoff document이며 다섯 단락이다.

필수는 doing과 next 두 단락뿐이다 —— "통하지 않은 길"이 비어 있으면 안 된다고 강제하면 모델이 꾸며내게 된다.

같은 모듈 안에 export되지 않지만 의미상 핵심적인 멤버 셋: is_overflow(*texts)는 prompt is too long, context length exceeded, maximum context length, too many total text bytes, input length and max_tokens exceed 등을 매칭해 "치명적 에러"를 "그 자리에서 handoff"로 바꾼다. HANDOFF_PROMPT는 현재 session 자신에게 handoff를 쓰게 하는 프롬프트다({used} {window} 두 자리표시자를 포함하며, 새로운 역할이 아니다 —— 그 컨텍스트를 가진 것은 자기 자신뿐이다). degraded(step, prompt, *, why="")는 handoff를 쓰지 못했을 때 기계적으로 한 부를 조립하며, scene에는 원래 작업의 앞 1200 자를 넣는다.

Goal¶

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

goal guard의 목표 + 판정 목록.

Verdict¶

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

judge의 한 라운드 판정 결과이며 정확히 세 단락이다: 결론 / 이유 / 미통과.

parse의 인식 순서:

1. 먼저 제목 단락으로 「결론」/「판정」을 취한다.
2. 제목 단락이 없으면, 전체를 strip한 뒤 fullmatch(r"1|true") → 달성; fullmatch(r"0|false") → 미달.
3. 그래도 아니면 결론 텍스트에서 상태어 표(긴 단어가 앞)로 첫 번째로 걸리는 단어를 찾는다. 「无法验证 / 没法验证 / 验证不了 / 无法判定 / unverifiable」은 전부 unreachable로 귀속된다 —— 실측으로 넘어진 적이 있다: 목표 플랫폼이 macOS인데 Linux 컨테이너에서 돌았고, judge가 소스의 분기만 보고 통과로 판정했다.
4. 여전히 없으면 → 고립된 \b1\b를 찾으면 달성, \b0\b면 미달.
5. 전부 안 걸리면 → state="", ok=False.

unreachable과 not_yet은 서로 다른 결론이다: 전자는 "멈추고 사람에게 묻는" 길로 가며, "한 라운드 더"가 아니다.

hook 계층¶

소스: flower/core/guard.py

이 계층은 flower의 실행 경계다. 어떤 도구를 main thread가 건드릴 수 없는지, 지나치게 긴 결과를 어떻게 자를지, 어떤 역할을 독립 worktree로 보낼지를 전부 SDK hook이 강제한다. prompt에 의존하지 않는다. 이유는 단순하다 — prompt는 권고일 뿐이고 모델은 무시할 수 있다. 실측에서도 시스템 prompt에 "worktree를 쓰지 마라"라고 명시했음에도 isolate_guard의 주입은 그대로 적용됐다(모델이 넘긴 값은 None, 실제로 반영된 값은 'worktree').

내보내는 것은 아홉 개: HookMatcher를 반환하는 guard 팩토리 다섯 개(whitelist_guard는 None을 반환할 수도 있다), 조립기 하나, 병합기 하나, isolation 표식 함수 둘. 직접 손으로 달 필요는 없다 — Runtime이 AgentSpec에 따라 자동으로 장착한다. 손으로 다는 것은 직접 SDK를 구동할 때 (Runtime을 거치지 않을 때)만 필요하다.

main thread 판정은 함수 하나로 통일한다: _is_main_thread(data) = not data.get("agent_id") —— subagent의 tool-lifecycle hook 데이터에는 agent_id가 실려 있고, main thread에는 없다. "main thread만 막는" 모든 guard가 이 한 줄에 의존한다.

도구 그룹 상수(모듈 수준, 내보내지 않지만 기본 matcher를 결정한다):

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

빠른 조회표: 어떤 guard가 어떤 SDK 이벤트에 붙는가¶

이 표에서 읽어낼 수 있는 핵심 추론: Runtime(workbench=False)이면 workbench_hooks 전체가 장착되지 않는다. 그리고 delegate_only=True인 coordinator에서는 whitelist_guard도 건너뛴다 — main thread에는 벽이 하나도 없다. 자세한 것은 Runtime의 그 경고를 보라.

whitelist_guard()¶

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

allowed_tools가 직접 작업하는 그 네 개 도구에 대해 진짜로 배타적이 되게 한다.

• PreToolUse에 붙는다. matcher는 "|".join(banned)이고, banned = Bash Write Edit NotebookEdit 중 allowed에 없는 것들이다.
• 걸리면 곧바로 permissionDecision: "deny", 문구의 요지: "XX에는 YY가 없다. 이것은 의도된 것이지 설정 누락이 아니다. 결론을 네 답변 본문에 쓰라. 프레임워크는 거기서 가져간다 — 다른 방식으로 우회하려 하지 마라."
• 이 session의 main thread만 막고 subagent는 통과시킨다 — subagent의 도구는 AgentDefinition.tools가 결정한다.
• 막을 도구가 없으면 None을 반환한다(예: worker()처럼 전체 도구를 가진 역할). 호출자는 이를 보고 장착 여부를 결정한다.

왜 반드시 존재해야 하는가: allowed_tools는 승인 면제 목록이지 배타적 화이트리스트가 아니다. 실측 증거 두 건 —— 목표를 정하는 judge가 Bash를 11번 실행했고, $0.1짜리 탐침에서 allowed_tools=["Read"]인 agent가 Write/Bash를 멀쩡히 호출했다. 그래서 clarify() / judge()의 "쓰기 도구가 없다"는 이 hook에 의존하는 것이지 화이트리스트 자체가 아니다.

좋은 점은 allowed_tools에서 파생된다는 것이다. 그래서 judge(can_run=True)는 자동으로 Bash를 남기면서 Write/Edit은 계속 막는다 — 별도 스위치가 필요 없다.

delegate_guard()¶

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

main thread가 직접 손대면 → 거부하고 길을 알려준다.

• PreToolUse에 붙고, matcher는 그대로 tools다.
• main thread가 이 네 도구를 호출하면 → deny, 사유 안에 다음에 무엇을 할지를 준다: Agent 도구로 subagent를 파견하고, 작업에 목표와 인수 기준을 명시하며, 긴 산출물은 .flower/artifacts/에 쓰고 답변에는 경로와 결론만 담게 하라고.
• subagent는 무조건 통과.

whitelist_guard와의 차이는 문구다. 둘은 같은 도구 묶음을 막지만, 이쪽은 "사람을 파견하라"고 말하므로 더 맞다. 그래서 delegate_only=True인 역할은 이 하나만 장착한다. 중복 장착하면 모델이 서로 모순되는 지침 두 개를 받는다.

allow_glance=True의 허용 판정 기준과 "결과가 잘릴 것인가"는 같은 함수(is_ephemeral)다 —— 허용 집합은 만료 집합과 반드시 같아야 하고, 한쪽을 고치면 다른 쪽도 고쳐야 한다.

spill_guard()¶

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

임계치를 넘은 도구 결과를 그 자리에서 spill 하고, 컨텍스트에는 한 줄 포인터만 남긴다 — 컨텍스트가 꽉 찬 뒤에 돌아가서 compact하는 것이 아니다.

• PostToolUse에 붙고, {"hookSpecificOutput": {"hookEventName": "PostToolUse", "updatedToolOutput": <잘린 것>}}을 반환한다.
• spill 파일 이름은 내용의 sha256 앞 16자리 + .txt이고, 컨텍스트에는 한 줄 포인터 + 앞부분 400자로 대체된다.
• updatedToolOutput은 원래 도구의 출력 구조를 유지해야 한다. 그래서 dict 안의 지나치게 긴 문자열 필드만 교체하고, list는 절대 건드리지 않는다(안에 이미지 블록이 있을 수 있다). 구조가 맞지 않으면 거부된다(원문 그대로 유지, 에러는 없음).
• spill 파일 자체를 읽을 때는 반드시 허용해야 한다 — 그러지 않으면 "Read로 그것을 읽어라"는 빈말이 된다. 읽어온 전문이 다시 spill되어 무한 루프에 빠진다. 실측에서 겪었고, 모델이 우회하려고 다섯 가지 방식을 연달아 시도했다.

index_guard()¶

def index_guard(workbench: Workbench) -> HookMatcher

workbench에 무언가를 쓰면 INDEX.md를 갱신해, 다음 agent가 시작하자마자 그것의 존재를 알게 한다.

PostToolUse에 붙고, matcher는 "Write|Edit"이다. tool_input["file_path"]를 resolve한 결과가 workbench.root 안에 있으면 workbench.refresh()를 호출한다. 항상 {}를 반환한다 — 아무것도 바꾸지 않고 부수효과만 있다.

isolate_guard()¶

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

역할별로 subagent에 독립 git worktree를 할당해 isolation을 구현한다.

PreToolUse에 붙고, matcher는 "Agent"다. 세 조건이 동시에 성립할 때만 주입한다: tool_name == "Agent", tool_input에 cwd도 isolation도 없음, subagent_type에 해당하는 역할이 isolated()로 표시됨. 성립하면 permissionDecision: "allow" + updatedInput(isolation을 "worktree"로 설정)을 반환한다.

isolation과 cwd는 Agent 도구에서 상호 배타적이다 — 모델이 스스로 cwd를 지정했다면 그것을 존중한다. "isolation을 할 것인가"는 역할의 속성이지 전역 스위치도 아니고, 파견할 때마다 판단하는 것도 아니다. isolation이 필요 없는 역할에는 단 1바이트도 추가되지 않는다.

isolation을 켜면 workbench를 저장소 밖으로 옮겨야 한다. isolation된 agent는 공유 checkout에 쓸 수 없으므로 workbench는 home=으로 저장소 밖을 가리켜야 한다. starter_flow(isolate=True)는 <ws>.parent/.flower-<ws.name>을, Runtime(workbench=True)는 <run_dir>/workbench를 쓴다 —— 둘 다 저장소 밖이지만 같은 디렉터리가 아니다. 섞어 쓰지 마라.

isolated() / wants_isolation()¶

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

subagent 정의에 "독립 작업 공간이 필요하다"는 표식을 달고, 그 표식을 다시 읽는다.

표식은 object.__setattr__로 붙인 Python 쪽 속성 _flower_isolate이며, dataclass 필드가 아니다 —— SDK는 asdict()로 직렬화하면서 선언된 필드만 인식하므로 이 표식은 CLI 쪽으로 새어 나가지 않는다(실측 완료).

대가: AgentDefinition에 dataclasses.replace()를 하면 이 표식이 사라지고 isolation이 조용히 무효화된다.

worker(isolate=True)는 내부적으로 isolated()를 거친다.

workbench_hooks()¶

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

workbench에 필요한 hook들을 한 번에 장착한다. Runtime._attempt가 호출하는 것이 바로 이것이다.

산출:

• PreToolUse: delegate_only=True → [delegate_guard(allow_glance=allow_glance)]; 표시된 역할이 있으면 → isolate_guard(agents) 추가.
• PostToolUse: 항상 [index_guard(workbench)]; spill_threshold가 참이면 → spill_guard(workbench, threshold=spill_threshold) 추가.
• 빈 리스트인 이벤트 키는 제거된다. 빈 list를 반환하지 않는다.

merge_hooks()¶

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

이벤트 이름별로 여러 hook 설정을 이어 붙인다.

extend를 쓰며 중복을 제거하지 않는다 — 같은 guard를 두 번 넘기면 두 번 장착된다. Runtime은 이것으로 spec.hooks, workbench_hooks(...), whitelist_guard를 합친다.

workbench¶

소스: flower/core/workbench.py

Workbench¶

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

spill을 위한 작업 디렉터리. 하위 디렉터리 세 개 + 인덱스 한 부. 인덱스는 system prompt에 주입되므로 agent는 매 턴 손에 무엇이 있는지 안다.

스크립트 자기소개 형식: 첫 8줄 안의 # desc: 한 줄 설명(//와 -- 주석 기호도 인식), 없으면 첫 번째 비어 있지 않은 주석 줄 또는 docstring 첫 줄로 폴백(100자에서 자름).

prompt_block()이 주입하는 세 가지 규칙:

1. 두 번째로 실행할 스크립트는 scripts/에 쓰고, 첫 줄에 # desc:를 넣는다.
2. 2000자를 넘는 산출물은 artifacts/에 쓰고, 대화에는 경로와 결론만 준다.
3. 핵심 결정은 notes/에 쓰고, 결정 하나에 파일 하나로 한다.

external=True이면 prompt_block()이 "절대 경로로 접근하라"는 문장을 하나 더 삽입한다.

인덱스는 subagent에게 상속되지 않는다. 이것은 session 수준의 system_prompt.append를 타는데, subagent는 자기 자신의 system prompt를 갖는다(실측 $0.2461). 그래서 "긴 산출물은 artifacts/에, workbench는 어디에" 이 두 가지는 coordinator가 task brief에 옮겨 적어야 한다 — 그것이 유일한 통로이고, 중복이 아니다. WORKER_RULES에는 일부러 이것을 쓰지 않았다: 실제 경로는 Workbench가 생성하므로 하드코딩하면 틀린다.

session store¶

소스: sqlite.py · trim.py · prune.py

세 겹 상속: SqliteSessionStore ← TrimmingSessionStore ← PruningSessionStore. Runtime은 항상 가장 바깥층을 쓰고, 세 층의 정책은 생성자 파라미터로 제어한다.

세 층은 각각 한 가지씩 맡는다: 디스크 기록, 부피와 가치에 따른 trim, "에러인가"에 따른 prune. trim과 prune은 모두 load() 시점(즉 resume가 이력을 모델에 다시 먹이는 순간)에 일어나며, SQLite 안의 원본 기록은 1바이트도 바뀌지 않는다.

SqliteSessionStore¶

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

SDK의 SessionStore 프로토콜을 구현한다. 테이블 세 개: entries / meta / summaries. store key는 project_key/session_id[/subpath] — 하위 agent의 transcript는 subpath로 구분한다.

내부의 _next_mtime이 엄격한 단조성을 보장한다 — list_sessions와 summary sidecar가 이 시계를 공유하며, 그렇지 않으면 SDK의 staleness 빠른 경로가 오판한다.

TrimPolicy¶

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

두 spill 디렉터리는 같은 것이 아니다. spill_guard는 <workbench.root>/spill/에 쓴다(작업 공간 밖일 수 있다). TrimPolicy.spill_dirname은 <workspace>/.flower/spill/에 쓴다(반드시 작업 공간 안이어야 한다). 둘은 각각 "그 자리에서 자르기"와 "resume 시 자르기"에 대응하며, 디렉터리가 다른 것은 의도된 것이다. 하나로 합치지 마라.

EphemeralPolicy¶

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

ephemeral command의 결과 만료 정책.

Bash 도구의 결과에만 적용되며, 명령이 ephemeral command 화이트리스트에 맞아야 한다. Read는 여기에 포함되지 않는다 —— 파일 내용은 시간이 흐른다고 오도할 정도로 어긋나지 않는다. 만료된 내용은 spill하지 않고 그냥 버린다.

is_ephemeral()¶

def is_ephemeral(cmd: str) -> bool

Bash 명령 하나가 ephemeral command인지 판정한다. delegate_guard의 허용 판정과 trim의 만료 판정이 이 함수 하나를 공유한다 — coordinator가 직접 실행할 수 있는 명령 집합은 결과가 만료로 표시되는 집합과 반드시 같아야 한다. 허용하고 trim하지 않으면 만료된 git status가 컨텍스트를 영구히 점유하면서 오도하고, trim하면서 허용하지 않으면 coordinator가 ls 하나 때문에 subagent를 파견해 4.3k의 기동 비용으로 수십 자를 얻는다.

판정 순서:

1. 비었거나 전부 공백 → False.
2. 명령 치환($(, 백틱, <(, >() 또는 "상태를 바꾸는" 표현이 걸리면 → False.
3. 안전한 리디렉션(2>&1, &> /dev/null 류)을 제거한 뒤에도 > 또는 <가 남으면 → False.
4. && / || / ; / |를 제거한 뒤에도 단독 &가 남으면(백그라운드 실행) → False.
5. && / || / ; / | 기준으로 쪼개서 모든 구간이 화이트리스트에 걸려야 한다.

화이트리스트 동사 대분류: 읽기 전용 git 하위 명령(status diff log show branch rev-parse 등), 디렉터리 및 시스템 정보(ls pwd df du date whoami env 등), 프로세스 및 컨테이너 (ps top lsof docker ps kubectl get 등), 파일 보기(cat head tail wc stat find tree), 경로 조회(which whereis command -v type), 텍스트 처리(grep rg sort uniq awk sed jq diff 등).

동사가 화이트리스트에 있어도 다음 표현들은 막힌다: xargs, exec, eval, source, tee, find -delete / -ok / -fprint, sed -i, sort -o, awk 안의 system(과 print >, git branch -D/-d/-m, git * --force/--hard/--prune.

첫 버전은 복합 명령을 일괄 거부했는데, 실측 결과 glance가 완전히 무력화됐다(coordinator의 세 번의 시도가 전부 막혔다). 그래서 구간별 판정으로 바꿨다.

TrimmingSessionStore¶

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

공개 속성: workspace, policy, ephemeral, last_report: dict[str, int].

load()의 순서: super().load() → last_report 비우기 → ephemeral.enabled면 expire() → policy.enabled면 trim(). enabled=False면 그 단계 전체를 건너뛴다.

순수 텍스트만 자른다: image / document 블록을 만나면 그대로 남긴다.

구조적 레드라인 두 가지: tool_result 블록 자체는 반드시 있어야 하고 content만 바꿀 수 있다(하나라도 빠지면 "Missing Tool Result Block"이다). isCompactSummary 항목은 건드릴 수 없다.