diff --git a/README.md b/README.md
index 1b22416..9f06a39 100644
--- a/README.md
+++ b/README.md
@@ -26,7 +26,7 @@ Use the Blacknode editor as the primary surface:
 4. Use Robot Monitor for read-only state and `RobotServo` for preview; arm only after identity, calibration, limits, and fresh feedback are correct.
 5. Use the guided calibration and editable-profile templates when defining a new physical assembly.
 
-Core nodes include `Robot`, `ComputeDevice`, `DeviceInspect`, profile load/save/duplicate nodes, calibration control/recording, capability and attachment nodes, `RobotMonitor`, and `RobotServo`. Local profiles and calibration data live outside package source under `robots/` and are intentionally ignored by Git.
+Core nodes include `Robot`, `ComputeDevice`, `PhysicalRobot`, `RobotDeployment`, `RobotStream`, `DeviceInspect`, profile load/save/duplicate nodes, calibration control/recording, capability and attachment nodes, `RobotMonitor`, and `RobotServo`. Build device workflows as `ComputeDevice` → `PhysicalRobot` → `RobotDeployment` → `RobotStream`. Each node owns one selection and passes credential-free live inspection state forward. Connect `RobotStream.topic` and `RobotStream.message_type` to generic ROS 2 stream inputs, then route the ROS message into a map, camera, LiDAR, IMU, or other compatible viewer or processing node. Local profiles and calibration data live outside package source under `robots/` and are intentionally ignored by Git.
 
 ## Safety
 
diff --git a/blacknode-package.toml b/blacknode-package.toml
index 1e8d5e6..64105f9 100644
--- a/blacknode-package.toml
+++ b/blacknode-package.toml
@@ -1,6 +1,6 @@
 [package]
 name = "blacknode-robot"
-version = "0.5.3"
+version = "0.5.4"
 description = "Robot contracts, connected-device lifecycle, normalized telemetry, profiles, and driver launch."
 requires-blacknode = ">=0.3.0"
 layer = "robot"
