Find a flow's declared inputs and outputs in the global search

/search lists each input and batch output under its qualified message,
and picking one opens the flow on the label's id, which the editor now
treats like a click on the label: it opens the flow's settings.

backend/fluksio/api/routes/search.py

@@ -16,6 +16,7 @@ from fluksio.api.routes.alerts import read_config as read_alerts_config
 from fluksio.flow import modules, panels
 from fluksio.flow.controller import FlowController
 from fluksio.flow.dashboards import DashboardNotFound, DashboardStore
+from fluksio.flow.messages import qualify
 from fluksio.flow.secrets import get_secrets
 from fluksio.flow.store import FlowNotFound
 
@@ -29,6 +30,8 @@ router = APIRouter(
 Category = Literal[
     "flow",
     "node",
+    "input",
+    "output",
     "dashboard",
     "widget",
     "panel",
@@ -48,13 +51,15 @@ class SearchEntry(BaseModel):
     """
 
     category: Category
-    #: The id the frontend routes on.
+    #: The id the frontend routes on. For an input or output, the qualified
+    #: message, which is what the canvas names its label by.
     name: str
     #: Human title, often empty — a flow is usually only its name.
     title: str = ""
-    #: The flow a node sits in, or the dashboard a widget sits on.
+    #: The flow a node, input or output sits in, or the dashboard a widget
+    #: sits on.
     parent: str = ""
-    #: Node type, widget type, channel kind.
+    #: Node type, input dtype, widget type, channel kind.
     kind: str = ""
 
 
@@ -84,6 +89,27 @@ def _build(
             )
             for node in flow.nodes
         )
+        # Its boundary, as the canvas draws it: a live flow never finishes, so
+        # only a batch one has outputs to report.
+        boundary: list[tuple[Category, str, str]] = [
+            ("input", flow_input.spec.name, flow_input.spec.dtype.value)
+            for flow_input in flow.inputs
+        ]
+        if flow.mode == "batch":
+            boundary += [("output", output, "") for output in flow.outputs]
+        for category, declared, kind in boundary:
+            message = qualify(flow.name, declared)
+            if message:
+                entries.append(
+                    SearchEntry(
+                        category=category,
+                        name=message,
+                        # The label the canvas draws: bare when it is the flow's own.
+                        title=message.removeprefix(f"{flow.name}."),
+                        parent=flow.name,
+                        kind=kind,
+                    )
+                )
 
     for summary in dashboards.list():
         try:

backend/tests/api/routes/test_search.py

@@ -61,6 +61,29 @@ def test_index_reaches_inside_flows_and_dashboards(
     assert found[("widget", "hall", "temperature")]["kind"] == "stat"
 
 
+def test_index_lists_a_flows_inputs_and_outputs(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """Named by the qualified message, which is the label the canvas draws."""
+    client.put(
+        f"{FLOWS}/fit",
+        headers=superuser_token_headers,
+        json={
+            "name": "fit",
+            "mode": "batch",
+            "inputs": [{"spec": {"name": "epochs", "dtype": "int"}}],
+            "outputs": ["accuracy"],
+        },
+    )
+
+    entries = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()
+    found = {(e["category"], e["name"]): e for e in entries if e["parent"] == "fit"}
+
+    assert found[("input", "fit.epochs")]["title"] == "epochs"
+    assert found[("input", "fit.epochs")]["kind"] == "int"
+    assert found[("output", "fit.accuracy")]["title"] == "accuracy"
+
+
 def test_secrets_are_named_only_to_a_superuser(
     client: TestClient,
     superuser_token_headers: dict[str, str],