outbound-frontend.md55.0 KBView on GitHub # Outbound Engine Frontend
## 1) Introduction — goal, present state, future state
We want a first-class `/outbound` section in the mail app that lets a user see and operate the headless outbound engine described in [outbound-engine.md](apps/server/docs/outbound/core/outbound-engine.md) and [flow.md](apps/server/docs/outbound/core/flow.md): a landing page listing every Flow and List, an editable React-Flow canvas for each Flow (drag stages, connect them, edit each stage's config, run it), and drill-down from a node into a full-page record table (a List rendered like the CRM conversation canvas). Today none of this surface exists — the entire engine (Flows, Lists, Qualifiers, Cedar Inputs) is reachable only over tRPC (`trpc.outbound.*`, mounted at [trpc/index.ts:49](apps/server/src/trpc/index.ts)) and MCP; the only React-Flow code in the repo is the read-only file-dependency graph at [FileGraph.tsx](apps/mail/modules/files/graph/FileGraph.tsx), and the only rich record table is the CRM [ConversationCanvas](apps/mail/modules/crm/components/conversation-canvas/ConversationCanvas.tsx). The change adds a new `apps/mail/modules/outbound` feature module plus `(routes)/outbound/*` route files. The flow canvas is a **new, standalone component ported from the gumloop flow-builder** (`/Users/jesse/Desktop/apps/gumloop_frontend`) — an editable, statically-positioned `@xyflow/react` editor — explicitly **not** built on the repo's read-only, d3-force FileGraph; the record table generalizes the ConversationCanvas's virtualized column table to person/account records — all backed by a small set of new and existing `outbound.flow.*` / `outbound.lists.*` endpoints.
## 2) Present state
### 2.1 Architecture diagram
```text
BACKEND (built) REFERENCES (models to mirror) FRONTEND (/outbound)
───────────────── ──────────────────────────────────── ────────────────────
trpc.outbound.flow.* gumloop_frontend (EXTERNAL) ★ THE MODEL
list ✓ get ✗ create ✓ update ✓ PipelineDiagram.tsx (editable canvas)
run ✓ getRun ✓ getRow ✓ getTrace ✓ OperatorNode + connection-points (handles)
trpc.outbound.lists.* flow-slice (nodes/edges store) ┌──────────────┐
list ✓ members ✓ create ✓ nodes-to-pipeline (graph→serialized) │ NO /outbound │
trpc.outbound.qualifier.* (CRUD ✓) node-library (drag palette) │ UI YET │
trpc.outbound.cedarInput.* (CRUD ✓) static positions · NO d3-force │ (only the UI │
│ is missing) │
db: flows / flow_runs / flow_rows FileGraph.tsx — prior @xyflow usage, but └──────────────┘
crm_lists / crm_list_members read-only + d3-force → NOT a model
↑ NOT empty: 6 flows · 17 qualifiers · 5 lists · 9 runs already exist as live records
crm_person / crm_account ConversationCanvas.tsx (virtua VList) — list-table model
useBulkSelection / conversationSelection
LeftSidebarContent.tsx primaryNavButtons[]
routes.ts layout()/route() manifest
```
### 2.2 Step-by-step walkthrough
This is how an existing top-level section (CRM) renders end-to-end today, plus the backend data the new section will consume. Every function/path the design reuses is listed.
1. **Route manifest** — [routes.ts:84](apps/mail/app/routes.ts) registers `layout('(routes)/crm/layout.tsx', [route('/crm', '(routes)/crm/page.tsx')])`. Top-level sections are one `layout()` wrapping one or more `route()` entries; dynamic segments use `:param` (e.g. [routes.ts:90](apps/mail/app/routes.ts) `route('/meetings/:sdkUploadId', …)`).
- Data after this step (the shape we add): a new block `layout('(routes)/outbound/layout.tsx', [ route('/outbound', …), route('/outbound/flows/:flowId', …), route('/outbound/lists/:listId', …) ])`.
2. **Shell + nav** — [LeftSidebarContent.tsx:125](apps/mail/modules/conversations/components/LeftSidebarContent.tsx) defines `primaryNavButtons: NavButton[]` (`{ id, title, icon, href }` at [:116](apps/mail/modules/conversations/components/LeftSidebarContent.tsx)); the section's page renders into the `<Outlet />` of [(routes)/layout.tsx](<apps/mail/app/(routes)/layout.tsx>). Active state via `isActive(href)` at [:290](apps/mail/modules/conversations/components/LeftSidebarContent.tsx).
3. **Page → module** — [(routes)/crm/page.tsx](<apps/mail/app/(routes)/crm/page.tsx>) imports the main view from the module barrel (`@/modules/crm`) and renders it. The module lives at [apps/mail/modules/crm/](apps/mail/modules/crm/) with `components/ hooks/ store/ types/ index.ts`.
4. **tRPC data fetch** — a page calls `const trpc = useTRPC()` (from `@/providers/query-provider`) then `useQuery(trpc.<router>.<proc>.queryOptions(input))`. The outbound router is mounted as `outbound` at [trpc/index.ts:49](apps/server/src/trpc/index.ts), so the new calls are `trpc.outbound.flow.list`, `trpc.outbound.lists.members`, etc.
- Flow list source — [outbound.ts:128](apps/server/src/trpc/routes/outbound.ts) `flow.list` → `flows.$inferSelect[]`:
```ts
{ id, name, slug, status, stages: FlowStage[], inputs, outputs, defaults, version, updatedAt }
```
- **Gap:** there is no `flow.get` ([outbound.ts:126-142](apps/server/src/trpc/routes/outbound.ts) has `list` but no single-fetch); `flow.getFlow(ctx, id)` exists in the service but is unexposed. The canvas page needs it.
5. **FlowStage union (the graph the canvas renders)** — [flow/types.ts:17](apps/server/src/services/outbound/flow/types.ts):
```ts
type FlowStage =
| { id; kind: 'cedarInput'; cedarInputId; after?: string[] }
| { id; kind: 'qualifier'; qualifierId; after?: string[]; inputMapping?; qualifierVersion? }
| { id; kind: 'gate'; on?; field?; op: GateOp; value?; onPass?; onFail?; after?: string[] }
| { id; kind: 'expand'; titles?: string[]; after?: string[]; carryCompanyFields? }
| {
id;
kind: 'list';
listName?;
listId?;
listKind?: 'contact' | 'account' | 'mixed';
after?: string[];
};
```
- Stages form a DAG via `after` edges (default = lexically previous stage). **Stages carry no `x`/`y`** — there is nothing to position the nodes with yet.
6. **Gumloop flow-builder — THE architectural model (external).** The canvas is a new, standalone component modeled on the gumloop flow-builder at `/Users/jesse/Desktop/apps/gumloop_frontend` (`@xyflow/react ^12.8.1`, Redux), **not** on FileGraph. Source files to mirror (read them while implementing — see the standing instruction in §3.2):
- `src/components/pipeline/canvas/PipelineDiagram.tsx` — the orchestrator: `nodeTypes`/`edgeTypes` registries (≈L103-115), `onConnect` validation (≈L419-475), palette `onDrop`/`onDragOver` + `screenToFlowPosition` (≈L498-517), `onNodeDragStop`, static positioning (no force sim).
- `src/components/OperatorNode/OperatorNode.tsx` + `connection-points.tsx` — a custom node: header, body, typed `Handle` ports.
- `src/components/OperatorNode/useRenderParameter.tsx` + `use-handle-parameter-change.ts` — per-node param forms whose edits write back to the store.
- `redux/pipeline/flow/flow-slice.ts` — the nodes/edges store + `onNodesChange`/`onConnect`/`setNode` actions (we port this to a Zustand slice).
- `src/utils/pipeline/nodes-to-pipeline.ts` — node graph → positioned step array (`pipeline[] = { id, operator, parameters, inputs, position }`) serialized to the backend. Our analog is `stages[]` + the `defaults.layout` positions map.
- `src/components/pipeline/nodes-menu/node-library/*` — the draggable palette (`onDragStart` sets the drag payload).
We diverge from gumloop only where it's heavier than we need: **Zustand instead of Redux**, **no run-status websocket** (we poll `flow.getRun`), and a small **static per-kind form registry** instead of gumloop's declaration-driven dynamic param engine.
7. **FileGraph — prior art we deliberately do NOT reuse.** [FileGraph.tsx](apps/mail/modules/files/graph/FileGraph.tsx) is the repo's only existing `@xyflow/react` usage, but it is **read-only** (`nodesConnectable={false}`, no palette, no persistence) and **d3-force-driven** (a live physics simulation for an Obsidian-style graph). The flow canvas is an *editor* with *deliberate, persisted* node positions — a force simulation is actively wrong for it (nodes must stay where the user drops them). So we **do not** import from `modules/files/graph` and **do not** use `d3-force`. The only thing worth borrowing is the trivially-generic fact that `@xyflow/react ^12.3.5` + `@dagrejs/dagre ^1.1.4` are already in [apps/mail/package.json](apps/mail/package.json) (dagre is used once for initial auto-layout of position-less flows; never a force sim).
8. **List members source** — [outbound.ts:173](apps/server/src/trpc/routes/outbound.ts) `lists.members({ listId?, name? })` → joins `crm_list_members` with `crm_person`/`crm_account` + emails/domains. A member row exposes identity (`displayName`, primary `email`, `currentTitle`, company `name/primaryDomain`) plus the additive `custom` jsonb field table.
9. **Record-table reference** — [ConversationCanvas.tsx](apps/mail/modules/crm/components/conversation-canvas/ConversationCanvas.tsx) renders a `virtua` `VList` (`itemSize≈43`, `overscan`, infinite-scroll on `onScroll`) with [ColumnConfigurationItem](apps/mail/modules/crm/components/conversation-canvas/ColumnConfigurationItem.tsx) as a width-synced header above [ConversationItem](apps/mail/modules/crm/components/conversation-canvas/ConversationItem.tsx) rows; columns come from [use-canvas-configuration.ts](apps/mail/modules/crm/hooks/use-canvas-configuration.ts) (`CRMColumn[]`, widths from [canvas-columns.ts](apps/mail/modules/crm/utils/canvas-columns.ts)); selection via [use-bulk-selection.ts](apps/mail/modules/crm/hooks/use-bulk-selection.ts) + `conversationSelection` in [conversationsSlice.ts](apps/mail/modules/conversations/slice/conversationsSlice.ts). The `virtua` VList, width system, and bulk-selection keyboard modes are conversation-agnostic; the data hook, filter/sort, and popovers are coupled to conversations.
## 3) Designed state
### 3.1 Architecture diagram
```text
ROUTES MODULE apps/mail/modules/outbound/
────── ───────────────────────────────────
/outbound ──renders──► components/OutboundHome.tsx
(tabs: Flows | Lists) ├─ FlowsTab → trpc.outbound.flow.list
└─ ListsTab → trpc.outbound.lists.list
/outbound/flows/:flowId ──renders──► components/flow-canvas/FlowCanvas.tsx
ReactFlowProvider
└─ ReactFlow (editable)
NODE_TYPES: cedarInput|qualifier|gate|
expand|list (StageNode/*)
EDGE_TYPES: afterEdge
├─ StagePalette (drag → onDrop) ┐
├─ StageInspector (per-kind config) │ store:
├─ FlowRunBar (run / poll / counts) │ outboundFlowSlice
└─ onConnect → after-edge ┘ (Zustand)
│ drill-down (list node click)
▼ navigate(/outbound/lists/:id)
/outbound/lists/:listId ──renders──► components/list-table/RecordCanvas.tsx
virtua VList + RecordHeader + RecordItem
useRecordColumns (union of member.custom)
recordSelection (generic selection slice)
▲ trpc.outbound.lists.members({listId})
NEW/CHANGED BACKEND: outbound.flow.get · outbound.flow.update persists stages+positions
(positions live in flows.defaults.layout — jsonb, no migration)
```
### 3.2 Step-by-step walkthrough
> **Standing implementation instruction — consult the gumloop source.** The flow canvas (everything under `components/flow-canvas/`) is a port of the gumloop flow-builder at `/Users/jesse/Desktop/apps/gumloop_frontend` (see §2.2 step 6 for the file map). Whenever a canvas decision is unclear or underspecified here — handle placement, `onConnect`/drag/drop edge cases, edge rendering, viewport/controls config, copy-paste, undo/redo, multi-select, node sizing, run-state styling — **read the corresponding gumloop file first and follow its approach** (adapted to Zustand + our `FlowStage` model), rather than inventing a new one. After resolving an ambiguity from the gumloop source, **report back**: note in the PR/commit (and as a one-line `// ref: gumloop <file>:<symbol>` comment at the call site) what gumloop did and the decision taken, so reviewers can trace it. Do **not** consult FileGraph for canvas behavior — it is read-only + d3-force and is not the model.
1. **Add the route block** — in [routes.ts](apps/mail/app/routes.ts), inside the `(routes)/layout.tsx` array, add:
```ts
layout('(routes)/outbound/layout.tsx', [
route('/outbound', '(routes)/outbound/page.tsx'),
route('/outbound/flows/:flowId', '(routes)/outbound/flows/[flowId]/page.tsx'),
route('/outbound/lists/:listId', '(routes)/outbound/lists/[listId]/page.tsx'),
]),
```
New files: `(routes)/outbound/layout.tsx` (Sidepanel + `<Outlet/>`, mirroring [(routes)/crm/layout.tsx](<apps/mail/app/(routes)/crm/layout.tsx>)) and the three `page.tsx` thin wrappers importing from `@/modules/outbound`.
2. **Add the nav entry** — append to `primaryNavButtons` at [LeftSidebarContent.tsx:125](apps/mail/modules/conversations/components/LeftSidebarContent.tsx): `{ id:'outbound', title:'Outbound', icon: Send, href:'/outbound' }`, and extend `isActive` so `/outbound/*` keeps it highlighted.
3. **Home page (Flows + Lists)** — `OutboundHome.tsx` calls `useTRPC()` then `useQuery(trpc.outbound.flow.list.queryOptions())` and `useQuery(trpc.outbound.lists.list.queryOptions())`, rendering two tabs of cards.
- Flow card data:
```json
{
"id": "f-abc",
"name": "Website visitors → AE sequence",
"status": "active",
"version": 1,
"stages": [
{ "id": "src", "kind": "cedarInput" },
{ "id": "score", "kind": "qualifier" }
],
"updatedAt": "…"
}
```
- Clicking a flow → `navigate('/outbound/flows/f-abc')`; clicking a list → `navigate('/outbound/lists/<id>')`; "New flow" → `flow.create` then navigate.
4. **`flow.get` endpoint (new)** — add to [outbound.ts](apps/server/src/trpc/routes/outbound.ts) flow router: `get: privateProcedure.input(z.object({ id: z.string() })).query(({ctx,input}) => flow.getFlow(octx(ctx), input.id))`. Returns the full `flows.$inferSelect` (stages + `defaults`).
5. **Stages → React-Flow graph** — `flow-canvas/graph/stagesToFlow.ts` maps `stages[]` to `{ nodes, edges }`. Each stage → a node `{ id: stage.id, type: stage.kind, data: { stage }, position }`. Edges come from `after` (and gate `onPass`/`onFail`): for each `s`, for each `a in s.after ?? [prevStageId]`, emit `{ id:`${a}->${s.id}`, source:a, target:s.id, type:'afterEdge' }`. Positions read from `flows.defaults.layout?.[stage.id]`; when absent, run a one-shot `@dagrejs/dagre` top-to-bottom layout (a small standalone helper in `flow-canvas/graph/autoLayout.ts` — **not** FileGraph's force layout) so config/agent-authored flows render laid-out immediately. Positions are static thereafter (gumloop-style): the user drags, we persist; nothing re-flows on its own.
- Data after this step:
```json
{
"nodes": [
{
"id": "src",
"type": "cedarInput",
"position": { "x": 0, "y": 0 },
"data": { "stage": { "id": "src", "kind": "cedarInput", "cedarInputId": "in-rb2b" } }
},
{
"id": "score",
"type": "qualifier",
"position": { "x": 0, "y": 120 },
"data": { "stage": { "id": "score", "kind": "qualifier", "qualifierId": "q-basic" } }
}
],
"edges": [{ "id": "src->score", "source": "src", "target": "score", "type": "afterEdge" }]
}
```
6. **Canvas state (Zustand slice)** — `store/outboundFlowSlice.ts` holds `{ flowId, nodes, edges, dirty, selectedStageId }` and actions `setGraph`, `onNodesChange`, `onEdgesChange`, `addStage`, `updateStage`, `deleteStage`, `connect`, `setStagePosition`. This is gumloop's `redux/pipeline/flow/flow-slice.ts` (nodes/edges + `onNodesChange`/`onConnect`/`setNode`) ported to a Zustand slice; `FlowCanvas` binds the actions to `<ReactFlow nodes edges onNodesChange onEdgesChange onConnect>` so every edit flows through the store and persists. Wired into [CedarStoreTypes.ts](apps/mail/modules/store/CedarStoreTypes.ts) + [store/index.ts](apps/mail/modules/store/index.ts) like every other slice.
7. **Stage nodes** — `flow-canvas/nodes/{CedarInputNode,QualifierNode,GateNode,ExpandNode,ListNode}.tsx`, registered in a `STAGE_NODE_TYPES` map (the standard `@xyflow/react` `nodeTypes` registry; mirror gumloop's `PipelineDiagram.tsx` `nodeTypes` ≈L109-115 and `OperatorNode.tsx`/`connection-points.tsx` for node + handle structure). Each renders a labeled card with a target `Handle` (top) and source `Handle` (bottom — gate has two), a kind icon, and a one-line summary (`gate: score ≥ 7`, `qualifier: q-basic-score`, `list: AE warm`). A `ListNode` (and a materializing `QualifierNode`) shows an "Open list →" affordance.
8. **Palette + add node** — `StagePalette.tsx` is grouped, not a flat list of stage kinds. The first group is **Inputs** (the source types — see [Appendix A.10](#a10-the-inputs-palette--source-type-model)); the rest are the transform/storage stages (`qualifier`, `gate`, `expand`, `list`). Each chip's `onDragStart` sets `application/x-outbound-stage` with the chosen kind **and** (for inputs) the source sub-type. `FlowCanvas.onDrop` computes `screenToFlowPosition` (gumloop pattern) and calls `addStage(kind, position, sourceType?)`, which appends a stage (or, for CSV/API doors, a flow-level input) with defaults and records its position in `defaults.layout`.
9. **Connect / route edges** — `onConnect({source,target})` validates (source≠target, no existing `source->target`, target not a `cedarInput` root) then `connect()` adds `target.id` to the target stage's `after[]` and emits the edge. Deleting an edge removes that `after` entry.
10. **Stage inspector (per-kind config)** — `StageInspector.tsx` (right panel) switches on `selectedStageId`'s kind and renders a small form, writing through `updateStage(id, patch)`:
- `cedarInput` → pick/create the source by **sub-type** (LinkedIn · Crustdata · crawler), then its config — e.g. LinkedIn is **one** entry with a `signal` dropdown (`my_post_engagement` | `profile_viewers` | `accepted_connections`), **not** three entries (see [Appendix A.10](#a10-the-inputs-palette--source-type-model)). Lists/creates via `trpc.outbound.cedarInput.list`/`create`.
- input doors (`csv` · `api`/adapter) → configured on the root door node, not a `cedarInput` stage.
- `qualifier` → select a saved qualifier (`trpc.outbound.qualifier.list`) + `inputMapping`.
- `gate` → `on` (field key), `op` (`GateOp`), `value`, `onFail` (`exit`|stageId).
- `expand` → `titles[]`, `perCompanyLimit`, `carryCompanyFields`.
- `list` → `listName`/`listId` (`trpc.outbound.lists.list`), `listKind`.
This mirrors gumloop's declaration-driven param forms, simplified to a static per-kind registry.
11. **Validate + save** — on change `dirty=true`; "Save" calls `trpc.outbound.flow.validateConfig` (surfacing `ConfigValidation.issues` inline on offending nodes) then `trpc.outbound.flow.update({ id, patch: { stages, defaults } })`, where `defaults.layout` carries `{ [stageId]: {x,y} }`. No DB migration — `defaults` is already jsonb and `update` accepts `z.any()`.
- Patch shape:
```json
{ "stages":[{"id":"src","kind":"cedarInput","cedarInputId":"in-rb2b"}, …],
"defaults":{ "autoRun":true, "layout":{ "src":{"x":0,"y":0}, "score":{"x":0,"y":120} } } }
```
12. **Run + overlay** — `FlowRunBar.tsx` calls `trpc.outbound.flow.run({ flowId })` → `RunFlowResult` ([flow/types.ts:103](apps/server/src/services/outbound/flow/types.ts)); polls `trpc.outbound.flow.getRun({ runId })` and overlays per-stage `{ in, out, exited }` onto each node, recoloring the node border (idle/running/done/exited) — the gumloop run-state pattern, sourced from `result.stages[]` keyed by `stageId`.
- Overlay data:
```json
{ "status":"completed","stages":[
{"stageId":"gate7","kind":"gate","in":47,"out":22,"exited":25}, … ] }
```
13. **Drill-down to a List** — clicking a `ListNode` (with a resolved `listId`) calls `navigate('/outbound/lists/<listId>')`. (A node whose list is `listName`-only/not yet materialized shows a disabled affordance.)
14. **List table page** — `list-table/RecordCanvas.tsx` calls `useQuery(trpc.outbound.lists.members.queryOptions({ listId }))` and renders a `virtua` `VList` (the [ConversationCanvas.tsx](apps/mail/modules/crm/components/conversation-canvas/ConversationCanvas.tsx) recipe) of `RecordItem` rows under a width-synced `RecordHeader`.
- Member row data:
```json
{
"id": "m-1",
"recordKind": "person",
"displayName": "Alice Lee",
"email": "<email>",
"currentTitle": "AE",
"company": { "name": "Acme", "primaryDomain": "acme.com" },
"custom": { "fit_score": 8, "why_now": "hiring SDRs" }
}
```
15. **Record columns + row** — `useRecordColumns(members)` builds a column set: fixed identity columns (avatar+name, title, company, email) + one column per key in the union of all `member.custom` keys (the additive field table). `RecordItem.tsx` is a generalized [ConversationItem.tsx](apps/mail/modules/crm/components/conversation-canvas/ConversationItem.tsx) — same fixed-name section + flexible typed cells, but values come from `member.custom[key]` / identity fields instead of conversation fields. `RecordHeader.tsx` reuses the width system from [canvas-columns.ts](apps/mail/modules/crm/utils/canvas-columns.ts).
16. **Generic selection** — `store/recordSelectionSlice.ts` is a record-agnostic extraction of the `conversationSelection` shape (`selectedIds`, `setSelection`, `toggleSelection`, `clearSelection`, `anchorId`); `RecordCanvas` drives it with the existing [use-bulk-selection.ts](apps/mail/modules/crm/hooks/use-bulk-selection.ts) hook (which already takes generic `items`/`selectedIds`/setters), enabling shift/cmd range + mass select on records.
## 4) Implementation phases
### Phase 1 — Section scaffold: route, nav, home page (Flows + Lists lists)
**Goal:** `/outbound` exists in the nav and lists every Flow and List as navigable cards.
- [x] Create `apps/mail/modules/outbound/` with `components/ hooks/ store/ types/ index.ts` (barrel).
- [x] Add `(routes)/outbound/layout.tsx` (Sidepanel + `<Outlet/>`, copying [(routes)/crm/layout.tsx](<apps/mail/app/(routes)/crm/layout.tsx>)) and `(routes)/outbound/page.tsx` importing `OutboundHome` from `@/modules/outbound`.
- [x] Register the `/outbound` route in [routes.ts](apps/mail/app/routes.ts) (flows/lists sub-routes added in later phases).
- [x] Add the `outbound` entry to `primaryNavButtons` and extend `isActive` in [LeftSidebarContent.tsx](apps/mail/modules/conversations/components/LeftSidebarContent.tsx).
- [x] Build `components/OutboundHome.tsx` with `Flows`/`Lists` tabs, fetching `trpc.outbound.flow.list` and `trpc.outbound.lists.list` via `useTRPC()` + `useQuery`.
- [x] Build `components/FlowCard.tsx` / `ListCard.tsx`; flow card → `navigate('/outbound/flows/:id')`, list card → `navigate('/outbound/lists/:id')` (routes land later — guard until then).
- [x] Add a "New flow" / "New list" action (`trpc.outbound.flow.create` / `lists.create`).
**Tests:**
- [x] `apps/mail/modules/outbound/components/__tests__/OutboundHome.test.tsx` — renders flow/list cards from mocked tRPC, tab switch works, card click calls `navigate` with the right path.
- [x] `pnpm --filter @cedar/mail test modules/outbound`
- [x] `pnpm --filter @cedar/mail typecheck`
### Phase 2 — Backend: `flow.get` + position-persisting `flow.update`
**Goal:** The frontend can fetch one flow and round-trip stages + node positions.
- [x] Add `flow.get` to [outbound.ts](apps/server/src/trpc/routes/outbound.ts) (`flow.getFlow(octx(ctx), input.id)`).
- [x] Confirm `flow.update` persists a `defaults.layout` map untouched (jsonb passthrough); add a service-level guard so `validateConfig` ignores `defaults.layout`.
- [x] Export the `FlowStage` / `Flow` types for frontend reuse (a `packages/db` or shared `types` re-export so the module imports the same union as [flow/types.ts:17](apps/server/src/services/outbound/flow/types.ts)).
**Tests:**
- [x] `apps/server/src/services/outbound/flow/__tests__/get-update-layout.test.ts` — `getFlow` returns stages; `updateFlow` with `defaults.layout` persists and re-reads identically; `validateConfig` passes when `defaults.layout` present.
- [x] `pnpm --filter @cedar/server test services/outbound/flow`
### Phase 3 — Read-only flow canvas (render the stage DAG)
**Goal:** Opening a flow renders its stages as a laid-out, pannable React-Flow graph (no editing yet).
- [x] Add `(routes)/outbound/flows/[flowId]/page.tsx` + its route entry; fetch via `trpc.outbound.flow.get`.
- [x] Build `components/flow-canvas/graph/stagesToFlow.ts` (stages+`after`+gate routes → nodes/edges; positions from `defaults.layout`, else `@dagrejs/dagre` top-to-bottom).
- [x] Build the five `nodes/*Node.tsx` + `STAGE_NODE_TYPES` map and an `afterEdge` edge type (port from gumloop `OperatorNode.tsx` + `connection-points.tsx`; standard `@xyflow/react` `nodeTypes`/`edgeTypes` registries).
- [x] Build `graph/autoLayout.ts` — a one-shot `@dagrejs/dagre` top-to-bottom pass for position-less flows (standalone; **no** d3-force, no FileGraph import).
- [x] Build `components/flow-canvas/FlowCanvas.tsx` (`ReactFlowProvider` → `ReactFlow` with `Background`/`Controls`/`MiniMap`, read-only: `nodesConnectable={false}`).
**Tests:**
- [x] `apps/mail/modules/outbound/components/flow-canvas/graph/__tests__/stagesToFlow.test.ts` — linear chain, gate `onFail` branch, and an `expand` flow each produce correct nodes/edges; dagre fallback fires only when `defaults.layout` absent.
- [x] `pnpm --filter @cedar/mail test modules/outbound/components/flow-canvas`
### Phase 4 — Editable canvas: palette, connect, drag-persist, Zustand slice
**Goal:** Add/move/connect/delete stages on the canvas and persist via `flow.update`.
- [x] Build `store/outboundFlowSlice.ts` (`nodes/edges/dirty/selectedStageId` + `setGraph/onNodesChange/onEdgesChange/addStage/deleteStage/connect/setStagePosition`); wire into [CedarStoreTypes.ts](apps/mail/modules/store/CedarStoreTypes.ts) + [store/index.ts](apps/mail/modules/store/index.ts).
- [x] Make `FlowCanvas` editable: bind slice actions, `nodesConnectable`, `onConnect` with validation (no self-loop/dup/root-target), `onNodeDragStop` → `setStagePosition`.
- [x] Build `components/flow-canvas/StagePalette.tsx` + `FlowCanvas.onDrop`/`onDragOver` (`screenToFlowPosition` → `addStage`).
- [x] Build `graph/flowToStages.ts` (inverse of `stagesToFlow`) and a "Save" action: `validateConfig` then `flow.update({ id, patch:{ stages, defaults:{…, layout } } })`; reflect `dirty`.
**Tests:**
- [x] `apps/mail/modules/outbound/store/__tests__/outboundFlowSlice.test.ts` — addStage appends with default `after`; connect mutates `after[]`; deleteStage prunes edges; flowToStages∘stagesToFlow round-trips.
- [x] `apps/mail/modules/outbound/components/flow-canvas/__tests__/onConnect.test.ts` — rejects self-loop, duplicate, and root-target connections.
- [x] `pnpm --filter @cedar/mail test modules/outbound`
### Phase 5 — Stage inspector (per-kind config forms)
**Goal:** Select a node and edit its stage config, persisting on save.
- [x] Build `components/flow-canvas/StageInspector.tsx` switching on stage kind, writing via `updateStage`.
- [x] Per-kind forms: `cedarInput` (source picker via `cedarInput.list`), `qualifier` (qualifier picker via `qualifier.list` + `inputMapping`), `gate` (`on`/`op`/`value`/`onFail`), `expand` (`titles`/`perCompanyLimit`/`carryCompanyFields`), `list` (`listName`/`listId`/`listKind`).
- [x] Surface `validateConfig` issues inline on the offending node + in the inspector.
**Tests:**
- [x] `apps/mail/modules/outbound/components/flow-canvas/__tests__/StageInspector.test.tsx` — editing a gate's `op`/`value` updates the stage; a qualifier picker sets `qualifierId`; invalid config shows the issue.
- [x] `pnpm --filter @cedar/mail test modules/outbound/components/flow-canvas`
### Phase 6 — Run bar + per-stage overlay
**Goal:** Run a flow from the canvas and visualize per-stage counts/status on nodes.
- [x] Build `components/flow-canvas/FlowRunBar.tsx` (`flow.run`, poll `flow.getRun`).
- [x] Overlay `RunFlowResult.stages[]` `{in,out,exited}` + status color onto nodes; idle when no run.
- [x] "View rows" opens a `flow.listRows` drawer; row → `flow.getTrace` panel.
**Tests:**
- [x] `apps/mail/modules/outbound/components/flow-canvas/__tests__/FlowRunBar.test.tsx` — run → polling → node overlay reflects mocked `getRun` stage counts and statuses.
- [x] `pnpm --filter @cedar/mail test modules/outbound/components/flow-canvas`
### Phase 7 — List record table (drill-down target)
**Goal:** A List node opens a full-page, virtualized, selectable table of its records.
- [x] Add `(routes)/outbound/lists/[listId]/page.tsx` + route entry; `ListNode` click → `navigate('/outbound/lists/:id')`.
- [x] Build `store/recordSelectionSlice.ts` (generic extraction of the `conversationSelection` shape) and wire into the store.
- [x] Build `components/list-table/RecordCanvas.tsx` (`virtua` `VList` + infinite scroll), `RecordHeader.tsx` (reuse [canvas-columns.ts](apps/mail/modules/crm/utils/canvas-columns.ts) widths), `RecordItem.tsx` (generalized [ConversationItem.tsx](apps/mail/modules/crm/components/conversation-canvas/ConversationItem.tsx)).
- [x] Build `hooks/use-record-columns.ts` (identity columns + union of `member.custom` keys) fed by `trpc.outbound.lists.members({ listId })`.
- [x] Drive selection with the existing [use-bulk-selection.ts](apps/mail/modules/crm/hooks/use-bulk-selection.ts) hook.
**Tests:**
- [x] `apps/mail/modules/outbound/hooks/__tests__/use-record-columns.test.ts` — columns = identity set ∪ all `custom` keys across members; missing values render empty.
- [x] `apps/mail/modules/outbound/components/list-table/__tests__/RecordCanvas.test.tsx` — renders rows from mocked `lists.members`, header widths match cells, shift/cmd selection updates `recordSelection`.
- [x] `pnpm --filter @cedar/mail test modules/outbound`
- [x] `pnpm --filter @cedar/mail typecheck`
---
## Appendix A — React Flow schema architecture (the type-level mapping)
This is the load-bearing part of the canvas: how the backend's `FlowStage[]` domain model is projected into React Flow's runtime `Node`/`Edge` model and reconstructed losslessly on save. Everything here lives in `apps/mail/modules/outbound/components/flow-canvas/graph/` + `store/outboundFlowSlice.ts`.
### A.1 The two type universes
There are two distinct models, and the whole canvas is the adapter between them. Keeping them separate is the design — React Flow types never leak into what we persist, and domain types never carry pixel coordinates.
| Concern | Backend domain model (persisted) | React Flow runtime model (in-memory) |
|---|---|---|
| A stage | `FlowStage` (discriminated union on `kind`) at [flow/types.ts:17](apps/server/src/services/outbound/flow/types.ts) | `Node<StageNodeData, kind>` |
| Topology | implicit: `after?: string[]` per stage + gate `onPass`/`onFail` | explicit: `Edge<FlowEdgeData>[]` |
| Position | **none** — stages have no x/y | `node.position: { x, y }` |
| Identity | `stage.id` | `node.id` (kept **equal** to `stage.id`) |
| Kind tag | `stage.kind` | `node.type` (kept **equal** to `stage.kind`) |
| Run/validation state | lives in `flow_runs` / `ConfigValidation`, keyed by `stageId` | merged onto `node.data.run` / `node.data.validation` for paint |
The invariants that make the adapter total: **`node.id === stage.id`** and **`node.type === stage.kind`**. Those two equalities are what let every keyed lookup (run counts, validation errors, edge endpoints) round-trip without a side table.
### A.2 The per-kind typed node model
React Flow v12's node generic is `Node<TData extends Record<string, unknown>, TType extends string>`. We instantiate it once per stage kind by carrying the *exact* `FlowStage` variant inside `data.stage`, so the node component for a kind gets a fully-narrowed stage. The config is **never flattened** into loose node-data keys — `data.stage` stays the single source of truth. (Gumloop's `OperatorNode` similarly nests its config under `node.data` — `name`/`parameters`/`declaration` — rather than spreading it onto the node; we keep that nesting but type it per-kind instead of using gumloop's single `OperatorNodeData` shape.)
```ts
// apps/mail/modules/outbound/components/flow-canvas/graph/node-types.ts
import type { Node, Edge } from '@xyflow/react';
import type { FlowStage } from '@/modules/outbound/types/flow'; // re-export of server FlowStage union
// One data payload per node. `stage` is the persisted config (narrowed by the
// generic S); the other two are derived UI overlays that are NEVER serialized.
export type StageNodeData<S extends FlowStage = FlowStage> = {
stage: S;
validation?: { code: string; message: string }[]; // from ConfigValidation, filtered by stageId
run?: { in: number; out: number; exited: number;
status: 'idle' | 'running' | 'done' | 'exited' }; // from RunFlowResult.stages[], keyed by stageId
} & Record<string, unknown>; // satisfies RF's TData constraint
// Narrow each stage kind into its own Node type. `Extract` pulls the matching
// FlowStage variant, so CedarInputNode.data.stage is exactly the cedarInput shape.
export type CedarInputNode = Node<StageNodeData<Extract<FlowStage, { kind: 'cedarInput' }>>, 'cedarInput'>;
export type QualifierNode = Node<StageNodeData<Extract<FlowStage, { kind: 'qualifier' }>>, 'qualifier'>;
export type GateNode = Node<StageNodeData<Extract<FlowStage, { kind: 'gate' }>>, 'gate'>;
export type ExpandNode = Node<StageNodeData<Extract<FlowStage, { kind: 'expand' }>>, 'expand'>;
export type ListNode = Node<StageNodeData<Extract<FlowStage, { kind: 'list' }>>, 'list'>;
export type StageNode = CedarInputNode | QualifierNode | GateNode | ExpandNode | ListNode;
```
The `nodeTypes` registry maps the `kind` string → component (the standard `@xyflow/react` pattern; cf. gumloop `PipelineDiagram.tsx` `nodeTypes` ≈L109-115). Each component is typed with `NodeProps<TheSpecificNode>`, so `props.data.stage` is already narrowed — no `switch` inside the component:
```ts
// apps/mail/modules/outbound/components/flow-canvas/nodes/index.ts
import type { NodeTypes, NodeProps } from '@xyflow/react';
export const STAGE_NODE_TYPES = {
cedarInput: CedarInputNode_, qualifier: QualifierNode_, gate: GateNode_,
expand: ExpandNode_, list: ListNode_,
} satisfies NodeTypes;
// nodes/GateNode.tsx
export function GateNode_({ data, selected }: NodeProps<GateNode>) {
const g = data.stage; // narrowed to { kind:'gate'; on?; op; value?; onPass?; onFail? }
// …render `${g.on} ${g.op} ${String(g.value)}`, the run overlay, the validation ring…
}
```
### A.3 Edge + handle model — how topology is encoded
Edges carry a single discriminator: which **relation** they represent. A plain dependency is `after`; a gate's two outgoing routes are `pass` and `fail`. That relation is the only thing the reverse transform needs.
```ts
// graph/node-types.ts (cont.)
export type EdgeRelation = 'after' | 'pass' | 'fail';
export type FlowEdgeData = { relation: EdgeRelation } & Record<string, unknown>;
export type FlowEdge = Edge<FlowEdgeData>;
```
**Edge id convention** is reversible: `` `${source}:${relation}->${target}` ``. From an id alone you can recover both endpoints and the relation, which keeps `onEdgesChange` deletes trivial to reverse-map.
**Handle (port) schema** — handles are how `onConnect` knows which relation a new edge encodes. The source handle id *is* the relation:
| Node kind | target handle (top) | source handle(s) (bottom) |
|---|---|---|
| `cedarInput` (root) | — (originates rows) | `out` |
| `qualifier` / `expand` / `list` | `in` | `out` |
| `gate` | `in` | `pass` (green) **and** `fail` (red) |
```ts
// nodes/GateNode.tsx — two distinct source handles
<Handle type="target" position={Position.Top} id="in" />
<Handle type="source" position={Position.Bottom} id="pass" style={{ left: '33%' }} />
<Handle type="source" position={Position.Bottom} id="fail" style={{ left: '66%' }} />
```
```ts
// onConnect: the sourceHandle dictates the relation
const relation: EdgeRelation =
conn.sourceHandle === 'pass' ? 'pass' : conn.sourceHandle === 'fail' ? 'fail' : 'after';
```
**Connection constraints (the builder's whole gate UX).** A gate node has exactly two outputs and the two edges you draw from them *are* the routing — you never touch `after`/`onPass`/`onFail` on the canvas:
- A gate's `pass` and `fail` handles each accept **at most one** outgoing edge (a route is single-target); `onConnect` replaces an existing edge on the same handle.
- **`fail` left unconnected ⇒ `onFail:'exit'`** (the common "drop sub-threshold tokens" case) — render the bare handle with a small red "✕ exit" hint so it reads as intentional.
- **`fail` connected to a real stage ⇒ the nurture branch** (`onFail` = that stage id).
- `pass` left unconnected is a lint-warn (a gate that passes to nowhere is almost always a misconfig).
- A non-gate `out` handle may fan out to several successors (several stages naming it in `after`).
### A.4 `stagesToFlow` — domain → React Flow
```ts
// graph/stagesToFlow.ts
export function stagesToFlow(flow: Flow): { nodes: StageNode[]; edges: FlowEdge[] }
```
**Nodes.** One per stage: `{ id: stage.id, type: stage.kind, data: { stage }, position }`. Position resolution: `flow.defaults.layout?.[stage.id]` if present, else a one-shot dagre top-to-bottom pass (`graph/autoLayout.ts`) so agent/config-authored flows — which have no saved positions — still render laid-out on first open. Positions are **static** after that (gumloop-style): the user drags, we persist; **no force simulation ever runs**. `cedarInput` nodes get `draggable: true` but render without a target handle.
**Edges.** **Every edge on the canvas is an `after`-edge** (the orchestrator's only topology source — `buildSuccessors`/`routeNext`/`routeGate` all read `successors`, built solely from `after`, at [orchestrator.ts:84-99](apps/server/src/services/outbound/flow/run/orchestrator.ts)). A gate doesn't get a *different kind* of edge — it gets the **same `after`-edges, just tagged** `pass`/`fail` by which source handle they leave from. The per-source rule:
- **Non-gate source `S`:** for every stage `T` with `S.id ∈ T.after`, emit one `after` edge `S → T` (`sourceHandle:'out'`).
- **Gate source `G`:** `G` has two outgoing `after`-edges at most — tag the one to `resolve(onPass)` as `pass` (leaving `sourceHandle:'pass'`) and the one to `onFail` (when it's a real stage id) as `fail`. `resolve(onPass)` = explicit `onPass`, else the single successor `T` with `G.id ∈ T.after`. When `onFail` is `'exit'` (the default), there is no `after`-edge — draw a `fail` stub to a synthetic `__exit` sink.
So topology is uniform (`after` everywhere); the gate's two handles only *color/route* its own out-edges, plus the `'exit'` stub that `after` can't express. Worked example — the linear flow from [flow.md II.7](apps/server/docs/outbound/core/flow.md):
```jsonc
// DOMAIN (stages[] + defaults.layout)
"stages": [
{ "id":"src", "kind":"cedarInput", "cedarInputId":"in-rb2b" },
{ "id":"score", "kind":"qualifier", "qualifierId":"q-basic", "after":["src"] },
{ "id":"gate7", "kind":"gate", "on":"score", "op":"gte", "value":7,
"onFail":"exit", "after":["score"] },
{ "id":"deep", "kind":"qualifier", "qualifierId":"q-deep", "after":["gate7"] },
{ "id":"land", "kind":"list", "listName":"AE warm", "listKind":"contact", "after":["deep"] }
],
"defaults": { "layout": { "src":{"x":0,"y":0}, "score":{"x":0,"y":120}, "gate7":{"x":0,"y":240},
"deep":{"x":0,"y":360}, "land":{"x":0,"y":480} } }
```
```jsonc
// REACT FLOW (after stagesToFlow)
"nodes": [
{ "id":"src", "type":"cedarInput", "position":{"x":0,"y":0}, "data":{ "stage":{…} } },
{ "id":"score", "type":"qualifier", "position":{"x":0,"y":120}, "data":{ "stage":{…} } },
{ "id":"gate7", "type":"gate", "position":{"x":0,"y":240}, "data":{ "stage":{…} } },
{ "id":"deep", "type":"qualifier", "position":{"x":0,"y":360}, "data":{ "stage":{…} } },
{ "id":"land", "type":"list", "position":{"x":0,"y":480}, "data":{ "stage":{…} } }
],
"edges": [
{ "id":"src:after->score", "source":"src", "target":"score", "data":{"relation":"after"} },
{ "id":"score:after->gate7","source":"score", "target":"gate7", "data":{"relation":"after"} },
{ "id":"gate7:pass->deep", "source":"gate7", "target":"deep", "sourceHandle":"pass", "data":{"relation":"pass"} },
{ "id":"gate7:fail->__exit","source":"gate7", "target":"__exit","sourceHandle":"fail", "data":{"relation":"fail"} }
]
// note: gate7→deep is the gate's `after`-edge to its one successor, TAGGED `pass` (onPass undefined
// → resolves to deep.after=[gate7]); gate7→__exit is the `fail` stub (onFail:"exit", no after-edge).
```
### A.5 `flowToStages` — React Flow → domain (the inverse, on save)
```ts
// graph/flowToStages.ts
export function flowToStages(
nodes: StageNode[], edges: FlowEdge[],
): { stages: FlowStage[]; layout: Record<string, { x: number; y: number }> }
```
The reconstruction is mechanical because of the invariants in A.1:
1. **Per node → stage:** start from `node.data.stage` (carries all config the inspector edited), then **overwrite its topology fields from the live edge set** so dragging connections is authoritative:
- `after` = the sources of all incoming edges to `node.id` (any relation — a `pass`/`fail` edge still means "depends on the gate"), minus the synthetic exit sink, deduped.
- For a **gate** node: `onPass` = target of its outgoing `pass` edge (or `'exit'` if it points at `__exit`/absent); `onFail` = target of its outgoing `fail` edge (or `'exit'`).
2. **Layout:** `layout[node.id] = node.position` for every non-synthetic node → persisted under `defaults.layout`.
3. **Order:** sort stages by `topoOrder(stages)` ([flow/types.ts:121](apps/server/src/services/outbound/flow/types.ts)) for a stable, diff-friendly serialization.
4. **Drop** the `__exit` sink and any `data.run`/`data.validation` overlays — they are never part of the persisted config.
Feeding the A.4 example back through `flowToStages` yields **semantically identical** `stages` plus the `defaults.layout` map — the round-trip property the Phase-4 test asserts. One deliberate **normalization**: because the two gate edges are always drawn explicitly, the reverse always writes `onPass`/`onFail` explicitly, so a gate authored with a defaulted `onPass` (undefined → natural successor) comes back as an explicit id. That's a clarification, not a change — the orchestrator routes the token to the same stage either way. (Test should assert semantic equality, normalizing defaulted routes, not a byte diff.)
### A.6 Position persistence schema
Positions ride in `flows.defaults.layout` — a `Record<stageId, { x: number; y: number }>`. `defaults` is already a `jsonb` column and `flow.update` accepts `z.any()` ([outbound.ts:130](apps/server/src/trpc/routes/outbound.ts)), so **no migration and no FlowStage change**. The only backend touch (Phase 2) is making `validateConfig` ignore `defaults.layout`. Rationale for not putting x/y on the stage itself: stages are the headless contract shared by MCP/CLI/agent callers that have no canvas; coordinates are a pure-frontend concern, so they live in the frontend-owned `defaults` bag, not on the domain object.
### A.7 The Zustand slice — events → mutations
The slice is a Zustand port of gumloop's `redux/pipeline/flow/flow-slice.ts` — its `nodes`/`edges` state and `onNodesChange`/`onConnect`/`setNode` actions — replacing React Flow's transient `useNodesState`/`useEdgesState` with a persisted store. `node.data.stage` owns config; `edges` own topology; `node.position` owns layout. Every React Flow callback is a thin sink that applies the change (via `applyNodeChanges`/`applyEdgeChanges` from `@xyflow/react`, exactly as gumloop's slice does) and recomputes `dirty`.
```ts
// store/outboundFlowSlice.ts
interface OutboundFlowSlice {
flowId: string | null;
nodes: StageNode[];
edges: FlowEdge[];
selectedStageId: string | null;
dirty: boolean;
validation: ConfigValidation | null;
// ── React Flow event sinks (wired straight to <ReactFlow/>) ──
onNodesChange: (changes: NodeChange<StageNode>[]) => void; // applyNodeChanges; dirty on 'position'|'remove'
onEdgesChange: (changes: EdgeChange<FlowEdge>[]) => void; // applyEdgeChanges; dirty on 'remove'
onConnect: (c: Connection) => void; // validate → addEdge (relation from sourceHandle)
// ── domain mutations (inspector + palette) ──
setGraph: (flow: Flow) => void; // stagesToFlow(flow) → {nodes,edges}; dirty=false
addStage: (kind: FlowStage['kind'], position: XYPosition) => void; // mint stage+node, after=[lastStageId]
updateStage: (id: string, patch: Partial<FlowStage>) => void; // merge into node.data.stage
deleteStage: (id: string) => void; // drop node + incident edges
setStagePosition: (id: string, pos: XYPosition) => void;
// ── serialize ──
toConfig: () => { stages: FlowStage[]; defaults: { layout: Record<string, XYPosition> } }; // flowToStages
}
```
| User action on canvas | React Flow callback | Slice mutation | Touches |
|---|---|---|---|
| Drag a node | `onNodesChange([{type:'position'}])` | `applyNodeChanges` → `node.position`; `dirty=true` | layout |
| Drop palette chip | `onDrop` → `screenToFlowPosition` | `addStage(kind, pos)` | new stage + node |
| Drag handle → handle | `onConnect(conn)` | validate, then add `FlowEdge` w/ `relation` | topology (`after`/`onPass`/`onFail`) |
| Delete edge | `onEdgesChange([{type:'remove'}])` | `applyEdgeChanges`; `dirty=true` | topology |
| Edit a field in inspector | — | `updateStage(id, patch)` | `node.data.stage` |
| Click "Save" | — | `toConfig()` → `validateConfig` → `flow.update` | persist |
### A.8 Overlay merges — validation + run state (keyed by `stageId`)
Both overlays exploit `node.id === stage.id`: they `map` over nodes and merge a derived field into `data`, which triggers React Flow to re-render just those nodes.
```ts
// after trpc.outbound.flow.validateConfig → ConfigValidation { errors: [{ code, message, stageId? }] }
setNodes((ns) => ns.map((n) => ({
...n,
data: { ...n.data, validation: errors.filter((e) => e.stageId === n.id) },
})));
// while polling trpc.outbound.flow.getRun → RunFlowResult.stages: [{ stageId, in, out, exited }]
setNodes((ns) => ns.map((n) => {
const s = result.stages.find((x) => x.stageId === n.id);
return s ? { ...n, data: { ...n.data, run: { ...s, status: deriveStatus(result.status, s) } } } : n;
}));
```
A node paints an error ring when `data.validation?.length`, and a counts badge + status border when `data.run` is set — identical to gumloop's run recolor, but sourced from `RunFlowResult.stages[]` ([flow/types.ts:110](apps/server/src/services/outbound/flow/types.ts)) instead of a websocket.
### A.9 End-to-end type flow
```text
PERSISTED (jsonb) IN-MEMORY (React Flow + Zustand) BACK TO PERSISTED
──────────────── ─────────────────────────────── ─────────────────
flows.stages: FlowStage[] ──┐ ┌──► flow.update({ patch:
flows.defaults.layout ──┼─ stagesToFlow(flow) ─► { nodes: StageNode[], stages, defaults.layout })
│ edges: FlowEdge[] } ▲
│ │ ▲ │
flow.get ──────────────────── │ │ edits │ flowToStages(nodes, edges)
onNodes/Edges/Connect │
inspector updateStage ─┘ │ │
ConfigValidation.errors[] ──── filter by stageId ─► node.data.validation │
RunFlowResult.stages[] ──── find by stageId ─► node.data.run │
(overlays, never persisted)┘
Invariants holding it together: node.id === stage.id · node.type === stage.kind
Ownership: data.stage = config · edges = topology · node.position = layout
```
### A.10 The Inputs palette — passive vs active, backfill vs live, with a throughput log
Every input is a root node, but the palette splits them on **how rows arrive**, which is the distinction operators actually reason about:
- **Active** — *we pull*. A user action or the engine on a cadence fetches rows. The user interacts (uploads, clicks "pull now", sets a schedule). CSV, a LinkedIn historical backfill, a Crustdata search, a web crawler.
- **Passive** — *rows are pushed to us* and fire **automatically**, no user action per row. An external system POSTs the API door; a LinkedIn webhook subscription fires the instant someone reacts.
This is the [`trigger`](apps/server/src/services/outbound/cedar-inputs/index.ts) the backend already records on every run — `manual`/`poll` are active, `subscription` is passive — surfaced as the organizing axis of the palette:
```text
Inputs ▾
ACTIVE (we pull — user/engine initiated)
├─ CSV upload door · csv — user uploads/pastes a file
├─ LinkedIn — backfill cedarInput · signal — "pull PAST reactions" (one-shot deep pull, now)
├─ Crustdata search cedarInput · signal — market-activity query, polled on a cadence
└─ Web crawler cedarInput · crawler — crawl a site on a schedule
PASSIVE (pushed to us — auto-fires into the pipeline)
├─ LinkedIn — live cedarInput · signal · subscription — "from now on, new reactions auto-fire"
└─ API / Adapter door · api — RB2B/Apollo/form POST to the auto-provisioned endpoint
```
**LinkedIn is one source with two modes (your #1 + #3).** A single LinkedIn source (provider `linkedin`, `signal` dropdown: `my_post_engagement` | `profile_viewers` | `accepted_connections`) exposes two run modes, not two sources:
- **Backfill (active, one-shot):** "pull PAST reactions" → a single deep `fireCedarInput({ trigger:'manual' })` with high page limits (`postMaxPages`/`reactionMaxPages`), stamping `highWaterAt`. Catches up history once.
- **Live (passive, going-forward):** "from now on, new signals" → a **webhook subscription**. A LinkedIn reaction lands on the Unipile webhook → the receiver calls [`emitToBoundProviderInput({ provider:'linkedin', row, trigger:'subscription' })`](apps/server/src/services/outbound/cedar-inputs/index.ts) → the row runs the same dedupe/cooldown/dispatch path and **auto-fires into the downstream qualifier**. No polling, no user action. (The service hook exists; the missing piece is wiring the Unipile reaction webhook to it — a build task below.) Dedup against `highWaterAt` + the cooldown ledger means backfill-then-live never double-emits the same person.
The dogfooded 17 `my_post_engagement` inputs were exactly the anti-pattern this kills: that's **one** LinkedIn source, backfilled once and then left live — not 17 sources and not one-per-flow. A flow references a saved source by id, so one LinkedIn input feeds many flows.
**Throughput log on every input node (your #4).** Each input node carries a small overlay: **leads funneled into the next node over a window (default last 7 days)**, plus a sparkline. It reads the input's own runs — `sum(cedar_input_runs.counts.emitted)` over the window — and follows `dispatchRunId` to the downstream qualifier run for the "→ qualifier" count, so the node shows e.g. `142 emitted → 138 qualified · last 7d` with a window selector (24h / 7d / 30d). Same shape as the run overlay in A.8, but time-windowed and sourced from the cedar-input runs/items rather than a single flow run. Backed by a new `cedarInput.throughput({ id, since })` query (aggregates `cedar_input_runs`/`_items`; `status='emitted'` is the numerator).
**Storage split behind the unified palette (for A.2).** Active/passive is a UX axis; underneath, signals/crawlers are `cedarInput` **stages** (`flow.stages[]` + a `cedar_inputs` row) while CSV/API are input **doors** (`flow.inputs[]`). The canvas renders both as leftmost root nodes; `addStage(kind, position, sourceType)` routes accordingly. The `cedarInput` node shows a provider/signal badge + mode (LinkedIn ▸ post-engagement ▸ live) from the resolved `cedar_inputs` row.
**Follow-up build tasks** (canvas was first built with one generic input node):
- [ ] Split `StagePalette` Inputs into **Active** (CSV · LinkedIn backfill · Crustdata · crawler) and **Passive** (LinkedIn live · API/Adapter) groups.
- [ ] LinkedIn source: one entry, `signal` dropdown + **mode** (backfill one-shot / live subscription); dedup saved sources by `(provider, signal)` so one source feeds many flows.
- [ ] Wire the **Unipile reaction webhook → `emitToBoundProviderInput`** so live LinkedIn signals auto-fire (passive path).
- [ ] CSV/API **door root node** backed by `flow.inputs[]` (distinct from `cedarInput` stages).
- [ ] `cedarInput.throughput({ id, since })` endpoint + the node overlay (default last 7d, window selector) reading `cedar_input_runs.counts.emitted` + `dispatchRunId`.