@@ -65,7 +65,8 @@ capabilities = ["robot.capabilities"]
 # Declared for the package roadmap; sources land in components/authorization/nodes.
 node-types = [
   "RobotAttachment", "RobotAttachmentList", "RobotCapabilityBinding",
-  "ComputeDevice", "DeviceInspect", "RobotCapabilityInspect",
+  "ComputeDevice", "PhysicalRobot", "RobotDeployment", "RobotStream",
+  "DeviceInspect", "RobotCapabilityInspect",
   "RobotCapabilityList", "RobotCapabilityProfile",
   "RobotConnectionDashboard", "RobotDiscovery", "RobotMonitor",
   "RobotRawMonitor", "RobotRawMonitorMockProvider",
diff --git a/nodes/compute_devices.py b/nodes/compute_devices.py
index 3adb980..6460631 100644
--- a/nodes/compute_devices.py
+++ b/nodes/compute_devices.py
@@ -79,10 +79,8 @@ def compute_device(ctx: dict) -> dict:
         report = "Choose a compute device in the node."
     elif live:
         checked_at = str(inspection.get("checked_at") or "").strip()
-        report = (
-            f"{device_name or device_id}: paired Runtime is live"
-            + (f"; ROS state checked {checked_at}." if checked_at else ".")
-        )
+        report = f"{device_name or device_id}: paired Runtime is live"
+        report += f"; ROS state checked {checked_at}." if checked_at else "."
     else:
         report = (
             f"{device_name or device_id}: selected, but its paired Runtime did "
@@ -97,6 +95,233 @@ def compute_device(ctx: dict) -> dict:
     }
 
 
+def _public_list(value: TypingAny) -> list[dict]:
+    if not isinstance(value, list):
+        return []
+    return [dict(item) for item in _public_value(value) if isinstance(item, dict)]
+
+
+@node(
+    name="PhysicalRobot",
+    component="capabilities",
+    category="Robot",
+    description="Select one physical robot registered under a compute device.",
+    inputs={
+        "device": Dict,
+        "inspection": Dict,
+        "robot_id": Text(default=""),
+        "robot_name": Text(default=""),
+    },
+    outputs={
+        "configured": Bool,
+        "robot": Dict,
+        "inspection": Dict,
+        "report": Text,
+    },
+    primary_inputs=["device", "inspection"],
+    primary_outputs=["robot", "inspection"],
+)
+def physical_robot(ctx: dict) -> dict:
+    device = _public_value(
+        ctx.get("device") if isinstance(ctx.get("device"), dict) else {}
+    )
+    inspection = _public_value(
+        ctx.get("inspection") if isinstance(ctx.get("inspection"), dict) else {}
+    )
+    robot_id = str(ctx.get("robot_id") or "").strip()
+    robot_name = str(ctx.get("robot_name") or "").strip()
+    robots = _public_list(inspection.get("robots"))
+    selected = next(
+        (item for item in robots if str(item.get("id") or "") == robot_id),
+        None,
+    )
+    if selected is None and not robot_id and len(robots) == 1:
+        selected = robots[0]
+    selected_id = str((selected or {}).get("id") or robot_id).strip()
+    selected_name = str((selected or {}).get("name") or robot_name).strip()
+    live = bool(inspection.get("ok") and inspection.get("live"))
+    robot = {
+        "kind": "blacknode.physical-robot-target",
+        "schema_version": 1,
+        "device_id": str(device.get("device_id") or ""),
+        "device_name": str(device.get("device_name") or ""),
+        "robot_id": selected_id,
+        "robot_name": selected_name,
+        "configured": bool(selected_id),
+        "live": live,
+        "read_only": True,
+    }
+    if selected_id:
+        report = f"Selected physical robot {selected_name or selected_id}."
+    elif len(robots) > 1:
+        report = "Choose one physical robot in Properties."
+    elif not device.get("device_id"):
+        report = "Connect a Compute Device node first."
+    else:
+        report = "No physical robot is registered under this compute device."
+    return {
+        "configured": bool(selected_id),
+        "robot": robot,
+        "inspection": inspection,
+        "report": report,
+    }
+
+
+@node(
+    name="RobotDeployment",
+    component="capabilities",
+    category="Robot",
+    description="Select one deployment that belongs to a physical robot.",
+    inputs={
+        "robot": Dict,
+        "inspection": Dict,
+        "deployment_id": Text(default=""),
+        "deployment_name": Text(default=""),
+    },
+    outputs={
+        "selected": Bool,
+        "deployment": Dict,
+        "inspection": Dict,
+        "report": Text,
+    },
+    primary_inputs=["robot", "inspection"],
+    primary_outputs=["deployment", "inspection"],
+)
+def robot_deployment(ctx: dict) -> dict:
+    robot = _public_value(
+        ctx.get("robot") if isinstance(ctx.get("robot"), dict) else {}
+    )
+    inspection = _public_value(
+        ctx.get("inspection") if isinstance(ctx.get("inspection"), dict) else {}
+    )
+    deployment_id = str(ctx.get("deployment_id") or "").strip()
+    deployment_name = str(ctx.get("deployment_name") or "").strip()
+    robot_id = str(robot.get("robot_id") or "").strip()
+    deployments = [
+        item for item in _public_list(inspection.get("deployments"))
+        if not robot_id or str(item.get("target_device_id") or "") in {"", robot_id}
+    ]
+    selected = next(
+        (item for item in deployments if str(item.get("id") or "") == deployment_id),
+        None,
+    )
+    if selected is None and not deployment_id and len(deployments) == 1:
+        selected = deployments[0]
+    deployment = dict(selected) if selected else {
+        "kind": "blacknode.robot-deployment",
+        "schema_version": 1,
+        "id": deployment_id,
+        "name": deployment_name,
+        "target_device_id": robot_id,
+        "state": "unavailable",
+        "available": False,
+    }
+    if selected:
+        report = f"Selected deployment {selected.get('name') or selected.get('id')}."
+    elif len(deployments) > 1:
+        report = "Choose one robot deployment in Properties."
+    elif not robot_id:
+        report = "Connect a Physical Robot node first."
+    else:
+        report = "No deployment is available for this physical robot."
+    return {
+        "selected": bool(selected),
+        "deployment": deployment,
+        "inspection": inspection,
+        "report": report,
+    }
+
+
+@node(
+    name="RobotStream",
+    component="capabilities",
+    category="Robot",
+    description="Select one deployed robot capability stream for downstream nodes.",
+    inputs={
+        "robot": Dict,
+        "deployment": Dict,
+        "inspection": Dict,
+        "capability": Text(default=""),
+        "topic": Text(default=""),
+        "message_type": Text(default=""),
+    },
+    outputs={
+        "available": Bool,
+        "stream": Dict,
+        "topic": Text,
+        "message_type": Text,
+        "report": Text,
+    },
+    primary_inputs=["robot", "deployment", "inspection"],
+    primary_outputs=["stream", "topic", "message_type"],
+)
+def robot_stream(ctx: dict) -> dict:
+    robot = _public_value(
+        ctx.get("robot") if isinstance(ctx.get("robot"), dict) else {}
+    )
+    deployment = _public_value(
+        ctx.get("deployment") if isinstance(ctx.get("deployment"), dict) else {}
+    )
+    inspection = _public_value(
+        ctx.get("inspection") if isinstance(ctx.get("inspection"), dict) else {}
+    )
+    capability = str(ctx.get("capability") or "").strip()
+    topic = str(ctx.get("topic") or "").strip()
+    message_type = str(ctx.get("message_type") or "").strip()
+    robot_id = str(robot.get("robot_id") or "").strip()
+    deployment_id = str(deployment.get("id") or "").strip()
+    streams = [
+        item for item in _public_list(inspection.get("streams"))
+        if (not robot_id or str(item.get("robot_id") or "") in {"", robot_id})
+        and (
+            not deployment_id
+            or str(item.get("deployment_id") or "") in {"", deployment_id}
+        )
+        and (not capability or str(item.get("capability") or "") == capability)
+    ]
+    selected = next(
+        (
+            item for item in streams
+            if str(item.get("topic") or "") == topic
+            and (
+                not message_type
+                or str(item.get("message_type") or "") == message_type
+            )
+        ),
+        None,
+    )
+    if selected is None and not topic and len(streams) == 1:
+        selected = streams[0]
+    stream = dict(selected) if selected else {
+        "kind": "blacknode.deployed-stream",
+        "schema_version": 1,
+        "source": "saved_selection",
+        "capability": capability,
+        "device_id": str(robot.get("device_id") or ""),
+        "robot_id": robot_id,
+        "deployment_id": deployment_id,
+        "state": "unavailable",
+        "available": False,
+        "topic": topic,
+        "message_type": message_type,
+    }
+    if selected:
+        report = f"Selected {selected.get('capability') or 'robot'} stream {selected.get('topic')}."
+    elif len(streams) > 1:
+        report = "Choose one capability stream in Properties."
+    elif not robot_id:
+        report = "Connect a Physical Robot node first."
+    else:
+        report = "No matching deployed stream is available."
+    return {
+        "available": bool(stream.get("available")),
+        "stream": stream,
+        "topic": str(stream.get("topic") or topic),
+        "message_type": str(stream.get("message_type") or message_type),
+        "report": report,
+    }
+
+
 @node(
     name="DeviceInspect",
     component="capabilities",
diff --git a/pyproject.toml b/pyproject.toml
index fa2053a..8106e5c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "blacknode-robot"
-version = "0.5.3"
+version = "0.5.4"
 description = "Robot contracts, connected-device lifecycle, and normalized telemetry for Blacknode."
 requires-python = ">=3.11"
 dependencies = ["pyserial>=3.5", "feetech-servo-sdk>=1.0", "roslibpy>=1.5"]
diff --git a/tests/test_compute_devices.py b/tests/test_compute_devices.py
index 8d9663e..3b644e9 100644
--- a/tests/test_compute_devices.py
+++ b/tests/test_compute_devices.py
@@ -2,6 +2,7 @@
 from pathlib import Path
 
 import blacknode  # noqa: F401
+from blacknode.graph import Graph
 from blacknode.node import _NODE_REGISTRY
 from blacknode.workflow import validate_workflow
 
@@ -30,11 +31,36 @@ def _inspection():
             },
             "report": "Generic ROS 2 capability discovery",
         },
