diff --git a/README.md b/README.md
index c843a83..f8ea3a4 100644
--- a/README.md
+++ b/README.md
@@ -1,12 +1,13 @@
 # BLE_Blockchain
 
-卒論向けに、Raspberry Pi 複数台で BLE スキャン結果を暗号化・署名して L2CAP で交換し、過半数合意でブロックチェーンを構築するシステムです。
+卒論向けに、複数台の端末（Raspberry Pi 等）で BLE スキャン結果を暗号化・署名して交換し、過半数合意でブロックチェーンを構築するシステムです。
 
 ## システムの概要
 
-- Raspberry Pi（4 台想定）が BLE ビーコン周辺をスキャンし、事前登録 CSV と突合したデータを扱う
+- Raspberry Pi（4 台想定）などの端末が BLE ビーコン周辺をスキャンし、事前登録 CSV と突合したデータを扱う
 - ペイロードは **ECDSA 署名** と **AES-256-GCM 暗号化** のあと JSON でシリアライズする
-- Pi 間は **Bluetooth L2CAP**（`l2cap_client` / `l2cap_server`）で通信する
+- 端末間のペイロード交換は **Transport 抽象層**（`src/ble_blockchain/transport.py`）を介して行い、`config/transport.json` の `mode` で **BLE 方式（既定）** と **file 方式** を切り替えられる
+- BLE 方式は従来どおり **Bluetooth L2CAP**（`l2cap_client` / `l2cap_server`）で通信する。file 方式は Bluetooth を使わず、共有 inbox ディレクトリとスキャン結果 CSV で交換するため、BLE 非対応の環境（macOS 等）でも同一パイプラインが動く
 - 同時送受信を避けるため、`config/runtime_profiles.json` で端末ごとに **ラウンドロビン**（send / receive / sleep）を定義する
 - 検証済みの受信データから **過半数** の報告が揃った `bt_addrs` をブロックに載せ、チェーンを標準出力する
 
@@ -20,11 +21,14 @@ BLE_Blockchain/
 ├── pyproject.toml               # 依存関係・ble-blockchain コンソール定義
 ├── settings1.json … settings4.json
 ├── config/                      # JSON 設定のみ（Python コードはパッケージ内）
+│   └── transport.json           # Transport 方式切替（mode: ble / file）
 ├── data_folder/                 # 事前登録 CSV（paths.json で参照）
 ├── keys/                        # 端末別 ECDSA 鍵（PEM は .gitignore）
 ├── scripts/                     # Pi 向けセットアップ・鍵生成
 ├── src/ble_blockchain/          # アプリケーション本体（import ble_blockchain）
 │   ├── app/main.py              # パイプライン
+│   ├── transport.py             # Transport 抽象層（ble / file 切替）
+│   ├── transport_file.py        # file 方式（共有 inbox ディレクトリ）
 │   ├── ble/                     # スキャン・L2CAP・メッセージ
 │   ├── blockchain/              # チェーン構築・エクスポート・集約
 │   ├── cipher/                  # ECDSA・AES-256-GCM
@@ -42,10 +46,13 @@ BLE_Blockchain/
 |------|------|
 | `main.py` | 後方互換ラッパー（`python main.py` → パッケージ本体） |
 | `src/ble_blockchain/app/main.py` | パイプライン全体（設定読込 → ペイロード生成 → 送受信 → チェーン出力） |
-| `src/ble_blockchain/pipeline/send_and_receive.py` | 他 Pi への L2CAP 送信（`SEND`） |
-| `src/ble_blockchain/ble/discover.py` | BLE スキャン（Bleak） |
-| `src/ble_blockchain/ble/l2cap_client.py` / `l2cap_server.py` | L2CAP 送受信（PyBlueZ・Linux / Pi） |
-| `src/ble_blockchain/ble/start_discoverable.py` | `bluetoothctl discoverable on` |
+| `src/ble_blockchain/transport.py` | Transport 抽象層（`load_transport` で ble / file を選択） |
+| `src/ble_blockchain/transport_file.py` | file 方式の Transport（共有 inbox ディレクトリ + scan CSV） |
+| `config/transport.json` | Transport 方式切替（`mode`: `"ble"` 既定 / `"file"`） |
+| `src/ble_blockchain/pipeline/send_and_receive.py` | 他端末への L2CAP 送信（`SEND`。BLE 方式で使用） |
+| `src/ble_blockchain/ble/discover.py` | BLE スキャン（Bleak。BLE 方式で使用） |
+| `src/ble_blockchain/ble/l2cap_client.py` / `l2cap_server.py` | L2CAP 送受信（PyBlueZ・Linux / Pi。BLE 方式で使用） |
+| `src/ble_blockchain/ble/start_discoverable.py` | `bluetoothctl discoverable on`（BLE 方式で使用） |
 | `src/ble_blockchain/ble/message_codec.py` | JSON ペイロードの `pack` / `unpack` |
 | `src/ble_blockchain/pipeline/delete_excess_data.py` | 事前登録 CSV との突合・フィルタ |
 | `src/ble_blockchain/pipeline/pandas_d_encode.py` | DataFrame ↔ CSV bytes |
@@ -127,6 +134,7 @@ export BLE_AES_KEY=$(grep BLE_AES_KEY .env | cut -d= -f2)
 | `config/paths.json` | 事前登録 CSV パス・チェーンエクスポート／集約ディレクトリ（`chain_export_dir`） |
 | `config/blockchain.json` | 過半数比率・最小検証済み受信数・content_hash 一致・集約時の最小 Pi 数など |
 | `config/runtime_profiles.json` | 端末別送受信ステップ |
+| `config/transport.json` | Transport 方式切替（`mode`: `"ble"` / `"file"`、file 設定） |
 
 各 Raspberry Pi には端末専用の設定ファイル（`settings1.json`〜`settings4.json`）を用意しています。
 設定ファイルには、**自端末以外**の Bluetooth アドレス（3 台分）、送受信フローを表す `profile`、`signing_key_path`（端末ごとの永続 ECDSA 秘密鍵）、`public_key_pem`、他 Pi の `peer_public_keys` を記載してください。
@@ -181,9 +189,42 @@ CLI オプション:
 | `require_content_hash_agreement` | 同一 `bt_addrs` 採用時に payload の content_hash 一致を要求 |
 | `min_distinct_devices_for_aggregate` | canonical 集約時に推奨する異なる `device_id` の最小数 |
 
