Flow Development Guide

This document is a complete development reference for the Flow subsystem. A Flow is Familiars' "multi-agent workflow engine": a main agent track serves as the backbone, with several worker tracks launched via cross-role jumps that can jump back, making it suitable for cross-role collaborative tasks.

1. Concept Model

1.1 What is a Flow

Flow is a multi-agent workflow whose backbone is a "main agent track":

[Starter] ─ Phase1 ─ Phase2 ─ … ─ PhaseN ─ [Summary]
     │ jump ▲ jump
     ▼      │
[子 agent 轨道 A：unit ─ unit ─ jump back to main]

Main track (main): always begins with a Starter module and ends with a Summary module, with several Phases (checkpoints) in between;
Worker tracks: launched via the main track's jump; after each worker finishes its own sequence of units, it jumps back to some anchor on the main track via jump;
Trigger: between two jumps the main track can wait for a condition (for example, a worker completing).

1.2 Template vs Instance

FlowTemplate (authoring time): the static graph written by the author plus the unit/checkpoint/trigger definitions. The on-disk file lives at workspace/manager/flows/<template_id>/template.json.
FlowInstance (runtime): a single concrete execution. The instance pins the template's current version at startup (pinned_template_json), so later changes by the author do not affect this instance.
Mutation log: runtime rewrites of the graph (adding units, rewiring, expand_repeat, etc.) are recorded as mutations in the SQLite flow_mutations table; on resume, the mutations are replayed with hash verification to guarantee a consistent landing state.

1.3 The Three Invariants (engine.py)

Only the owner writes. Workers only post messages; the owner folds each message into FlowState and applies mutations. The sole exception is role status, which the Scheduler writes under its own lock (the agent_state predicate must be observable immediately).
The graph is frozen, not torn. rewrite.apply_spec copies + validates, then the owner atomically swaps references. Workers hold a complete, self-consistent graph that may be stale but is never half-rewritten.
Staleness is caught, not prevented. Workers compare the generation before each unit and reseat when stale. Mutations initiated by a worker are acknowledged synchronously, so the requester never walks through a node it just asked to be rewritten.

2. Directory and Files

2.1 Source Tree

src/flow/
├── engine.py             # FlowEngine：owner thread + 每 role worker thread
├── resume.py             # rehydrate()：从 mutation log + pinned template 重建 engine
├── rewrite.py            # apply_and_validate：mutation → 新图（单源真相）
├── state.py              # FlowState：run state（dotted-path 写）
├── bus.py                # FlowEventBus
├── cancel.py             # CancelToken + Cancelled
├── compile.py            # 模板 → 运行时图编译
├── trace.py              # FlowTrace（事件流）
├── scheduler.py          # RoleCursor + Scheduler
├── template_store.py     # 模板磁盘 store
├── packaging_cli.py      # CLI 打包
├── adapters/             # legacy pipeline 适配
├── bus.py                # 事件总线
├── checkpoints/          # checkpoint 处理器
├── expr/                 # 表达式求值（AST 白名单）
├── models/               # ★ 数据模型
│   ├── template.py       # FlowTemplate / RendererDecl / FLOW_SCHEMA_VERSION
│   ├── graph.py          # FlowGraph + graph_hash（canonical）
│   ├── instance.py       # FlowInstance
│   ├── mutation.py       # MutationSpec
│   ├── unit.py           # WorkUnit
│   ├── workflow.py       # AgentWorkflow
│   ├── checkpoint.py     # Checkpoint (V1)
│   ├── checkpoint_v2.py  # CheckpointV2 + JumpTarget
│   ├── trigger.py        # Trigger
│   ├── event.py          # FlowEvent / FlowEventType
│   ├── field_spec.py     # FieldSpec（IO 字段）
│   └── workflow.py       # AgentWorkflow
├── render/               # .vue 渲染器编译
├── units/                # unit 类型
│   ├── agent_unit.py     # agent unit（LLM 调用）
│   └── tool_unit.py      # tool unit
└── _builtin/             # ★ 内置 flow 模板（只读）
    ├── hello_world/      # 最小跨 agent demo
    │   ├── template.json
    │   ├── layout.json
    │   └── README.md
    └── weekly_report/    # 多角色周报协作
        ├── template.json
        ├── layout.json
        └── README.md