+        "robots": [{"id": "robot-01", "name": "Workshop Rover"}],
+        "deployments": [{
+            "id": "map-01",
+            "name": "Room map",
+            "state": "running",
+            "target_device_id": "robot-01",
+            "mapping_control_count": 1,
+            "mapping_topic": "/map",
+        }],
+        "streams": [{
+            "kind": "blacknode.deployed-stream",
+            "schema_version": 1,
+            "source": "deployment",
+            "capability": "map",
+            "device_id": "jetson-01",
+            "robot_id": "robot-01",
+            "deployment_id": "map-01",
+            "state": "running",
+            "available": True,
+            "topic": "/map",
+            "message_type": "nav_msgs/msg/OccupancyGrid",
+        }],
     }
 
 
 def test_compute_device_nodes_are_registered():
     assert _NODE_REGISTRY["ComputeDevice"]._bn_package == "blacknode-robot"
+    assert _NODE_REGISTRY["PhysicalRobot"]._bn_package == "blacknode-robot"
+    assert _NODE_REGISTRY["RobotDeployment"]._bn_package == "blacknode-robot"
+    assert _NODE_REGISTRY["RobotStream"]._bn_package == "blacknode-robot"
     assert _NODE_REGISTRY["DeviceInspect"]._bn_package == "blacknode-robot"
     assert _NODE_REGISTRY["ComputeDevice"]._bn_primary_outputs == [
         "device",
@@ -67,6 +93,64 @@ def test_compute_device_emits_only_a_stable_public_handle():
     assert "runtime_token" not in serialized
 
 
+def test_device_robot_deployment_stream_nodes_form_a_connectable_flow():
+    device = _NODE_REGISTRY["ComputeDevice"]({
+        "device_id": "jetson-01",
+        "device_name": "Workshop Jetson",
+        "inspection": _inspection(),
+    })
+    robot = _NODE_REGISTRY["PhysicalRobot"]({
+        "device": device["device"],
+        "inspection": device["inspection"],
+        "robot_id": "robot-01",
+        "robot_name": "Workshop Rover",
+    })
+    deployment = _NODE_REGISTRY["RobotDeployment"]({
+        "robot": robot["robot"],
+        "inspection": robot["inspection"],
+        "deployment_id": "map-01",
+    })
+    stream = _NODE_REGISTRY["RobotStream"]({
+        "robot": robot["robot"],
+        "deployment": deployment["deployment"],
+        "inspection": deployment["inspection"],
+        "topic": "/map",
+        "message_type": "nav_msgs/msg/OccupancyGrid",
+        "capability": "map",
+    })
+
+    assert robot["robot"]["robot_id"] == "robot-01"
+    assert deployment["deployment"]["id"] == "map-01"
+    assert stream["stream"]["available"] is True
+    assert stream["stream"]["deployment_id"] == "map-01"
+    assert stream["topic"] == "/map"
+    assert stream["message_type"] == "nav_msgs/msg/OccupancyGrid"
+
+
+def test_modular_device_flow_cooks_as_a_graph():
+    graph = Graph()
+    device = graph.node(
+        "ComputeDevice",
+        device_id="jetson-01",
+        device_name="Workshop Jetson",
+        inspection=_inspection(),
+    )
+    robot = graph.node("PhysicalRobot")
+    deployment = graph.node("RobotDeployment")
+    stream = graph.node("RobotStream", capability="map")
+
+    device.out("device") >> robot.inp("device")
+    device.out("inspection") >> robot.inp("inspection")
+    robot.out("robot") >> deployment.inp("robot")
+    robot.out("inspection") >> deployment.inp("inspection")
+    robot.out("robot") >> stream.inp("robot")
+    deployment.out("deployment") >> stream.inp("deployment")
+    deployment.out("inspection") >> stream.inp("inspection")
+
+    assert stream.cook("topic") == "/map"
+    assert stream.cook("message_type") == "nav_msgs/msg/OccupancyGrid"
+
+
 def test_device_inspect_exposes_read_only_inventory_and_candidates():
     selected = _NODE_REGISTRY["ComputeDevice"]({
         "device_id": "jetson-01",