Add flow settings, pulse emitting nodes, and simplify node state

- One dot per node now carries the whole story: primary while running, sage
  after a good run, red when anything is wrong, with the explanation on hover.
  The corner badge is gone, along with the second way of saying the same thing.
- A node that publishes something flashes a ring, so a running flow is legible
  without reading the edge values. Nodes that consume but publish nothing stay
  quiet, which is why the event carries an output count.
- Flow settings open in the same panel its nodes use, from a pencil in the
  dock: the title, the name, and deleting the flow. NodePanel and FlowPanel
  share the panel chrome rather than each drawing their own.
- Renaming is a server operation, because a flow's name is the namespace of its
  messages: the directory moves and every other flow reading `old.message` is
  repointed, instead of being left pointing at a flow that no longer exists.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016WzrvW7rjQbynnhF6pxh6i

backend/app/api/routes/flows.py

@@ -21,6 +21,7 @@ from app.flow.events import event_bus
 from app.flow.messages import qualify
 from app.flow.pipeline import ValidationIssue
 from app.flow.schemas import (
+    NAME_PATTERN,
     FlowDef,
     FlowsPublic,
     FlowStatePublic,
@@ -30,7 +31,7 @@ from app.flow.schemas import (
     NodeStatusPublic,
     NodeTypeInfo,
 )
-from app.flow.store import FlowNotFound
+from app.flow.store import FlowExists, FlowNotFound
 from app.models import Message
 
 router = APIRouter(
@@ -53,6 +54,10 @@ class ValidationResult(BaseModel):
     issues: list[ValidationIssue] = []
 
 
+class RenameRequest(BaseModel):
+    new_name: str
+
+
 class RunRequest(BaseModel):
     inputs: dict[str, Any] = {}
 
@@ -154,6 +159,33 @@ async def delete_flow(name: str, controller: FlowControllerDep) -> Any:
     return Message(message=f"Deleted flow '{name}'")
 
 
+@router.post("/{name}/rename", response_model=FlowDetail)
+async def rename_flow(
+    name: str,
+    body: RenameRequest,
+    controller: FlowControllerDep,
+) -> Any:
+    """Rename a flow, along with every reference to its messages."""
+    if not NAME_PATTERN.match(body.new_name):
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                "Use lowercase letters, digits and underscores, starting with a letter"
+            ),
+        )
+    try:
+        renamed = await run_in_threadpool(
+            controller.store.rename_flow, name, body.new_name
+        )
+    except FlowNotFound:
+        raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
+    except FlowExists as exc:
+        raise HTTPException(status_code=409, detail=str(exc))
+
+    await controller.reload()
+    return _detail(controller, renamed)
+
+
 # -----------------------------------------------------------------------------
 # Node source
 # -----------------------------------------------------------------------------

backend/app/flow/pipeline.py

@@ -338,6 +338,9 @@ class Pipeline:
                     "type": "node_executed",
                     "flow": node.flow,
                     "node": node.id,
+                    # A node that returns nothing ran but published nothing,
+                    # which is a different thing to show than one that emitted.
+                    "outputs": len(result or {}),
                     "duration_ms": round((time.perf_counter() - started) * 1000, 2),
                     "ts": time.time(),
                 }
@@ -443,6 +446,18 @@ class Pipeline:
                         "ts": ts,
                     }
                 )
+            # An injecting node — an MQTT subscriber, a webhook — publishes
+            # without going through the executor, but it did emit.
+            self._publish(
+                {
+                    "type": "node_executed",
+                    "flow": node.flow,
+                    "node": node.id,
+                    "outputs": len(outputs),
+                    "duration_ms": 0,
+                    "ts": ts,
+                }
+            )
 
         downstream = set(self._get_downstream(node))
         if not downstream:

backend/app/flow/store.py

@@ -34,6 +34,15 @@ class FlowNotFound(KeyError):
         return f"No flow named '{self.name}'"
 
 
+class FlowExists(ValueError):
+    def __init__(self, name: str) -> None:
+        super().__init__(name)
+        self.name = name
+
+    def __str__(self) -> str:
+        return f"There is already a flow named '{self.name}'"
+
+
 class FlowStore:
     """Reads and writes flows, committing every change."""
 
@@ -129,6 +138,50 @@ class FlowStore:
         shutil.rmtree(directory)
         self._commit(f"Delete flow '{name}'")
 
