> ## Documentation Index
> Fetch the complete documentation index at: https://docs.peaq.xyz/llms.txt
> Use this file to discover all available pages before exploring further.

# Stream Edge Agent

> The peaq_ros2_stream node: sign, encrypt, and chunk machine data on the robot and serve it to buyers.

The **Stream Edge Agent** is the on-machine half of [Stream](https://docs.peaq.xyz/peaqos/functions/stream.md). It is a ROS 2 node (`stream_agent_node`, package `peaq_ros2_stream`) that subscribes to the topics you allow, applies Data Event Map (The policy file a machine owner writes to control what streams out: which topics to read, which fields to keep, drop, or encrypt, and where the signed data goes.) field rules, then signs, encrypts, and chunks (A bounded, individually encrypted slice of a data stream (by time window or size). The unit a buyer actually purchases and decrypts.) each message, stores the ciphertext, posts a signed manifest (A record describing a chunk: its hashes, storage location, and encryption details, without the data itself. The machine signs the chunk's encrypted data hash; buyers check the signature and hashes before trusting or buying.) to the Stream backend, and serves purchased chunks to buyers.

It needs no other node: it reads the machine's identity from its own `identity_ref` setting. For the trust model behind the chunks, see [Data streams](https://docs.peaq.xyz/peaqos/concepts/data-streams.md).

The agent's optional on-chain anchoring (`peaqos_event`) goes through a ROS 2 node that cannot serve Economics 2.0 machines, so leave `peaqos_event.enabled: false`.

<details><summary>Key terms in this guide</summary></details>

## How it works

```mermaid theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
flowchart LR
  Topics["ROS 2 topics"] -->|subscribe| Agent["stream_agent_node"]
  Policy["Data Event Map<br/>(YAML)"] -->|configures| Agent
  Agent -->|field rules + sign<br/>+ encrypt + chunk| Store["Storage adapter<br/>local / Walrus / S3 / Drive"]
  Agent -->|signed manifest| Backend["Stream backend"]
  Backend -->|paid order| Agent
  Agent -->|re-wrap keys + serve| Buyer["Buyer"]
```

Per inbound message: **capture → field transform → sign → encrypt → chunk → store → post manifest**. When an order is paid, the agent re-wraps that buyer's chunk keys and serves the encrypted chunks from a local delivery server.

## Install and run

The agent lives in `peaq_ros2_stream` and depends on `peaq_ros2_interfaces`. ROS 2 Humble (Docker image) or Jazzy (native host) are both supported.

```bash theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
git clone https://github.com/peaqnetwork/peaq-robotics-ros2.git
cd peaq-robotics-ros2
git checkout feature/peaqos-ros2-runtime

source /opt/ros/jazzy/setup.bash
# Use /opt/ros/humble/setup.bash inside the Docker image.

colcon build --packages-select peaq_ros2_interfaces peaq_ros2_stream
source install/setup.bash
```

Start it with the launch file (which declares the `config_yaml` argument):

```bash theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
ros2 launch peaq_ros2_stream stream_agent.launch.py \
  config_yaml:=/path/to/stream_config.yaml
```

Or run the node directly:

```bash theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
ros2 run peaq_ros2_stream stream_agent_node --ros-args \
  -p config.yaml_path:=/path/to/stream_config.yaml
```

Python dependencies (installed by `rosdep` or `pip`): `PyNaCl`, `PyYAML`, `boto3`, `google-api-python-client`, `google-auth`, `requests`. The agent is inert until `stream_agent.enabled` is `true`.

Configuration resolves in this order, last wins: built-in defaults, the `config_yaml` file, `PEAQOS_STREAM_*` environment variables, ROS parameters, then an optional standalone `policy_path` file.

## Configure: the Data Event Map

The agent reads one YAML block, `stream_agent` (the alias `stream` also works; keys accept snake\_case or camelCase). This is the [Data Event Map](https://docs.peaq.xyz/peaqos/concepts/data-streams.md): what to read, how to protect each field, where ciphertext goes, and who can decrypt.

```yaml stream_config.yaml theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
stream_agent:
  enabled: true                                  # inert unless true

  # Identity + backend (all four required when enabled)
  machine_id: "mach_01HXYZ"
  agent_id: "agent_ros2_stream"
  agent_token: "stk_live_xxx"                     # per-agent secret, keep it in env
  identity_ref: "did:peaq:0xMACHINE"              # written into every manifest signature
  api_base_url: "https://stream.example.com"      # your Stream API host; default http://127.0.0.1:8000
  api_key: ""                                     # optional; sent as Bearer

  # Topic subscriptions + per-field rules
  topics:
    - topic: "/battery_state"
      message_type: "sensor_msgs/msg/BatteryState"
      qos_preset: "sensor_data"                   # default | sensor_data | reliable
      field_rules:                                # ordered
        - { path: "voltage",        action: "include" }
        - { path: "serial_number",  action: "hash" }
        - { path: "location",       action: "anonymize" }
        - { path: "raw_cells",      action: "exclude" }
        - { path: "operator_note",  action: "encrypt", public_key_hex: "aa11…(32 bytes)" }
    - topic: "/cmd_vel"
      message_type: "geometry_msgs/msg/Twist"
      qos_preset: "reliable"

  # Who can unwrap each chunk's key (≥1 required; buyers are added later, on purchase)
  key_recipients:
    - { recipient_id: "owner-1",    recipient_type: "owner",    public_key_hex: "11…(32 bytes)" }
    - { recipient_id: "operator-1", recipient_type: "operator", public_key_hex: "22…(32 bytes)" }
    - { recipient_id: "mach_01",    recipient_type: "machine",  public_key_hex: "33…(32 bytes)" }

  # Where encrypted chunks go (a local copy is always written first)
  storage:
    backend: "walrus"                             # local | walrus | s3 | google-drive
    walrus:
      publisher_url:  "https://walrus-publisher.example"
      aggregator_url: "https://walrus-aggregator.example"
      epochs: 5
      permanent: true

  # On-chain anchoring: keep disabled
  peaqos_event:
    enabled: false

  # Local HTTP server that serves purchased chunks to buyers
  delivery:
    enabled: true
    host: "127.0.0.1"
    port: 8765
    token: "deliver_xxx"                          # required when delivery.enabled
```

Validation is strict: when `enabled`, `machine_id` / `agent_id` / `agent_token` / `identity_ref` and at least one topic and one key recipient are required; `recipient_type` is `machine` / `owner` / `operator`; each `public_key_hex` is a 32-byte (64-hex) x25519 key; `storage.backend` outside the known set falls back to `local`; `delivery.token` is required if delivery is on.

Local state defaults under `~/.peaq_robot/` (signing key, sequence counters, chunk `.bin` files, manifests, the SQLite catalog and key store) and is configurable per path. An offline `buffer` (SQLite) holds envelopes when the backend is unreachable and drains on a retry timer.

## Field rules

Rules run **before** signing, so protected fields never leave the machine in the clear while the package stays verifiable. Rules are ordered; if any `include` rule is present the agent starts from an allow-list, otherwise from the full message.

| Action | Effect |
| :- | :- |
| `include` | Keep the field as-is (and, if used, allow-list it). |
| `exclude` | Drop the field entirely. |
| `hash` | Replace the value with `sha256:<hex>`. |
| `anonymize` | Replace the value with `{ "redacted": true }`. |
| `encrypt` | Seal the value to the rule's `public_key_hex` (x25519 sealed box); only that key holder can read it. |

## How a message becomes a sellable chunk

1. **Capture.** The subscription fires; the ROS message is converted to a canonical ordered dict.
2. **Transform.** Field rules apply (above).
3. **Sequence.** A monotonic per-(machine, topic, policy version) number is assigned.
4. **Sign.** The agent builds a `peaqos-stream-envelope@v1` carrying the `payloadHash` and signs it with the machine's **Ed25519** key (`signature.keyId`, `algorithm: "ed25519"`).
5. **Encrypt.** The transformed payload is encrypted with a fresh per-chunk **XChaCha20-Poly1305** key (32-byte key, 24-byte nonce), yielding `plaintextHash`, `encryptedDataHash`, and a `keyCommitment`.
6. **Chunk ID.** A deterministic `sha256:` id is computed over `{ schemaVersion, previousChunkId, index, plaintextHash, encryptedDataHash }`, forming a hash chain (A run of chunks linked in order, each referencing the one before it, so missing, reordered, or edited chunks are detectable. Tamper-evidence for a continuous stream.).
7. **Store.** The chosen adapter writes the ciphertext and returns a `storageRef`.
8. **Manifest.** A `peaq.stream.chunks.v1` manifest wraps the chunk key to every `key_recipient` (x25519 sealed box) and is Ed25519-signed over `encryptedDataHash`, then `POST`ed to `/api/v1/stream/chunks`. The envelope is `POST`ed to `/api/v1/stream/events`, returning a receipt.

On startup the agent syncs its policy to the backend and registers its Ed25519 **public** key (`keyId = {agent_id}-stream-ed25519`).

## Storage adapters

Every adapter writes the local `.bin` first (so the delivery server always has a copy), then pushes to the remote.

| Backend | Required config | `storageRef` |
| :- | :- | :- |
| `local` | `chunk_storage_path` | `file://<abs-path>` |
| `walrus` | `walrus.publisher_url` (+ `aggregator_url` for reads) | `walrus://<blobId>` |
| `s3` | `s3.bucket` (+ region / endpoint / creds) | `s3://<bucket>/<prefix>/<id>.bin` |
| `google-drive` | `google_drive.folder_id` + `credentials_path` | `gdrive://<fileId>` |

## Selling and delivery

The agent polls the backend for orders. When an order is `paid`, it looks up each purchased chunk's symmetric key from the local key store, re-wraps it to the buyer's public key (x25519 sealed box, `peaq.stream.buyer-access.v1`), and submits the access grant (What a buyer receives after paying: the chunk keys they bought, each locked to their public key. They unlock with their private key and decrypt only those chunks.). The chunk data is never re-encrypted.

If `delivery.enabled`, a token-gated local HTTP server serves the encrypted chunks:

| Route | Returns |
| :- | :- |
| `GET /health` | `{ ok: true }` |
| `GET /chunks` | Catalog list (filter by topic, time, status) |
| `GET /chunks/{id}/manifest` | The `peaq.stream.chunks.v1` manifest |
| `GET /chunks/{id}/data` | Raw encrypted bytes (supports HTTP Range) |

## Environment variables

Every config field has a `PEAQOS_STREAM_*` override, applied above the YAML file and below ROS params. Use them to keep secrets out of committed YAML. The essentials:

```bash theme={"theme":{"light":"github-light-default","dark":"github-dark"}}
export PEAQOS_STREAM_ENABLED=true
export PEAQOS_STREAM_API_BASE_URL=https://stream.example.com
export PEAQOS_STREAM_MACHINE_ID=mach_01HXYZ
export PEAQOS_STREAM_AGENT_ID=agent_ros2_stream
export PEAQOS_STREAM_AGENT_TOKEN=stk_live_xxx
export PEAQOS_STREAM_IDENTITY_REF=did:peaq:0xMACHINE
export PEAQOS_STREAM_STORAGE_BACKEND=walrus
```

Storage credentials also accept the standard vendor variables: Walrus (`WALRUS_PUBLISHER_URL`, …), S3 (`AWS_ACCESS_KEY_ID`, `AWS_REGION`, `AWS_ENDPOINT_URL_S3`, …), and Google Drive (`GOOGLE_APPLICATION_CREDENTIALS`, `GOOGLE_DRIVE_FOLDER_ID`).

## Security

* The **Ed25519 signing key** is generated on first run, written to `signing_key_path` (`0600`), and never sent or logged. Only the public key is registered and embedded in manifests.
* **Per-chunk symmetric keys** live only in the local SQLite key store and travel only as sealed-box-wrapped blobs (to recipients, then to the buyer). Plaintext keys never appear in ROS messages, manifests, the backend payload, or logs.
* **No secrets cross ROS.** The agent publishes no topics and, with anchoring disabled, makes no ROS service calls.
* The delivery server silences request logging and gates every route except `/health` behind `delivery.token`. Keep `agent_token`, `api_key`, `delivery.token`, and storage credentials in environment variables, not committed YAML.

## Related

* [Stream function](https://docs.peaq.xyz/peaqos/functions/stream.md)
* [Data streams concept](https://docs.peaq.xyz/peaqos/concepts/data-streams.md)
* [Stream SDK reference](https://docs.peaq.xyz/peaqos/sdk-reference/stream.md)
* [peaqOS CLI: `peaqos stream`](https://docs.peaq.xyz/peaqos/cli#peaqos-stream)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.