+### Transport 方式の切替
+
+ペイロード交換は `config/transport.json` の `mode` で **BLE 方式**（既定 `"ble"`）と **file 方式**（`"file"`）を切り替えられます。`load_transport()`（`src/ble_blockchain/transport.py`）が起動時にこの設定を読み、`scan` / `start_discoverable` / `send_payload` / `receive_payload` を提供する Transport サービス（`BleTransportService` / `FileTransportService`）を選択します。
+
+```json
+{
+  "mode": "ble",
+  "file": {
+    "inbox_dir": "data/transport/inbox",
+    "scan_csv": "data/transport/scan_results.csv",
+    "sender_id": "device1",
+    "poll_interval_sec": 0.5
+  }
+}
+```
+
+| キー | 内容 |
+|------|------|
+| `mode` | `"ble"`（既定、BLE 方式）または `"file"`（file 方式） |
+| `file.inbox_dir` | file 方式の共有受信ディレクトリ（`data/transport/inbox`） |
+| `file.scan_csv` | file 方式でスキャン結果を読み込む CSV（`data/transport/scan_results.csv`） |
+| `file.sender_id` | 自端末の ID（file 方式で受信ファイルの識別に使う） |
+| `file.poll_interval_sec` | file 方式の受信ポーリング間隔（秒、既定 `0.5`） |
+
+#### file 方式の動かし方
+
+1. `config/transport.json` の `mode` を `"file"` に変更します。
+2. `file.scan_csv` のパス（既定 `data/transport/scan_results.csv`）に、`bt_addrs` / `device_name` の **2 列の CSV** を用意します（BLE 無しでスキャン結果を模擬）。ファイルが無い場合はスキャン結果が空になります。
+3. 受信は共有の inbox ディレクトリ（`file.inbox_dir`）で行います。送信側は受信者 ID をプレフィックスにしたファイルを書き、受信側は自分の `sender_id` をプレフィックスにしたファイルを読みます。
+4. 端末ごとに `file.sender_id` を変えてください（例: `device1` / `device2` / …）。`sender_id` が未指定の場合は `settingsN.json` のファイル名（`settings1` 等）が使われます。
+
+file 方式は Bluetooth を使わないため、macOS 等の BLE/L2CAP 非対応環境でも同一パイプラインを動かせます。
+
 ## 処理の流れ
 
-実装（`src/ble_blockchain/app/main.py`）に基づく全体フローです。**BLE スキャンとペイロード生成は `send` ステップの直前**に行い、その後 `runtime_profiles` の `steps` を順に実行します。署名には `settingsN.json` の永続秘密鍵を使います。
+実装（`src/ble_blockchain/app/main.py`）に基づく全体フローです。**スキャンとペイロード生成は `send` ステップの直前**に行い、その後 `runtime_profiles` の `steps` を順に実行します。スキャン・送信・受信・discoverable はすべて Transport 経由（`transport.scan()` 等）で実行し、BLE 方式なら BLE のスキャン結果を、file 方式なら `scan_results.csv` のスキャン結果を使います。署名には `settingsN.json` の永続秘密鍵を使います。
 
 ### 全体フローチャート
 
@@ -198,7 +239,7 @@ flowchart TD
 
     subgraph build_on_send["build_send_payload（send 直前）"]
         BuildPayload[build_send_payload] --> LoadKey[永続 ECDSA 秘密鍵を読込]
-        LoadKey --> Scan[BLE スキャン discover.py]
+        LoadKey --> Scan["transport.scan()<br/>BLE または file のスキャン結果"]
         Scan --> Filter[pipeline.delete_excess_data<br/>事前登録 CSV と突合]
         Filter --> EncPlain[pandas_encode]
         EncPlain --> Sign[ECDSA 署名]
@@ -208,10 +249,10 @@ flowchart TD
 
     Pack --> StepsLoop{"run_communication_steps<br/>profile の steps を順次"}
 
-    StepsLoop -->|discoverable| Disc[start_discoverable]
+    StepsLoop -->|discoverable| Disc["transport.start_discoverable()<br/>BLE 方式: bluetoothctl discoverable on<br/>file 方式: 何もしない"]
     StepsLoop -->|send| BuildPayload
-    StepsLoop -->|send| Send["SEND → 各 peer へ L2CAP 送信"]
-    StepsLoop -->|receive| Recv[L2CAP 受信]
+    StepsLoop -->|send| Send["transport.send_payload()<br/>BLE 方式: 各 peer へ L2CAP 送信<br/>file 方式: inbox にファイル書き込み"]
+    StepsLoop -->|receive| Recv["transport.receive_payload()<br/>BLE 方式: L2CAP 受信<br/>file 方式: inbox からファイル読み取り"]
     StepsLoop -->|sleep| Sleep[time.sleep]
 
     Recv --> Proc[process_received_payload]
@@ -242,14 +283,14 @@ flowchart TD
 
 ### テキスト要約
 
-1. `settingsN.json` から他 Pi の BT アドレスと `profile` を読み込む（`ble_blockchain.config.device_settings`）
+1. `settingsN.json` から他端末の BT アドレスと `profile` を読み込む（`ble_blockchain.config.device_settings`）
 2. `settingsN.json` の `signing_key_path` から ECDSA 秘密鍵を読み込む（`ble_blockchain.cipher`）
-3. BLE 端末をスキャンする（`ble_blockchain.ble.discover`）
+3. Transport 経由でスキャンする（`transport.scan()`。BLE 方式は `ble_blockchain.ble.discover`、file 方式は `scan_results.csv`）
 4. 事前登録 CSV と照合し不要データを除去する（`ble_blockchain.pipeline.delete_excess_data`）
 5. CSV bytes に ECDSA 署名する
 6. 同一 CSV bytes を AES-256-GCM で暗号化する
 7. JSON ペイロードにシリアライズする（`ble_blockchain.ble.message_codec`）
-8. `runtime_profiles` の `steps` に従い、discoverable / send / receive / sleep を実行する
+8. `runtime_profiles` の `steps` に従い、discoverable / send / receive / sleep を Transport 経由で実行する
 9. 受信ごとに復号・署名検証する（検証失敗はチェーン追加対象外）
 10. 検証済み受信が `min_verified_receives` 以上のとき、**ユニーク報告者数**が過半数以上かつ `content_hash` が一致した `bt_addrs` を、CSV 再突合後の gakuseki 多数決でブロックに追加し、チェーンを出力する
 
@@ -317,8 +358,8 @@ docker run --rm -v "$(pwd)/data:/data" ble-chain-validator /data/chains/canonica
 
 ### 実行環境の注意
 
-- L2CAP（PyBlueZ）は **Linux（Raspberry Pi）** 向けです。macOS では送受信部分は動作しません。
-- 全 Pi で `ble-blockchain`（または `main.py`）を起動するタイミングの同期は、主に `sleep` 秒数に依存します。
+- BLE 方式（L2CAP / PyBlueZ）は **Linux（Raspberry Pi）** 向けです。macOS では BLE 方式の送受信部分は動作しません。ファイル共有が可能な環境であれば、**file 方式**（`config/transport.json` の `mode` を `"file"` に変更）で macOS 等でも同一パイプラインが動きます。
+- 全端末で `ble-blockchain`（または `main.py`）を起動するタイミングの同期は、主に `sleep` 秒数に依存します。
 
 ## セキュリティ
 