2.2 Runtime Disk Layout

Each custom flow template:

workspace/manager/flows/<template_id>/
├── template.json           # FlowTemplate（核心）
├── layout.json             # V2 track/module/variable 布局
├── docs/                   # Per-flow 文档（DESIGN.md / DEVELOPMENT.md / *.md）
│   ├── DESIGN.md
│   └── DEVELOPMENT.md
├── renderers/              # 自定义 .vue 渲染器
│   └── *.vue
├── contract.md             # 可选契约说明
└── README.md               # 可选

2.3 Built-in Templates

Registered at startup by install_built_ins(), listed under the "System" group in the frontend, read-only:

ID	Path	Description
`hello_world`	`src/flow/_builtin/hello_world/`	Minimal cross-agent collaboration demo: starter → jump to agent1 → agent1 jumps back to summary when done
`weekly_report`	`src/flow/_builtin/weekly_report/`	Multi-role collaboration: coordinator → data collector → writer → reviewer (conditional branch fallback) → summarizer

3. Schema Reference

3.1 `FlowTemplate` (template.json Top Level)

python

class FlowTemplate(BaseModel):
    template_id: str                       # [a-z0-9_-]{1,64}
    name: str = ""
    schema_version: int = FLOW_SCHEMA_VERSION  # 当前 = 3
    template_version: int = 1              # 作者每次编辑 +1
    main_agent_description: str = ""       # 主 agent 职责描述
    main_recipe_id: str = ""               # 主 agent 绑定的 recipe
    main_agent_id: Optional[int] = None
    workflows: list[AgentWorkflow]         # ★ 每个 role 的轨道
    units: dict[str, WorkUnit]             # 所有 unit（id → unit）
    checkpoints: dict[str, Checkpoint | CheckpointV2]  # V1/V2 共存
    triggers: dict[str, Trigger]
    renderers: list[RendererDecl]          # 自定义 .vue 渲染器
    contract_md: str = ""
    created_at: float
    updated_at: float