+    def rename_flow(self, name: str, new_name: str) -> FlowDef:
+        """Rename a flow, carrying its nodes and any references to it.
+
+        A flow's name is the namespace of its messages, so other flows reading
+        ``old.temperature`` are rewritten to read ``new.temperature`` — leaving
+        them pointing at a flow that no longer exists would break them silently.
+        """
+        if not self.exists(name):
+            raise FlowNotFound(name)
+        if self.exists(new_name):
+            raise FlowExists(new_name)
+
+        flow = self.read_flow(name)
+        self._flow_dir(name).rename(self._flow_dir(new_name))
+
+        renamed = flow.model_copy(update={"name": new_name})
+        self._flow_file(new_name).write_text(renamed.model_dump_json(indent=2) + "\n")
+
+        for other in self.read_all():
+            if other.name == new_name:
+                continue
+            if self._retarget(other, f"{name}.", f"{new_name}."):
+                self._flow_file(other.name).write_text(
+                    other.model_dump_json(indent=2) + "\n"
+                )
+
+        self._commit(f"Rename flow '{name}' to '{new_name}'")
+        return renamed
+
+    @staticmethod
+    def _retarget(flow: FlowDef, old_prefix: str, new_prefix: str) -> bool:
+        """Point this flow's cross-flow message names at a renamed flow."""
+        changed = False
+        for node in flow.nodes:
+            for specs in (node.requires, node.provides):
+                for position, spec in enumerate(specs):
+                    if spec.name.startswith(old_prefix):
+                        tail = spec.name[len(old_prefix) :]
+                        specs[position] = spec.model_copy(
+                            update={"name": new_prefix + tail}
+                        )
+                        changed = True
+        return changed
+
     # -------------------------------------------------------------------------
     # Node source
     # -------------------------------------------------------------------------

backend/tests/api/routes/test_flows.py

@@ -144,3 +144,65 @@ def test_node_types_are_listed(
     by_type = {entry["type"]: entry for entry in types}
     assert by_type["python"]["has_source"] is True
     assert "properties" in by_type["mqtt"]["params_schema"]
+
+
+def test_rename_flow(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
+
+    response = client.post(
+        f"{PREFIX}/demo/rename",
+        headers=superuser_token_headers,
+        json={"new_name": "demo_renamed"},
+    )
+    assert response.status_code == 200
+    assert response.json()["definition"]["name"] == "demo_renamed"
+
+    assert (
+        client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 404
+    )
+    assert (
+        client.get(
+            f"{PREFIX}/demo_renamed", headers=superuser_token_headers
+        ).status_code
+        == 200
+    )
+
+    # Put it back so the tests that follow find the flow they expect.
+    client.post(
+        f"{PREFIX}/demo_renamed/rename",
+        headers=superuser_token_headers,
+        json={"new_name": "demo"},
+    )
+
+
+def test_rename_onto_a_taken_name_is_refused(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
+    client.put(
+        f"{PREFIX}/occupied", headers=superuser_token_headers, json=a_flow("occupied")
+    )
+
+    response = client.post(
+        f"{PREFIX}/demo/rename",
+        headers=superuser_token_headers,
+        json={"new_name": "occupied"},
+    )
+    assert response.status_code == 409
+
+    client.delete(f"{PREFIX}/occupied", headers=superuser_token_headers)
+
+
+def test_rename_rejects_an_invalid_name(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow())
+
+    response = client.post(
+        f"{PREFIX}/demo/rename",
+        headers=superuser_token_headers,
+        json={"new_name": "Not A Flow Name"},
+    )
+    assert response.status_code == 400

backend/tests/flow/test_store.py

@@ -5,7 +5,7 @@ import pytest
 
 from app.flow.messages import MessageSpec
 from app.flow.schemas import FlowDef, NodeDef
-from app.flow.store import FlowNotFound, FlowStore
+from app.flow.store import FlowExists, FlowNotFound, FlowStore
 
 
 @pytest.fixture
@@ -73,3 +73,47 @@ def test_deleting_removes_flow_and_its_nodes(store: FlowStore):
 
     assert store.list_flows() == []
     assert not (store.root / "heating").exists()
+
+
+def test_renaming_a_flow_carries_its_nodes(store: FlowStore):
+    store.write_flow(a_flow())
+    store.write_node_source(
+        "heating", "sensor", "def process(params):\n    return {}\n"
+    )
+
+    renamed = store.rename_flow("heating", "warmth")
+
+    assert renamed.name == "warmth"
+    assert store.list_flows() == ["warmth"]
+    assert "def process" in store.read_node_source("warmth", "sensor")
+
+
+def test_renaming_a_flow_repoints_the_flows_reading_it(store: FlowStore):
+    store.write_flow(a_flow())
+    store.write_flow(
+        FlowDef(
+            name="display",
+            nodes=[
+                NodeDef(
+                    id="gauge",
+                    # Reads across the flow boundary, so the name must follow.
+                    requires=[MessageSpec(name="heating.temp")],
+                )
+            ],
+        )
+    )
+
+    store.rename_flow("heating", "warmth")
+
+    display = store.read_flow("display")
+    assert display.nodes[0].requires[0].name == "warmth.temp"
+
+
+def test_renaming_onto_an_existing_name_is_refused(store: FlowStore):
+    store.write_flow(a_flow())
+    store.write_flow(FlowDef(name="warmth"))
+
+    with pytest.raises(FlowExists):
+        store.rename_flow("heating", "warmth")
+
+    assert store.list_flows() == ["heating", "warmth"]