diff --git a/config/transport.json b/config/transport.json
new file mode 100644
index 0000000..6fcbdad
--- /dev/null
+++ b/config/transport.json
@@ -0,0 +1,9 @@
+{
+  "mode": "ble",
+  "file": {
+    "inbox_dir": "data/transport/inbox",
+    "scan_csv": "data/transport/scan_results.csv",
+    "sender_id": "device1",
+    "poll_interval_sec": 0.5
+  }
+}
diff --git a/docs/guides/overview.rst b/docs/guides/overview.rst
index ca15dd7..4a6184a 100644
--- a/docs/guides/overview.rst
+++ b/docs/guides/overview.rst
@@ -5,6 +5,12 @@
 卒業研究用システムです。Raspberry Pi 複数台が BLE スキャン結果を暗号化・署名して
 L2CAP で交換し、過半数合意でブロックチェーンを構築します。
 
+ペイロード交換は Transport 抽象層（``src/ble_blockchain/transport.py``）を介して
+行います。``config/transport.json`` の ``mode`` で **BLE 方式**（既定）と **file 方式**
+を切り替えられ、file 方式では Bluetooth を使わず共有 inbox ディレクトリと
+スキャン結果 CSV で交換するため、BLE 非対応環境（macOS 等）でも同一パイプラインが
+動きます。
+
 リポジトリレイアウト
 --------------------
 
@@ -16,10 +22,12 @@ L2CAP で交換し、過半数合意でブロックチェーンを構築しま
 
    BLE_Blockchain/
    ├── main.py                 # 後方互換ラッパー
-   ├── config/                 # JSON 設定
-   ├── settings1.json …
+   ├── config/                 # JSON 設定（transport.json 等）
+   ├── settings1.json …        # 端末別設定
    ├── src/ble_blockchain/
    │   ├── app/main.py         # パイプライン
+   │   ├── transport.py        # Transport 抽象層（ble / file 切替）
+   │   ├── transport_file.py   # file 方式（共有 inbox）
    │   ├── ble/
    │   ├── blockchain/
    │   ├── cipher/
diff --git a/src/ble_blockchain/app/main.py b/src/ble_blockchain/app/main.py
index 8475399..c3c97c8 100644
--- a/src/ble_blockchain/app/main.py
+++ b/src/ble_blockchain/app/main.py
@@ -1,7 +1,6 @@
 """CLI entry point for the BLE scan, exchange, and blockchain pipeline."""
 
 import argparse
-import asyncio
 import json
 import time
 from pathlib import Path
@@ -9,9 +8,7 @@
 
 import pandas as pd
 
-from ble_blockchain.ble.discover import scan
 from ble_blockchain.ble.message_codec import MessagePayload, pack, unpack
-from ble_blockchain.ble.start_discoverable import start_discoverable
 from ble_blockchain.blockchain.aggregator import aggregate_chains
 from ble_blockchain.blockchain.export import export_chain as write_chain_export
 from ble_blockchain.blockchain.myblock import MyBlockChain, payload_content_hash
@@ -28,6 +25,7 @@
 from ble_blockchain.paths import repo_root
 from ble_blockchain.pipeline.delete_excess_data import delete_excess_data
 from ble_blockchain.pipeline.pandas_d_encode import pandas_decode, pandas_encode
+from ble_blockchain.transport import TransportService, load_transport
 from ble_blockchain.types import ReceivedPayload
 
 RUNTIME_PROFILES_PATH = repo_root() / "config/runtime_profiles.json"
@@ -44,12 +42,12 @@ def load_runtime_profile(profile_name: str) -> dict[str, Any]:
     return profiles[profile_name]
 
 
-def build_send_payload(settings: DeviceSettings) -> bytes:
+def build_send_payload(settings: DeviceSettings, transport: TransportService) -> bytes:
     """Scan, filter, sign, and encrypt a payload ready for BLE send."""
     secret_key = load_signing_key_from_pem(settings.signing_key_path)
     public_key = secret_key.verifying_key
 
-    bt_addrs, device_name = asyncio.run(scan())
+    bt_addrs, device_name = transport.scan()
     df = pd.DataFrame(
         list(zip(bt_addrs, device_name)), columns=["bt_addrs", "device_name"]
     )
@@ -71,7 +69,7 @@ def build_send_payload(settings: DeviceSettings) -> bytes:
 def run_communication_steps(
     profile: dict[str, Any],
     settings: DeviceSettings,
-    tanmatsu_bt_addrs: list[str],
+    transport: TransportService,
     receive_data_list: list[ReceivedPayload],
 ) -> None:
     """Execute discoverable, send, receive, and sleep steps from a profile."""
@@ -85,22 +83,14 @@ def run_communication_steps(
         action = step["action"]
 
         if action == "discoverable":
-            start_discoverable()
+            transport.start_discoverable()
         elif action == "send":
-            # PyBluez is Linux-only; defer import so macOS dev/tests can load main.
-            from ble_blockchain.pipeline.send_and_receive import (  # pylint: disable=import-outside-toplevel
-                SEND,
-            )
-
-            payload_bytes = build_send_payload(settings)
-            SEND(tanmatsu_bt_addrs, payload_bytes)
+            payload_bytes = build_send_payload(settings, transport)
+            transport.send_payload(settings.tanmatsu_bt_addrs, payload_bytes)
         elif action == "receive":
-            from ble_blockchain.ble.l2cap_server import (  # pylint: disable=import-outside-toplevel
-                l2cap_server,
-            )
-
+            raw = transport.receive_payload()
             receive_data_list.append(
-                process_received_payload(l2cap_server(), settings.trusted_peer_pems)
+                process_received_payload(raw, settings.trusted_peer_pems)
             )
         elif action == "sleep":
             time.sleep(step.get("seconds", defaults["sleep_seconds"]))
@@ -174,10 +164,11 @@ def run_pipeline(  # pylint: disable=too-many-locals
 
     print(settings.tanmatsu_bt_addrs)
 
+    transport = load_transport(settings)
     receive_data_list: list[ReceivedPayload] = []
 
     run_communication_steps(
-        profile, settings, settings.tanmatsu_bt_addrs, receive_data_list
+        profile, settings, transport, receive_data_list
     )
 
     chain = MyBlockChain()
diff --git a/src/ble_blockchain/ble/start_discoverable.py b/src/ble_blockchain/ble/start_discoverable.py
index 59db0fd..ce354fe 100644
--- a/src/ble_blockchain/ble/start_discoverable.py
+++ b/src/ble_blockchain/ble/start_discoverable.py
@@ -1,7 +1,5 @@
 """Raspberry Pi BLE を discoverable にするヘルパー。"""
 
-import subprocess
-
 
 def start_discoverable() -> None:
     """
@@ -14,6 +12,6 @@ def start_discoverable() -> None:
 
 
     """
-    subprocess.run(
-        ["sudo", "bluetoothctl", "discoverable", "on"], check=False
-    )
+    import subprocess  # pylint: disable=import-outside-toplevel
+
+    subprocess.run(["sudo", "bluetoothctl", "discoverable", "on"], check=False)
diff --git a/src/ble_blockchain/config/device_settings.py b/src/ble_blockchain/config/device_settings.py
index 9d398be..60cc89b 100644
--- a/src/ble_blockchain/config/device_settings.py
+++ b/src/ble_blockchain/config/device_settings.py
@@ -16,6 +16,7 @@ class DeviceSettings:
     signing_key_path: str
     public_key_pem: str
     trusted_peer_pems: frozenset[str]
+    settings_path: Path
 
 
 def load_device_settings(settings_path: Union[str, Path]) -> DeviceSettings:
@@ -50,4 +51,5 @@ def load_device_settings(settings_path: Union[str, Path]) -> DeviceSettings:
         signing_key_path=signing_key_path,
         public_key_pem=public_key_pem,
         trusted_peer_pems=trusted_peer_pems,
+        settings_path=path.resolve(),
     )
diff --git a/src/ble_blockchain/config/loader.py b/src/ble_blockchain/config/loader.py
index 4abcd3b..6a7743c 100644
--- a/src/ble_blockchain/config/loader.py
+++ b/src/ble_blockchain/config/loader.py
@@ -54,6 +54,16 @@ class DataSchema:
     output_field: str = "bt_addrs"
 
 
+@dataclass(frozen=True)
+class TransportConfig:
+    """Transport settings (BLE or file) for payload exchange."""
+    mode: str
+    file_inbox_dir: str
+    file_scan_csv: str
+    file_sender_id: str
+    file_poll_interval_sec: float
+
+
 @dataclass(frozen=True)
 class BlockchainConfig:
     """Blockchain majority and export rules."""
@@ -94,6 +104,21 @@ def load_paths_config() -> PathsConfig:
     )
 
 