3.2 `AgentWorkflow` (a Role's Track)

python

class AgentWorkflow(BaseModel):
    role: str                              # 角色名（"main" / "agent1" / "collector" / ...）
    title: str = ""
    planning: {"strategy": "fixed", "needs_main_agent": bool}
    entry: str                             # 入口 node_id
    nodes: list[WorkflowNode]              # 节点序列

3.3 `WorkflowNode` (Nodes in a Track)

python

class WorkflowNode(BaseModel):
    node_id: str
    node_kind: str                         # starter | summary | checkpoint | unit | trigger
    ref: str                               # 指向 checkpoint_id / unit_id / trigger_id
    next: list[str]                        # 后继 node_id

3.4 `WorkUnit` (Execution Unit)

python

class WorkUnit(BaseModel):
    unit_id: str
    kind: str                              # "agent" | "tool"
    owner_role: str
    title: str = ""
    instruction: str                       # ★ 给 agent 的指令（system prompt 的一部分）
    input: {"fields": {name: type_str}}
    output: {"fields": {name: type_str}}
    tools: list[str]                       # 工具白名单
    subflow_template_id: str = ""          # 子 flow 模板
    input_fields: list[FieldSpec] = []     # starter default [prompt, attachments]
    output_fields: list[FieldSpec] = []    # summary default [answer, attachments]

3.5 `CheckpointV2` (Decision Point, with Branches/Jumps)

python

class CheckpointV2(BaseModel):
    checkpoint_id: str
    title: str = ""
    sub_modules: list[SubModule]           # ★ 分支 / 跳转 / 容器
    blocking: bool = True
    input_fields: list[FieldSpec] = []

SubModule types:

type	Meaning
`branch`	Conditional branch (case list + branch_bodies)
`jump`	Cross-role jump (target: {track_role, module_id, anchor})
`parallel_group`	Parallel group (holds side-by-side chains)
`combination_group`	Combination group (enforces head/tail adjacency)
`chain`	Sequential chain (a sequence of nodes executed in order)

Where branch bodies live (V2-critical): each case's branch body lives in branch_bodies[case.id], and the else branch lives in branch_bodies["__else__"]. Not case.modules or else_modules (deprecated). The migration is idempotent.

3.6 `Trigger` (Wait Condition)

python

class Trigger(BaseModel):
    trigger_id: str
    title: str = ""
    kind: str                              # "agent_state" | "unit_state" | ...
    params: dict

agent_state is the only trigger the engine injects automatically: it fires when a role's status becomes done.

3.7 `FieldSpec` (IO Fields)

python

class FieldSpec(BaseModel):
    name: str
    type: str = "str"
    required: bool = True
    description: str = ""

The Starter's default input_fields = [prompt, attachments]; the Summary's default output_fields = [answer, attachments]. These are reserved names — custom fields cannot share those names.

3.8 `RendererDecl` (Custom Renderers)

python

class RendererDecl(BaseModel):
    name: str
    source_file: str                       # 相对 renderers/ 的 .vue 路径
    description: str = ""

On save the backend's src.flow.render.compiler pre-compiles it to ESM JS, which the frontend mounts by name.

4. REST API

4.1 Flow Templates CRUD (`/api/flow-templates`)

Method	Path	Description
GET	`/api/flow-templates`	List all templates (summary rows)
POST	`/api/flow-templates`	Create
GET	`/api/flow-templates/{id}`	Fetch the full body
PUT	`/api/flow-templates/{id}`	Save (validate + persist)
DELETE	`/api/flow-templates/{id}`	Delete (refused when instances reference it)
POST	`/api/flow-templates/{id}/validate`	Dry-run validation
POST	`/api/flow-templates/{id}/duplicate`	Duplicate

4.2 Per-flow Docs (`/api/flow-templates/{id}/docs`)

Method	Path	Description
GET	`/{id}/docs`	List all markdown under docs/
GET	`/{id}/docs/{name}`	Read a document (DESIGN / DEVELOPMENT / custom)
PUT	`/{id}/docs/{name}`	Write a document (max 256KB)
DELETE	`/{id}/docs/{name}`	Delete

DESIGN.md: the flow's design document (returns the template skeleton when missing);
DEVELOPMENT: special — always points to the root FLOW_DEVELOPMENT.md (framework doc), is_global: true, the frontend warns before editing;
Name allowlist ^[A-Z0-9_-]{1,32}$, auto-uppercased.

4.3 Bundle Protocol (V2 track layout, schema_version=3)

Method	Path	Description
POST	`/api/flow-templates/preview_bundle`	Dry-run bundle validation
POST	`/api/flow-templates/import_bundle`	Upload a .flow.zip or folder path
GET	`/{id}/export_bundle`	Download .flow.zip

A bundle is a directory containing manifest.json + template.json + layout.json (+ contract.md / README.md / schemas/). The layout is the V2 track/module/variable layer; the template is the V1 pool layer. The two must cross-validate.

4.4 Renderers

Method	Path	Description
GET	`/{id}/renderers`	List renderers
GET	`/{id}/renderers/{name}`	Read .vue source
POST	`/{id}/renderers/{name}`	Write .vue source
DELETE	`/{id}/renderers/{name}`	Delete

4.5 Flow Instances (`/api/flows`)

Method	Path	Description
GET	`/api/flows`	List instances (supports status/template_id filtering)
GET	`/api/flows/{id}`	Fetch one instance + recent trace_tail events
GET	`/api/flows/{id}/trace`	Fetch the full trace (post-hoc analysis)
POST	`/api/flows/{id}/cancel`	Cancel (idempotent)

Launching a flow instance goes through src.pipeline.Generation._run_flow_recipe_chain, not these routes.

5. FlowEngine Runtime

5.1 Startup Sequence

python

engine = FlowEngine(
    graph=template.to_graph(),
    trace=trace,
    repo=repo,
    instance_id=instance_id,
    fields={"prompt": "...", "attachments": [...]},
    executor=default_executor,  # src.flow.units.dispatch
    ...
)
engine.run()

5.2 Thread Model

Owner thread: the thread calling run(); it processes the message queue, writes FlowState, and applies mutations.
Worker thread per role: one worker per AgentWorkflow. The worker takes the next node from the Scheduler, calls the unit executor (agent_unit / tool_unit), and posts the result back to the owner.

5.3 Mutation Types

kind	Meaning	Rewrites graph?
`add_unit`	Add a unit	Yes
`remove_unit`	Remove a unit	Yes
`rewire`	Rewire edges	Yes
`replace_workflow`	Replace a role's entire workflow	Yes
`expand_repeat`	Expand a repeat module	Yes
`set_config`	Change config	Yes
`set_field`	Write run state (dotted path)	No

5.4 State Machine

pending → running → done | failed
              ↓
         cancelled

pending: instance created but not started
running: engine.run() is executing
done: all roles have completed (cursor.ended)
failed: uncaught exception / Cancelled
cancelled: CancelToken fired

5.5 Resume (`src/flow/resume.py`)

rehydrate(instance_id, repo) reads pinned_template_json to rebuild the graph, replays flow_mutations in seq order, and compares each step's result_graph_hash against the apply_and_validate result; a mismatch raises FlowResumeError.

Fresh instance (no mutations) → directly returns an engine built from the base graph (equivalent to a fresh run)
set_field replay: writes FlowState without affecting the graph hash

6. Expression Evaluation

src/flow/expr/ provides AST-allowlist evaluation (sharing the Recipe syntax):

Supported: Python comparison/boolean operators, variable references, len(), literals
Forbidden: function calls (outside the allowlist), attribute access (outside a safe subset), imports, class definitions
Every edge condition= is evaluated through this

7. Common Pitfalls

Pitfall	Symptom	Fix
Parallel-group nodes wait on each other	Deadlock	Keep dependency directions one-way; avoid circular waits
Branch bodies in the wrong place	Frontend doesn't see branches / backend refuses to load	Use `branch_bodies[case.id]` / `branch_bodies["__else__"]`
Template modified, then run resumed	Hash mismatch	Resume refuses to continue; you must pin the version first
VueFlow node components not markRaw	"Component made reactive" warning + clicks fail	`markRaw(nodeTypes)`
VueFlow custom edge with two roots	"Extraneous non-props attributes" warning on every render	Wrap BaseEdge + EdgeLabelRenderer in a single `<g>` root
`watch(() => store.graph, ..., {deep: true})`	Render loop (fullscreen freezes)	Switch to a fingerprint string watcher
Starter field name taken	Validation error	Don't use `prompt` / `attachments` / `answer` as custom field names

8. Development Workflow (End-to-End)

Create a template in the flow editor (CreativeWorkshop) → drag nodes / connect edges;
Add the Starter header (auto-created) and Summary tail (auto-created), and add workflows/checkpoints/triggers in between;
Configure each node's parameters:
- unit: instruction, input/output fields, tools allowlist
- checkpoint: branch conditions (AST expressions), jump target (track_role + module_id)
- trigger: wait condition (agent_state / unit_state)
Write docs/DESIGN.md (design) + docs/DEVELOPMENT.md (development notes);
Validate with POST /api/flow-templates/{id}/validate;
Bind to an agent → start the instance (the chain runs automatically);
Inspect the trace at /api/flows/{id};
Interrupted runs can be continued with rehydrate (refused if the template was modified).

9. Relationship with Recipe

Recipe is a "single-agent, single-turn state graph"; Flow is a "multi-agent, cross-turn workflow";
A Flow unit (agent unit) invokes a Recipe internally (once flow_template_id is bound to an agent, that unit runs that recipe);
Flow exposes tools to agents via filtered_schemas, not through tools.json;
The Flow editor's "Compile / Dry run" buttons go through POST /api/flow-templates/{id}/validate;
A Flow template's docs/ is shown as DESIGN / DEVELOPMENT tabs in the frontend.

Reference Source Code

Template model: src/flow/models/template.py
Graph model: src/flow/models/graph.py
Engine: src/flow/engine.py
Resume: src/flow/resume.py
REST (templates): src/bridge/routes/flow_templates.py
REST (instances): src/bridge/routes/flows.py
Built-ins: src/flow/_builtin/{hello_world,weekly_report}/

Flow Development Guide ​

1. Concept Model ​

1.1 What is a Flow ​

1.2 Template vs Instance ​

1.3 The Three Invariants (engine.py) ​

2. Directory and Files ​

2.1 Source Tree ​

2.2 Runtime Disk Layout ​

2.3 Built-in Templates ​

3. Schema Reference ​

3.1 FlowTemplate (template.json Top Level) ​

3.2 AgentWorkflow (a Role's Track) ​

3.3 WorkflowNode (Nodes in a Track) ​

3.4 WorkUnit (Execution Unit) ​

3.5 CheckpointV2 (Decision Point, with Branches/Jumps) ​

3.6 Trigger (Wait Condition) ​

3.7 FieldSpec (IO Fields) ​

3.8 RendererDecl (Custom Renderers) ​

4. REST API ​

4.1 Flow Templates CRUD (/api/flow-templates) ​

4.2 Per-flow Docs (/api/flow-templates/{id}/docs) ​

4.3 Bundle Protocol (V2 track layout, schema_version=3) ​

4.4 Renderers ​

4.5 Flow Instances (/api/flows) ​

5. FlowEngine Runtime ​

5.1 Startup Sequence ​

5.2 Thread Model ​

5.3 Mutation Types ​

5.4 State Machine ​

5.5 Resume (src/flow/resume.py) ​

6. Expression Evaluation ​

7. Common Pitfalls ​

8. Development Workflow (End-to-End) ​

9. Relationship with Recipe ​

Reference Source Code ​

Flow Development Guide

1. Concept Model

1.1 What is a Flow

1.2 Template vs Instance

1.3 The Three Invariants (engine.py)

2. Directory and Files

2.1 Source Tree

2.2 Runtime Disk Layout

2.3 Built-in Templates

3. Schema Reference

3.1 `FlowTemplate` (template.json Top Level)

3.2 `AgentWorkflow` (a Role's Track)

3.3 `WorkflowNode` (Nodes in a Track)

3.4 `WorkUnit` (Execution Unit)

3.5 `CheckpointV2` (Decision Point, with Branches/Jumps)

3.6 `Trigger` (Wait Condition)

3.7 `FieldSpec` (IO Fields)

3.8 `RendererDecl` (Custom Renderers)

4. REST API

4.1 Flow Templates CRUD (`/api/flow-templates`)

4.2 Per-flow Docs (`/api/flow-templates/{id}/docs`)

4.3 Bundle Protocol (V2 track layout, schema_version=3)

4.4 Renderers

4.5 Flow Instances (`/api/flows`)

5. FlowEngine Runtime

5.1 Startup Sequence

5.2 Thread Model

5.3 Mutation Types

5.4 State Machine

5.5 Resume (`src/flow/resume.py`)

6. Expression Evaluation

7. Common Pitfalls

8. Development Workflow (End-to-End)

9. Relationship with Recipe

Reference Source Code