+def load_transport_config() -> TransportConfig:
+    """Load transport settings from config/transport.json."""
+    data = load_json_config("transport.json")
+    file_block = data.get("file", {})
+    return TransportConfig(
+        mode=str(data["mode"]),
+        file_inbox_dir=str(file_block.get("inbox_dir", "data/transport/inbox")),
+        file_scan_csv=str(file_block.get("scan_csv", "data/transport/scan_results.csv")),
+        file_sender_id=str(file_block.get("sender_id", "device1")),
+        file_poll_interval_sec=float(
+            file_block.get("poll_interval_sec", 0.5)
+        ),
+    )
+
+
 def load_data_schema(raw: dict[str, Any]) -> DataSchema:
     """Build a DataSchema from a partial data_schema config block."""
     rename = raw.get("csv_identity_rename", {"学籍番号": "gakuseki"})
diff --git a/src/ble_blockchain/transport.py b/src/ble_blockchain/transport.py
new file mode 100644
index 0000000..f7e70bd
--- /dev/null
+++ b/src/ble_blockchain/transport.py
@@ -0,0 +1,86 @@
+"""Transport abstraction for payload exchange (BLE or file based)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Protocol
+
+from ble_blockchain.config.device_settings import DeviceSettings
+from ble_blockchain.config.loader import load_transport_config
+from ble_blockchain.transport_file import FileTransportService
+
+
+class TransportService(Protocol):
+    """Common interface for exchanging payloads with peers."""
+
+    # pylint: disable=missing-function-docstring
+    def scan(self) -> tuple[list[str], list[str]]:
+        ...
+
+    def start_discoverable(self) -> None:
+        ...
+
+    def send_payload(self, peers: list[str], payload_bytes: bytes) -> None:
+        ...
+
+    def receive_payload(self) -> bytes:
+        ...
+
+    # pylint: enable=missing-function-docstring
+
+
+class BleTransportService:
+    """Transport over BLE. 依存ライブラリは実行時 import で解決する。"""
+
+    def scan(self) -> tuple[list[str], list[str]]:
+        """Scan BLE devices via the discover module."""
+        # bleak をモジュール import 時に解決しないため実行時 import する
+        from ble_blockchain.ble.discover import (  # pylint: disable=import-outside-toplevel
+            scan,
+        )
+
+        import asyncio  # pylint: disable=import-outside-toplevel
+
+        return asyncio.run(scan())
+
+    def start_discoverable(self) -> None:
+        """Make the device discoverable via bluetoothctl."""
+        from ble_blockchain.ble.start_discoverable import (  # pylint: disable=import-outside-toplevel
+            start_discoverable,
+        )
+
+        start_discoverable()
+
+    def send_payload(self, peers: list[str], payload_bytes: bytes) -> None:
+        """Send the packed payload to each peer over L2CAP."""
+        from ble_blockchain.pipeline.send_and_receive import (  # pylint: disable=import-outside-toplevel
+            SEND,
+        )
+
+        SEND(peers, payload_bytes)
+
+    def receive_payload(self) -> bytes:
+        """Receive one packed payload over L2CAP."""
+        from ble_blockchain.ble.l2cap_server import (  # pylint: disable=import-outside-toplevel
+            l2cap_server,
+        )
+
+        return l2cap_server()
+
+
+def load_transport(settings: DeviceSettings) -> TransportService:
+    """Build a transport service from config/transport.json mode."""
+    config = load_transport_config()
+    if config.mode == "ble":
+        return BleTransportService()
+    if config.mode == "file":
+        sender_id = config.file_sender_id
+        if not sender_id:
+            sender_id = settings.settings_path.stem
+        return FileTransportService(
+            inbox_dir=Path(config.file_inbox_dir),
+            scan_csv=Path(config.file_scan_csv),
+            sender_id=sender_id,
+            poll_interval_sec=config.file_poll_interval_sec,
+        )
+    raise ValueError(f"Unknown transport mode: {config.mode}")
diff --git a/src/ble_blockchain/transport_file.py b/src/ble_blockchain/transport_file.py
new file mode 100644
index 0000000..0899bc2
--- /dev/null
+++ b/src/ble_blockchain/transport_file.py
@@ -0,0 +1,61 @@
+"""File-based transport that exchanges payloads via local directories."""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pandas as pd
+
+
+class FileTransportService:
+    """Exchange payloads by writing/reading files under a shared inbox dir."""
+
+    def __init__(
+        self,
+        *,
+        inbox_dir: Path,
+        scan_csv: Path,
+        sender_id: str,
+        poll_interval_sec: float = 0.5,
+    ) -> None:
+        self.inbox_dir = inbox_dir
+        self.scan_csv = scan_csv
+        self.sender_id = sender_id
+        self.poll_interval_sec = poll_interval_sec
+
+    def scan(self) -> tuple[list[str], list[str]]:
+        """Read peer addresses and names from the configured scan CSV."""
+        if not self.scan_csv.exists():
+            return [], []
+
+        df = pd.read_csv(self.scan_csv)
+        return list(df["bt_addrs"]), list(df["device_name"])
+
+    def start_discoverable(self) -> None:
+        """No-op for file transport."""
+        return None
+
+    def send_payload(self, peers: list[str], payload_bytes: bytes) -> None:
+        """Write payloads to the shared inbox dir, prefixed by recipient IDs."""
+        self.inbox_dir.mkdir(parents=True, exist_ok=True)
+        timestamp = int(time.time() * 1000)
+        for peer in peers:
+            safe_peer = peer.replace(":", "_")
+            filename = f"{safe_peer}__{self.sender_id}-{timestamp}.bin"
+            (self.inbox_dir / filename).write_bytes(payload_bytes)
+
+    def receive_payload(self) -> bytes:
+        """Read the oldest inbox file addressed to this sender, or poll."""
+        self.inbox_dir.mkdir(parents=True, exist_ok=True)
+        while True:
+            candidates = sorted(
+                self.inbox_dir.glob(f"{self.sender_id}*.bin"),
+                key=lambda path: path.stat().st_mtime,
+            )
+            if candidates:
+                path = candidates[0]
+                payload = path.read_bytes()
+                path.unlink()
+                return payload
+            time.sleep(self.poll_interval_sec)
diff --git a/tests/unit/test_main_transport.py b/tests/unit/test_main_transport.py
new file mode 100644
index 0000000..ba5bcab
--- /dev/null
+++ b/tests/unit/test_main_transport.py
@@ -0,0 +1,213 @@
+"""Unit tests for main.py behaviors reached through the transport layer."""
+
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+import pytest
+
+from ble_blockchain.app.main import (
+    build_send_payload,
+    process_received_payload,
+    run_communication_steps,
+)
+from ble_blockchain.cipher.cipher import make_key, public_key_to_pem
+from ble_blockchain.config.device_settings import DeviceSettings
+from ble_blockchain.transport_file import FileTransportService
+
+
+def _make_settings(*, signing_key_path: str, trusted_pems: frozenset[str]) -> DeviceSettings:
+    """Build a DeviceSettings with the given key paths (test helper)."""
+    return DeviceSettings(
+        profile="device1",
+        tanmatsu_bt_addrs=["AA:BB:CC:DD:EE:FF"],
+        signing_key_path=signing_key_path,
+        public_key_pem="pem-self",
+        trusted_peer_pems=trusted_pems,
+        settings_path=Path("settings1.json"),
+    )
+
+
+def _write_signing_key(path: Path) -> str:
+    """Write an ECDSA signing key in PEM and return its peer public key PEM (test helper)."""
+    secret_key, public_key = make_key()
+    pem = public_key_to_pem(public_key)
+    path.write_bytes(secret_key.to_pem())
+    return pem
+
+
+def _read_first_registered_bt_addr() -> str:
+    """Read the first bt_addrs from the registered preliminary CSV (test helper)."""
+    df = pd.read_csv("data_folder/事前取得データ.csv")
+    return str(df["bt_addrs"].iloc[0])
+
+
+class FakeScanTransport:  # pylint: disable=too-few-public-methods
+    """TransportService 実装: scan が固定タプルを返し、呼び出しを記録するフェイク。"""
+
+    def __init__(self, bt_addrs: list[str], device_names: list[str]) -> None:
+        self._bt_addrs = bt_addrs
+        self._device_names = device_names
+        self.send_calls: list[tuple[list[str], bytes]] = []
+        self.discoverable_calls = 0
+        self.receive_calls = 0
+
+    def scan(self) -> tuple[list[str], list[str]]:
+        """Fixed addresses and device names (recorded via __init__)."""
+        return self._bt_addrs, self._device_names
+
+    def start_discoverable(self) -> None:
+        """Record one discoverable call (no-op)."""
+        self.discoverable_calls += 1
+
+    def send_payload(self, peers: list[str], payload_bytes: bytes) -> None:
+        """Record one send call (no-op)."""
+        self.send_calls.append((peers, payload_bytes))
+
+    def receive_payload(self) -> bytes:
+        """Record one receive call and return an empty payload."""
+        self.receive_calls += 1
+        return b""
+
+
+class TestBuildSendPayload:  # pylint: disable=too-few-public-methods
+    """build_send_payload(): scan → 署名・暗号化・pack のテスト。"""
+
+    def test_returns_verifiable_payload(self, tmp_path: Path) -> None:
+        """正常系: pack された bytes が返り、受信側 process_received_payload が検証できる。"""
+        # Given: 事前登録 CSV に載っている bt_addrs と署名鍵
+        registered_addr = _read_first_registered_bt_addr()
+        signing_key_path = tmp_path / "signing.pem"
+        key_pem = _write_signing_key(signing_key_path)
+        settings = _make_settings(
+            signing_key_path=str(signing_key_path),
+            trusted_pems=frozenset({key_pem}),
+        )
+        fake = FakeScanTransport([registered_addr], ["phone"])
+
+        # When: build_send_payload() を呼ぶ
+        payload = build_send_payload(settings, fake)
+
+        # Then: pack された bytes が返り、受信検証で verified=True になる
+        assert isinstance(payload, bytes)
+        assert len(payload) > 0
+        result = process_received_payload(payload, settings.trusted_peer_pems)
+        assert result.verified is True
+        assert result.df is not None
+        assert registered_addr in result.df["bt_addrs"].values
+
+
+class TestRunCommunicationSteps:
+    """run_communication_steps(): 各 action の実行と委譲のテスト。"""
+
+    def test_receive_appends_verified_payload(self, tmp_path: Path) -> None:
+        """正常系: receive で検証可能な payload が 1 件追加される。"""
+        # Given: 送信側 payload を検証可能な file transport と受信側 settings
+        sender_pem = _write_signing_key(tmp_path / "sender.pem")
+        receiver_settings = _make_settings(
+            signing_key_path=str(tmp_path / "sender.pem"),
+            trusted_pems=frozenset({sender_pem}),
+        )
+        inbox = tmp_path / "inbox"
+        inbox.mkdir(parents=True)
+        scan_csv = tmp_path / "scan.csv"
+        scan_csv.write_text("bt_addrs,device_name\n", encoding="utf-8")
+        transport = FileTransportService(
+            inbox_dir=inbox,
+            scan_csv=scan_csv,
+            sender_id="device1",
+            poll_interval_sec=0.0,
+        )
+
+        # When: 検証可能な payload を device1 宛に入れて receive を実行する
+        payload_bytes = build_send_payload(receiver_settings, FakeScanTransport([], []))
+        (inbox / "device1__peer-1.bin").write_bytes(payload_bytes)
+        receive_data_list: list[Any] = []
+        run_communication_steps(
+            {"steps": [{"action": "receive"}]},
+            receiver_settings,
+            transport,
+            receive_data_list,
+        )
+
+        # Then: 1 件検証済みで追加される
+        assert len(receive_data_list) == 1
+        assert receive_data_list[0].verified is True
+        assert receive_data_list[0].df is not None
+
+    def test_sleep_action_calls_time_sleep(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """正常系: sleep action は time.sleep を指定秒数で呼ぶ。"""
+        # Given: settings と fake transport
+        _write_signing_key(tmp_path / "signing.pem")
+        settings = _make_settings(
+            signing_key_path=str(tmp_path / "signing.pem"),
+            trusted_pems=frozenset(),
+        )
+        fake = FakeScanTransport([], [])
+        sleep_log: list[float] = []
+
+        def record_sleep(seconds: float) -> None:
+            sleep_log.append(seconds)
+
+        monkeypatch.setattr("ble_blockchain.app.main.time.sleep", record_sleep)
+
+        # When: sleep を含む steps を実行する
+        run_communication_steps(
+            {"steps": [{"action": "sleep", "seconds": 2.5}]},
+            settings,
+            fake,
+            [],
+        )
+
+        # Then: time.sleep が指定秒数で呼ばれる
+        assert sleep_log == [2.5]
+
+    def test_discoverable_and_send_delegate_to_transport(self, tmp_path: Path) -> None:
+        """正常系: discoverable / send が transport メソッドへ委譲される。"""
+        # Given: 事前登録 bt_addr を返す fake transport と署名鍵
+        registered_addr = _read_first_registered_bt_addr()
+        signing_key_path = tmp_path / "signing.pem"
+        _write_signing_key(signing_key_path)
+        settings = _make_settings(
+            signing_key_path=str(signing_key_path),
+            trusted_pems=frozenset(),
+        )
+        fake = FakeScanTransport([registered_addr], ["phone"])
+
+        # When: discoverable / send を含む steps を実行する
+        run_communication_steps(
+            {
+                "steps": [
+                    {"action": "discoverable"},
+                    {"action": "send"},
+                ]
+            },
+            settings,
+            fake,
+            [],
+        )
+
+        # Then: start_discoverable が 1 回、send が settings の bt_addrs で呼ばれる
+        assert fake.discoverable_calls == 1
+        assert len(fake.send_calls) == 1
+        assert fake.send_calls[0][0] == settings.tanmatsu_bt_addrs
+        assert isinstance(fake.send_calls[0][1], bytes)
+
+    def test_unknown_action_raises_value_error(self, tmp_path: Path) -> None:
+        """異常系: 未知の action では ValueError が投げられる。"""
+        # Given: settings と fake transport
+        signing_key_path = tmp_path / "signing.pem"
+        _write_signing_key(signing_key_path)
+        settings = _make_settings(
+            signing_key_path=str(signing_key_path),
+            trusted_pems=frozenset(),
+        )
+        fake = FakeScanTransport([], [])
+
+        # When/Then: 未知 action で ValueError が投げられる
+        with pytest.raises(ValueError, match="Unknown step action:"):
+            run_communication_steps(
+                {"steps": [{"action": "broadcast"}]}, settings, fake, []
+            )
diff --git a/tests/unit/test_transport.py b/tests/unit/test_transport.py
new file mode 100644
index 0000000..8d4c0ec
--- /dev/null
+++ b/tests/unit/test_transport.py
@@ -0,0 +1,174 @@
+"""Unit tests for transport selection and BLE transport delegation."""
+
+from pathlib import Path
+from typing import Optional
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from ble_blockchain.config.device_settings import DeviceSettings
+from ble_blockchain.config.loader import TransportConfig
+from ble_blockchain.transport import BleTransportService, FileTransportService, load_transport
+
+
+def _make_settings(*, settings_path: Optional[Path] = None) -> DeviceSettings:
+    """Build a minimal DeviceSettings for transport selection (test helper)."""
+    return DeviceSettings(
+        profile="device1",
+        tanmatsu_bt_addrs=["AA:BB:CC:DD:EE:FF"],
+        signing_key_path="keys/device1_private.pem",
+        public_key_pem="pem-self",
+        trusted_peer_pems=frozenset({"pem-2"}),
+        settings_path=settings_path or Path("settings1.json"),
+    )
+
+
+class TestLoadTransport:
+    """load_transport(): 設定 mode と設定値のテスト。"""
+
+    def test_ble_mode_returns_ble_transport(self) -> None:
+        """正常系: config/transport.json の mode="ble" で BleTransportService が返る。"""
+        # Given: 本物の config/transport.json（mode="ble"）を読む transport
+        settings = _make_settings()
+
+        # When: load_transport() を呼ぶ
+        transport = load_transport(settings)
+
+        # Then: BleTransportService が返る
+        assert isinstance(transport, BleTransportService)
+
+    def test_file_mode_returns_file_transport(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """正常系: mode="file" で FileTransportService が返る。"""
+        # Given: mode="file" の設定を返すスタブへ差し替え
+        monkeypatch.setattr(
+            "ble_blockchain.transport.load_transport_config",
+            lambda: TransportConfig(
+                mode="file",
+                file_inbox_dir="data/inbox",
+                file_scan_csv="data/scan.csv",
+                file_sender_id="device9",
+                file_poll_interval_sec=0.25,
+            ),
+        )
+
+        # When: load_transport() を呼ぶ
+        transport = load_transport(_make_settings())
+
+        # Then: FileTransportService が設定値で構築される
+        assert isinstance(transport, FileTransportService)
+        assert transport.sender_id == "device9"
+        assert transport.poll_interval_sec == 0.25
+        assert Path("data/inbox") == transport.inbox_dir
+        assert Path("data/scan.csv") == transport.scan_csv
+
+    def test_file_mode_empty_sender_id_falls_back_to_settings_stem(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path,
+    ) -> None:
+        """正常系(境界値): sender_id 空文字なら settings_path.stem が使われる。"""
+        # Given: sender_id が空の file 設定と settings_path を持つ settings
+        settings_path = tmp_path / "settings5.json"
+        settings_path.touch()
+        settings = _make_settings(settings_path=settings_path)
+        monkeypatch.setattr(
+            "ble_blockchain.transport.load_transport_config",
+            lambda: TransportConfig(
+                mode="file",
+                file_inbox_dir="data/inbox",
+                file_scan_csv="data/scan.csv",
+                file_sender_id="",
+                file_poll_interval_sec=0.5,
+            ),
+        )
+
+        # When: load_transport() を呼ぶ
+        transport = load_transport(settings)
+
+        # Then: sender_id に settings5 が使われる
+        assert isinstance(transport, FileTransportService)
+        assert transport.sender_id == "settings5"
+
+    def test_unknown_mode_raises_value_error(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """異常系: 未知の mode では ValueError が投げられる。"""
+        # Given: mode="udp" の設定を返すスタブ
+        monkeypatch.setattr(
+            "ble_blockchain.transport.load_transport_config",
+            lambda: TransportConfig(
+                mode="udp",
+                file_inbox_dir="data/inbox",
+                file_scan_csv="data/scan.csv",
+                file_sender_id="device1",
+                file_poll_interval_sec=0.5,
+            ),
+        )
+
+        # When/Then: メッセージに mode 名が含まれる ValueError が投げられる
+        with pytest.raises(ValueError, match="udp"):
+            load_transport(_make_settings())
+
+
+class TestBleTransportService:
+    """BleTransportService: 各 ble/ 配下関数への委譲のテスト。"""
+
+    def test_scan_delegates_to_discover_scan(self) -> None:
+        """正常系: scan() は discover.scan の結果をそのまま返す。"""
+        # Given: discover.scan がタプルを返す AsyncMock
+        expected = (["AA:BB:CC:DD:EE:FF"], ["phone"])
+        service = BleTransportService()
+
+        # When: scan() を呼ぶ（内部で asyncio.run が使われるため二重実行しない）
+        with patch(
+            "ble_blockchain.ble.discover.scan",
+            new=AsyncMock(return_value=expected),
+        ):
+            result = service.scan()
+
+        # Then: タプルがそのまま返る
+        assert result == expected
+
+    def test_send_payload_delegates_to_send(
+        self,
+    ) -> None:
+        """正常系: send_payload() は SEND(peers, payload) を呼ぶ。"""
+        # Given: SEND を Mock へ差し替え
+        with patch("ble_blockchain.pipeline.send_and_receive.SEND") as mock_send:
+            service = BleTransportService()
+
+            # When: send_payload() を呼ぶ
+            service.send_payload(["AA:BB:CC:DD:EE:FF"], b"bytes")
+
+        # Then: SEND が peers と payload で 1 回呼ばれる
+        mock_send.assert_called_once_with(["AA:BB:CC:DD:EE:FF"], b"bytes")
+
+    def test_receive_payload_delegates_to_l2cap_server(self) -> None:
+        """正常系: receive_payload() は l2cap_server() の戻り値を返す。"""
+        # Given: l2cap_server が bytes を返す Mock
+        with patch(
+            "ble_blockchain.ble.l2cap_server.l2cap_server",
+            return_value=b"received",
+        ) as mock_server:
+            service = BleTransportService()
+
+            # When: receive_payload() を呼ぶ
+            result = service.receive_payload()
+
+        # Then: l2cap_server の戻り値がそのまま返る
+        assert result == b"received"
+        mock_server.assert_called_once_with()
+
+    def test_start_discoverable_delegates_to_start_discoverable(self) -> None:
+        """正常系: start_discoverable() は ble の関数へ委譲する。"""
+        # Given: start_discoverable を Mock へ差し替え
+        with patch(
+            "ble_blockchain.ble.start_discoverable.start_discoverable",
+        ) as mock_start:
+            service = BleTransportService()
+
+            # When: start_discoverable() を呼ぶ
+            service.start_discoverable()
+
+        # Then: 委譲先が 1 回呼ばれる
+        mock_start.assert_called_once_with()
diff --git a/tests/unit/test_transport_file.py b/tests/unit/test_transport_file.py
new file mode 100644
index 0000000..9185fd8
--- /dev/null
+++ b/tests/unit/test_transport_file.py
@@ -0,0 +1,262 @@
+"""Unit tests for FileTransportService (shared-inbox file transport)."""
+
+import os
+import time
+from pathlib import Path
+
+import pytest
+
+from ble_blockchain.transport_file import FileTransportService
+
+class _PollingObserved(Exception):
+    """receive_payload がポーリング（sleep）に達したことを示す番兵例外。"""
+
+
+def _make_service(
+    inbox_dir: Path,
+    scan_csv: Path,
+    *,
+    sender_id: str = "device1",
+    poll_interval_sec: float = 0.01,
+) -> FileTransportService:
+    """Build a FileTransportService with the given paths (test helper)."""
+    return FileTransportService(
+        inbox_dir=inbox_dir,
+        scan_csv=scan_csv,
+        sender_id=sender_id,
+        poll_interval_sec=poll_interval_sec,
+    )
+
+
+def _write_scan_csv(path: Path, *, header: str = "bt_addrs,device_name") -> None:
+    """Write a scan CSV with the given header and rows (test helper)."""
+    path.write_text(
+        f"{header}\n"
+        "FC:66:CF:BE:10:BF,phone\n"
+        "AA:BB:CC:DD:EE:FF,laptop\n",
+        encoding="utf-8",
+    )
+
+
+def _assert_polls_then_returns(svc: FileTransportService) -> None:
+    """sleep を番兵化して receive_payload のポーリング発生を安全に検証する。"""
+    real_sleep = time.sleep
+    try:
+        time.sleep = (  # type: ignore[assignment]
+            lambda _seconds: (_ for _ in ()).throw(_PollingObserved())
+        )
+        svc.receive_payload()
+        pytest.fail("sleep (polling) should have been observed")
+    except _PollingObserved:
+        return  # ポーリングが観測された（以降のアサーションは呼び出し側で行う）
+    finally:
+        time.sleep = real_sleep  # type: ignore[assignment]
+
+
+class TestScan:
+    """scan(): CSV 読み取りのテスト。"""
+
+    def test_scan_returns_addresses_and_names(self, tmp_path: Path) -> None:
+        """正常系: bt_addrs / device_name の2列 CSV からタプルが返る。"""
+        # Given: 有効な scan CSV を指定した transport
+        csv_path = tmp_path / "scan.csv"
+        _write_scan_csv(csv_path)
+        svc = _make_service(tmp_path / "inbox", csv_path)
+
+        # When: scan() を呼ぶ
+        bt_addrs, device_name = svc.scan()
+
+        # Then: アドレスと端末名のタプルが返る
+        assert bt_addrs == ["FC:66:CF:BE:10:BF", "AA:BB:CC:DD:EE:FF"]
+        assert device_name == ["phone", "laptop"]
+
+    def test_scan_missing_csv_returns_empty(self, tmp_path: Path) -> None:
+        """正常系(境界値): CSV が存在しないと空タプルが返る。"""
+        # Given: 存在しない scan CSV を指定した transport
+        svc = _make_service(tmp_path / "inbox", tmp_path / "nonexistent.csv")
+
+        # When: scan() を呼ぶ
+        bt_addrs, device_name = svc.scan()
+
+        # Then: 空タプル（[]、[]）が返る
+        assert not bt_addrs
+        assert not device_name
+
+    def test_scan_missing_bt_addrs_column_raises_key_error(
+        self, tmp_path: Path,
+    ) -> None:
+        """異常系: bt_addrs 列が無い CSV では KeyError が発生する。"""
+        # Given: bt_addrs 列が無い CSV を指定した transport
+        csv_path = tmp_path / "bad.csv"
+        csv_path.write_text("device_name\nphone\n", encoding="utf-8")
+        svc = _make_service(tmp_path / "inbox", csv_path)
+
+        # When/Then: scan() が KeyError を投げる
+        with pytest.raises(KeyError):
+            svc.scan()
+
+
+class TestSendPayload:
+    """send_payload(): inbox への書き込みのテスト。"""
+
+    def test_send_payload_no_peers_creates_no_files(self, tmp_path: Path) -> None:
+        """正常系(境界値): peers が 0 件ならファイルは生成されない。"""
+        # Given: peers を渡さない transport
+        inbox = tmp_path / "inbox"
+        svc = _make_service(inbox, tmp_path / "scan.csv")
+
+        # When: peers = [] で send_payload() を呼ぶ
+        svc.send_payload([], b"payload")
+
+        # Then: inbox ディレクトリは作成されるがファイルは無い
+        assert inbox.is_dir()
+        assert not list(inbox.iterdir())
+
+    def test_send_payload_single_peer(self, tmp_path: Path) -> None:
+        """正常系: peers 1 件で受信者プレフィックスのファイルが作られる。"""
+        # Given: sender_id="device1" の transport
+        inbox = tmp_path / "inbox"
+        svc = _make_service(inbox, tmp_path / "scan.csv", sender_id="device1")
+
+        # When: 1 件の peer へ送信する
+        svc.send_payload(["device2"], b"hello")
+
+        # Then: device2 プレフィックスのファイルが 1 つ生成され内容が一致する
+        files = list(inbox.glob("*.bin"))
+        assert len(files) == 1
+        assert files[0].name.startswith("device2__device1-")
+        assert files[0].read_bytes() == b"hello"
+
+    def test_send_payload_multiple_peers(self, tmp_path: Path) -> None:
+        """正常系: peers 複数件で受信者ごとにファイルが生成される。"""
+        # Given: sender_id="device1" の transport
+        inbox = tmp_path / "inbox"
+        svc = _make_service(inbox, tmp_path / "scan.csv", sender_id="device1")
+
+        # When: 2 件の peer（うち1件は ':' を含むアドレス）へ送信する
+        svc.send_payload(["device2", "AA:BB:CC:DD:EE:FF"], b"data")
+
+        # Then: 受信者プレフィックス付きファイルが 2 つ生成される
+        files = sorted(path.name for path in inbox.glob("*.bin"))
+        assert len(files) == 2
+        assert any(name.startswith("device2__device1-") for name in files)
+        # アドレスの ':' は '_' に置換される
+        assert any(name.startswith("AA_BB_CC_DD_EE_FF__device1-") for name in files)
+        assert all((inbox / name).read_bytes() == b"data" for name in files)
+
+
+class TestReceivePayload:
+    """receive_payload(): inbox からの読み取り・削除・ポーリングのテスト。"""
+
+    def test_receive_returns_bytes_and_deletes_file(self, tmp_path: Path) -> None:
+        """正常系: 自分宛ファイルがあれば bytes が返りファイルが削除される。"""
+        # Given: device2 宛のファイルが inbox に 1 件ある
+        inbox = tmp_path / "inbox"
+        inbox.mkdir(parents=True)
+        (inbox / "device2__device1-100.bin").write_bytes(b"hello")
+        svc = _make_service(inbox, tmp_path / "scan.csv", sender_id="device2")
+
+        # When: receive_payload() を呼ぶ
+        result = svc.receive_payload()
+
+        # Then: 内容が返り、ファイルは削除されている
+        assert result == b"hello"
+        assert not list(inbox.glob("*.bin"))
+
+    def test_receive_ignores_other_senders(self, tmp_path: Path) -> None:
+        """異常系: 自分宛でないファイルは返さず、残される。"""
+        # Given: device1 / device3 宛のファイルのみがある
+        inbox = tmp_path / "inbox"
+        inbox.mkdir(parents=True)
+        other_a = inbox / "device1__x-1.bin"
+        other_b = inbox / "device3__x-2.bin"
+        other_a.write_bytes(b"for-device1")
+        other_b.write_bytes(b"for-device3")
+        svc = _make_service(inbox, tmp_path / "scan.csv", sender_id="device2")
+
+        # When: sleep を番兵化して receive_payload() を回す（無限ループ回避）
+        _assert_polls_then_returns(svc)
+
+        # Then: 自分宛でないファイルは残っている
+        assert other_a.exists()
+        assert other_b.exists()
+
+    def test_receive_polls_until_file_arrives(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """正常系(外部依存): 空の間はポーリングし、ファイル到着後に返す。"""
+        # Given: 最初は空の inbox と、1 回目の sleep でファイルを注入する sleep
+        inbox = tmp_path / "inbox"
+        svc = _make_service(inbox, tmp_path / "scan.csv", sender_id="device1")
+        sleep_calls: list[float] = []
+
+        def fake_sleep(seconds: float) -> None:
+            sleep_calls.append(seconds)
+            (inbox / "device1__peer-1.bin").write_bytes(b"late")
+
+        monkeypatch.setattr("ble_blockchain.transport_file.time.sleep", fake_sleep)
+
+        # When: receive_payload() を呼ぶ
+        result = svc.receive_payload()
+
+        # Then: ポーリング後に内容が返り、sleep は 1 回だけ呼ばれる
+        assert result == b"late"
+        assert len(sleep_calls) == 1
+
+    def test_receive_returns_oldest_by_mtime(self, tmp_path: Path) -> None:
+        """正常系: 2 件ある場合は最古の mtime のものが返される。"""
+        # Given: device1 宛ファイルが 2 件（mtime が異なる）
+        inbox = tmp_path / "inbox"
+        inbox.mkdir(parents=True)
+        recent = inbox / "device1__a-2.bin"
+        oldest = inbox / "device1__a-1.bin"
+        recent.write_bytes(b"recent")
+        oldest.write_bytes(b"oldest")
+        os.utime(recent, (2000, 2000))
+        os.utime(oldest, (1000, 1000))
+        svc = _make_service(inbox, tmp_path / "scan.csv", sender_id="device1")
+
+        # When: receive_payload() を呼ぶ
+        result = svc.receive_payload()
+
+        # Then: 最古の内容が返り、新しい方は残る
+        assert result == b"oldest"
+        assert not oldest.exists()
+        assert recent.exists()
+
+
+class TestRoundtrip:  # pylint: disable=too-few-public-methods
+    """端末間ラウンドトリップのテスト。"""
+
+    def test_roundtrip_device_a_to_device_b(self, tmp_path: Path) -> None:
+        """正常系: device1 の送信を device2 が受信できる。"""
+        # Given: 共有 inbox と scan CSV を共有する 2 台の transport
+        inbox = tmp_path / "inbox"
+        csv_path = tmp_path / "scan.csv"
+        _write_scan_csv(csv_path)
+        svc_a = _make_service(inbox, csv_path, sender_id="device1")
+        svc_b = _make_service(inbox, csv_path, sender_id="device2")
+
+        # When: device1 が device2 宛に送信し、device2 が受信する
+        svc_a.send_payload(["device2"], b"hello")
+        result = svc_b.receive_payload()
+
+        # Then: ペイロードが一致し、scan も共有 CSV を返す
+        assert result == b"hello"
+        assert svc_b.scan() == (
+            ["FC:66:CF:BE:10:BF", "AA:BB:CC:DD:EE:FF"],
+            ["phone", "laptop"],
+        )
+
+
+class TestStartDiscoverable:  # pylint: disable=too-few-public-methods
+    """start_discoverable(): no-op のテスト。"""
+
+    def test_start_discoverable_returns_none(self, tmp_path: Path) -> None:
+        """正常系: no-op であり None が返る。"""
+        # Given: file transport
+        svc = _make_service(tmp_path / "inbox", tmp_path / "scan.csv")
+
+        # When/Then: start_discoverable() を呼んでも副作用がない
+        svc.start_discoverable()
+        assert not (tmp_path / "inbox").exists()