738 lines
31 KiB
Markdown
738 lines
31 KiB
Markdown
# Walkthrough: msghandler
|
||
|
||
**Version**: 1.7.0
|
||
**Date**: 2026-09-16
|
||
**Status**: Active
|
||
**Ground Truth**: [`src/msghandler.jl`](../src/msghandler.jl)
|
||
**ASG Framework Alignment**: Eight pillars — Requirements → Solution Design → Walkthrough → Specification → UI Specification → Validation → Implementation Plan → Runbook
|
||
|
||
---
|
||
|
||
## Structural Guidance
|
||
|
||
This walkthrough follows the ASG Framework walkthrough guideline: one section per system **participant**, each tracing that participant's complete journey from session start to task completion. Every major step carries a `WT-XXX` id and cites the `SD-XXX` decision(s) and `FR-XXX` requirement(s) it implements. Where a participant's action triggers or depends on another participant, a cross-reference is included.
|
||
|
||
- **Overview sequence diagram** — participant section headers each start with the participant's high-level flow.
|
||
- **Detailed sequence diagrams** — per scenario, showing request/response and transport selection.
|
||
- **State machine** — the direct→link transport decision (multi-participant state).
|
||
- **Error paths** — each participant section covers the failure modes that participant sees.
|
||
|
||
**Participants**: JavaScript (browser/frontend), Julia (backend — ground truth), Python (backend), Rust (service), MicroPython (IoT device).
|
||
|
||
---
|
||
|
||
## 0. Overview: The Big Picture (all participants)
|
||
|
||
msghandler delivers two tiny, symmetric functions that **every participant calls the same way**: `smartpack()` (send) and `smartunpack()` (receive). A developer passes a list of `(dataname, data, type)` tuples and receives them back on the other side with types intact, without ever choosing how the data moves. Per payload, msghandler decides whether to send it inline (direct) or hand it to a file server (link).
|
||
|
||
```
|
||
Sender (smartpack) Transport Layer Receiver (smartunpack)
|
||
┌─────────────────┐ ┌───────────────┐ ┌───────────────────┐
|
||
│ 1. Data tuples │────────────>│ │───────────>│ 1. Parse envelope │
|
||
│ [(name, │ JSON │ Message │ JSON │ 2. Check transport│
|
||
│ data, type)]│ format │ Broker │ format │ 3. Fetch/Decode │
|
||
└─────────────────┘ │ (NATS/MQTT/ │ │ 4. Return tuples │
|
||
│ WebSocket) │ └───────────────────┘
|
||
└───────────────┘
|
||
```
|
||
|
||
### 0.1 Transport Layer Pattern (shared across all participants)
|
||
|
||
**Critical**: msghandler is **transport-agnostic**. The caller is responsible for:
|
||
|
||
1. **Sending side**: call `smartpack()` → receive `(envelope, json_string)` → publish the JSON string via transport
|
||
2. **Receiving side**: subscribe to transport → receive JSON string → call `smartunpack()`
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
subgraph msghandler["msghandler Module (all participants)"]
|
||
direction TB
|
||
S1["smartpack(data)"]
|
||
S2["Returns (envelope, json_string)"]
|
||
R1["smartunpack(json_string)"]
|
||
R2["Returns payloads (tuples)"]
|
||
end
|
||
|
||
subgraph Transport["Transport Layer (Caller's Responsibility)"]
|
||
direction TB
|
||
PUBLISH["Publish JSON string"]
|
||
SUBSCRIBE["Subscribe to messages"]
|
||
end
|
||
|
||
S1 --> S2
|
||
S2 --> PUBLISH
|
||
SUBSCRIBE --> R1
|
||
R1 --> R2
|
||
PUBLISH --> SUBSCRIBE
|
||
|
||
style msghandler fill:#b3e5fc,stroke:#0288d1
|
||
style Transport fill:#ffe0b2,stroke:#f57c00
|
||
```
|
||
|
||
### 0.2 Per-Payload Transport Decision (state machine — spans all participants)
|
||
|
||
`smartpack()` serializes each payload first, then compares the serialized size against `size_threshold`. This per-payload decision is the single state transition shared by every participant.
|
||
|
||
```mermaid
|
||
stateDiagram-v2
|
||
[*] --> Serialize: smartpack((name, data, type))
|
||
Serialize --> Direct: size < size_threshold (base64 inline)
|
||
Serialize --> Link: size >= size_threshold (upload to file server)
|
||
Direct --> Envelope: encoding = base64/json/arrow-ipc
|
||
Link --> Envelope: encoding = none/json/arrow-ipc, data = URL
|
||
Envelope --> Publish: envelope_to_json() -> caller publishes
|
||
Publish --> Transport: JSON string over broker
|
||
Transport --> Unpack: smartunpack(json_string)
|
||
Unpack --> DirectData: direct -> decode + deserialize
|
||
Unpack --> LinkData: link -> download + deserialize
|
||
DirectData --> Return: (name, data, type)
|
||
LinkData --> Return: (name, data, type)
|
||
Return --> [*]
|
||
```
|
||
|
||
**Key Design Principles** (from `2-solution_design.md`):
|
||
|
||
| Principle | Description | Rationale |
|
||
|-----------|-------------|-----------|
|
||
| **Claim-Check Pattern** | Large payloads uploaded to HTTP server, URL sent via transport | Transport has message size limits; avoids overflow (`SD-001`) |
|
||
| **Automatic Transport Selection** | Direct (< threshold) vs Link (≥ threshold) based on size | Optimizes memory vs network I/O (`SD-002`) |
|
||
| **Unified Tuple Format** | Same `(dataname, data, type)` format across all platforms | Platform-agnostic I/O (`SD-004`) |
|
||
| **Base64 Encoding** | JSON-compatible binary data transport | Cross-transport compatibility (`SD-005`) |
|
||
| **Handler Abstraction** | Pluggable file server (Plik / AWS S3 / custom) | No vendor lock-in (`SD-003`) |
|
||
| **Transport Abstraction** | Works over NATS / MQTT / WebSocket | No caller changes (`SD-006`) |
|
||
| **Exponential Backoff** | Retry failed downloads | Handles transient failures (`SD-007`) |
|
||
| **Correlation ID Propagation** | One id threads through every step | Distributed tracing (`SD-008`) |
|
||
|
||
**Supported Transports**: NATS (WebSocket/TCP), MQTT (WebSocket/TCP), WebSocket (direct HTTP), HTTP (polling/long-polling). msghandler only handles **serialization/deserialization**; the caller owns transport.
|
||
|
||
**Supported Payload Types** (`FR-006`, `FR-007`): `text`, `dictionary`, `arrowtable`, `jsontable`, `image`, `audio`, `video`, `binary`.
|
||
|
||
---
|
||
|
||
## 1. JavaScript (Browser / Frontend)
|
||
|
||
This section traces the journey from the browser's point of view: it starts a session, sends mixed content (text + image), receives a reply, and handles a large-file upload via link transport.
|
||
|
||
> Cross-references: the browser's request is consumed by the **Julia** backend (see [Section 2](#2-julia-backend-ground-truth)) and the browser's reply is produced there. The browser also broadcasts to other clients (see [Section 4](#4-rust-service) / [Section 3](#3-python-backend) receive the same topic).
|
||
|
||
### 1.1 Session Start
|
||
|
||
**WT-001**: The browser loads the frontend module and establishes a transport connection to the broker.
|
||
|
||
> Cross-reference: `2-solution_design.md:SD-006` (transport abstraction over NATS/WebSocket).
|
||
|
||
**Sequence**:
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant User as Browser (JS)
|
||
participant JS as msghandler.js
|
||
participant T as Transport (NATS/WebSocket)
|
||
|
||
User->>JS: load frontend
|
||
JS->>T: connect({ servers: "ws://localhost:4222" })
|
||
T-->>JS: connected
|
||
JS->>T: subscribe("/agent/wine/api/v1/response")
|
||
T-->>JS: subscription active
|
||
```
|
||
|
||
### 1.2 Send Mixed Payloads (Chat)
|
||
|
||
**WT-002**: The browser calls `smartpack()` with a mixed payload (text message + avatar image).
|
||
|
||
```javascript
|
||
// JavaScript (Browser) - WT-002
|
||
const [env, msgJson] = await msghandler.smartpack(
|
||
"/agent/wine/api/v1/prompt",
|
||
[
|
||
["msg", "Hello! I'm Ton.", "text"],
|
||
["avatar", avatarImageData, "image"]
|
||
],
|
||
{
|
||
broker_url: "ws://localhost:4222",
|
||
receiver_name: "agent-backend",
|
||
msg_purpose: "chat"
|
||
}
|
||
);
|
||
```
|
||
|
||
> Traceability: `FR-006` (multi-payload), `FR-007` (type preservation), `FR-001` (cross-platform text), `SD-004` (unified tuple format), `SD-005` (base64), `SD-008` (correlation id auto-generated).
|
||
|
||
**WT-002.1**: For each payload, msghandler serializes and selects transport by size.
|
||
|
||
| Payload | Size | Transport | Encoding | Traceability |
|
||
|---------|------|-----------|----------|--------------|
|
||
| `"msg"` (text) | ~20 bytes | direct | base64 | `SD-002`, `SD-005` |
|
||
| `"avatar"` (image) | ~2 KB | direct | base64 | `SD-002`, `SD-005` |
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant User as Browser
|
||
participant JS as msghandler.js
|
||
participant E as Envelope
|
||
|
||
User->>JS: smartpack([msg, avatar])
|
||
JS->>JS: _serialize_data() per payload (SD-004)
|
||
JS->>JS: size < threshold? (SD-002)
|
||
JS->>E: build envelope (SD-005, SD-008)
|
||
E-->>JS: (env, msgJson)
|
||
```
|
||
|
||
**Rationale**: Text is smaller and sent via direct transport (fast, no file server). The avatar stays direct because it is under the threshold. `smartpack()` returns `(env, msgJson)` — **the browser must publish** (`FR-013`).
|
||
|
||
**WT-003**: The browser publishes the JSON string via the transport.
|
||
|
||
```javascript
|
||
// JavaScript (Browser) - WT-003
|
||
const conn = await transportClient.connect({ servers: "ws://localhost:4222" });
|
||
await conn.publish("/agent/wine/api/v1/prompt", msgJson);
|
||
```
|
||
|
||
> Cross-reference: this publish triggers the **Julia** backend (see [Section 2, WT-010](#2-julia-backend-ground-truth)). `SD-006`.
|
||
|
||
### 1.3 Envelope Shape (what crosses the wire)
|
||
|
||
**WT-004**: The published envelope matches the ground-truth `msg_envelope_v1` JSON shape in [`src/msghandler.jl`](../src/msghandler.jl:274).
|
||
|
||
```json
|
||
{
|
||
"correlation_id": "a1b2c3d4...",
|
||
"msg_id": "e5f6g7h8...",
|
||
"timestamp": "2026-09-16T02:08:27.000Z",
|
||
"send_to": "/agent/wine/api/v1/prompt",
|
||
"msg_purpose": "chat",
|
||
"sender_name": "browser-frontend",
|
||
"sender_id": "browser-uuid...",
|
||
"receiver_name": "agent-backend",
|
||
"receiver_id": "",
|
||
"reply_to": "/agent/wine/api/v1/response",
|
||
"reply_to_msg_id": "",
|
||
"broker_url": "ws://localhost:4222",
|
||
"metadata": {},
|
||
"payloads": [
|
||
{
|
||
"id": "payload-uuid...",
|
||
"dataname": "msg",
|
||
"payload_type": "text",
|
||
"transport": "direct",
|
||
"encoding": "base64",
|
||
"size": 20,
|
||
"data": "SGVsbG8hIEknIHRlbCB5b3UgSW4gZW5nbGlzaC4=",
|
||
"metadata": { "payload_bytes": 20 }
|
||
},
|
||
{
|
||
"id": "payload-uuid...",
|
||
"dataname": "avatar",
|
||
"payload_type": "image",
|
||
"transport": "direct",
|
||
"encoding": "base64",
|
||
"size": 2048,
|
||
"data": "iVBORw0KGgo=",
|
||
"metadata": { "payload_bytes": 2048 }
|
||
}
|
||
]
|
||
}
|
||
```
|
||
|
||
> Traceability: `FR-011` (correlation id), `FR-013` (publish), `SD-008`, `SD-005`.
|
||
|
||
### 1.4 Receive Reply
|
||
|
||
**WT-005**: The browser receives the backend reply on its `reply_to` topic and deserializes payloads.
|
||
|
||
```javascript
|
||
// JavaScript (Browser) - WT-005
|
||
const conn = await transportClient.connect({ servers: "ws://localhost:4222" });
|
||
conn.subscribe("/agent/wine/api/v1/response", async (msg) => {
|
||
const payloads = await msghandler.smartunpack(String(msg.payload));
|
||
// payloads = [
|
||
// ["response", "Hello Ton! I'm the AI assistant.", "text"],
|
||
// ["generated_image", binaryData, "image"]
|
||
// ];
|
||
renderResponse(payloads);
|
||
});
|
||
```
|
||
|
||
> Traceability: `FR-014` (subscription), `FR-007` (type preservation), `SD-004`, `SD-007`.
|
||
|
||
### 1.5 Error Paths (browser POV)
|
||
|
||
**WT-006**: Large file (≥ threshold) triggers link transport; upload fails.
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant User as Browser
|
||
participant JS as msghandler.js
|
||
participant FS as Plik File Server
|
||
participant T as Transport
|
||
|
||
User->>JS: smartpack([dataset], size_threshold)
|
||
JS->>JS: size >= threshold -> link (SD-001)
|
||
JS->>FS: plik_oneshot_upload()
|
||
FS-->>JS: 500 Upload Failed
|
||
JS->>JS: throw "Failed to upload data to fileserver"
|
||
User->>User: surface upload error / fall back to smaller payloads
|
||
```
|
||
|
||
> Traceability: `FR-003` (large file handling), `FR-008` (Plik), `SD-001`, `SD-003`. Recovery: fall back to a smaller payload or direct transport.
|
||
|
||
**WT-007**: Transport connection lost before publish — the caller's transport client handles auto-reconnect; msghandler has already returned `(env, msgJson)`, so the caller retries the publish.
|
||
|
||
> Traceability: `NFR-203` (connection recovery).
|
||
|
||
---
|
||
|
||
## 2. Julia (Backend — Ground Truth)
|
||
|
||
This section traces the Julia backend, which is the reference implementation in [`src/msghandler.jl`](../src/msghandler.jl). It receives the browser's chat message, processes it, sends a reply, and downloads a large file via link transport.
|
||
|
||
> Cross-references: consumes the browser publish (see [Section 1, WT-003](#13-envelope-shape-what-crosses-the-wire)); produces the browser reply (see [Section 1, WT-005](#15-receive-reply)). Also the topic is broadcast to **Python** and **Rust** subscribers (see [Section 3](#3-python-backend), [Section 4](#4-rust-service)).
|
||
|
||
### 2.1 Session Start
|
||
|
||
**WT-010**: The Julia backend subscribes to the topic and is ready to `smartunpack()`.
|
||
|
||
> Cross-reference: `2-solution_design.md:SD-006` (transport abstraction).
|
||
|
||
### 2.2 Receive & Unpack Chat Message
|
||
|
||
**WT-011**: The backend receives the JSON string from the transport and calls `smartunpack()`.
|
||
|
||
```julia
|
||
# Julia (Backend) - WT-011
|
||
transport_msg = transport_subscription.next()
|
||
env = smartunpack(String(transport_msg.payload))
|
||
|
||
# env["payloads"] is now a Vector of tuples:
|
||
# [("msg", "Hello! I'm Ton.", "text"),
|
||
# ("avatar", UInt8[...], "image")]
|
||
```
|
||
|
||
> Traceability: `FR-014` (subscription), `FR-001` (text), `FR-006` (multi-payload), `FR-007` (type preservation), `SD-004`, `SD-007`.
|
||
|
||
**WT-011.1**: On the direct path, `smartunpack()` base64-decodes each payload and deserializes by type (`_deserialize_data`).
|
||
|
||
```julia
|
||
# _deserialize_data (src/msghandler.jl:940)
|
||
# text -> String(data)
|
||
# dictionary-> JSON.parse(json_str)
|
||
# arrowtable-> Arrow.Table(io) -> DataFrame
|
||
# image/audio/video/binary -> data (bytes)
|
||
```
|
||
|
||
> Traceability: `SD-004`, `SD-005`.
|
||
|
||
### 2.3 Send Reply
|
||
|
||
**WT-012**: The backend processes the message and builds a reply with `smartpack()`.
|
||
|
||
```julia
|
||
# Julia (Backend) - WT-012
|
||
response_text = "Hello Ton! I'm the AI assistant."
|
||
generated_image = generate_ai_image(response_text)
|
||
|
||
env, msg_json = smartpack(
|
||
"/agent/wine/api/v1/response",
|
||
(
|
||
("response", response_text, "text"),
|
||
("generated_image", generated_image, "image"),
|
||
),
|
||
reply_to = "/chat/user/v1/message",
|
||
reply_to_msg_id = env["msg_id"]
|
||
)
|
||
publish(transport_conn, "/agent/wine/api/v1/response", msg_json)
|
||
```
|
||
|
||
> Traceability: `FR-001` (reply text), `FR-013` (publish — caller responsible), `FR-011` (`reply_to_msg_id` links the reply), `SD-004`, `SD-005`, `SD-008`.
|
||
|
||
**Rationale**: `reply_to_msg_id = env["msg_id"]` correlates this reply to the original request (`SD-008`). `smartpack()` returns `(env, env_json_str)` — **the backend must publish** (`FR-013`).
|
||
|
||
### 2.4 Receive Large File (Link Transport)
|
||
|
||
**WT-013**: A client uploads a 10 MB file; msghandler downloads it with exponential backoff.
|
||
|
||
```julia
|
||
# Julia (Backend) - WT-013
|
||
transport_msg = transport_subscription.next()
|
||
env = smartunpack(String(transport_msg.payload))
|
||
# The "dataset" payload has transport="link"; smartunpack():
|
||
# 1. extracts the URL from payload["data"]
|
||
# 2. downloads with exponential backoff (_fetch_with_backoff)
|
||
# 3. deserializes (binary -> bytes)
|
||
```
|
||
|
||
> Traceability: `FR-003` (large file handling), `FR-008` (Plik), `FR-010` (backoff retry), `SD-001`, `SD-002`, `SD-003`, `SD-007`.
|
||
|
||
**State (link path)** — from the backend's POV, per-payload decision `SD-002`:
|
||
```mermaid
|
||
stateDiagram-v2
|
||
[*] --> InspectEnvelope: smartunpack(json_string)
|
||
InspectEnvelope --> Direct: transport == "direct"
|
||
InspectEnvelope --> Link: transport == "link"
|
||
Direct --> Deserialize: base64decode + _deserialize_data
|
||
Link --> Download: fileserver_download_handler (SD-007)
|
||
Download --> Deserialize: download_failed? retry w/ backoff
|
||
Deserialize --> Return: push (dataname, data, type)
|
||
Return --> [*]
|
||
```
|
||
|
||
### 2.5 Error Paths (Julia POV)
|
||
|
||
**WT-014**: File server download fails after `max_retries` (`FR-010`, `SD-007`).
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant BJ as Julia Backend
|
||
participant D as _fetch_with_backoff
|
||
participant FS as Plik File Server
|
||
|
||
BJ->>D: fetch(url, 5, 100, 5000)
|
||
FS-->>D: 503 / connection refused
|
||
D->>D: sleep(backoff), double delay (SD-007)
|
||
loop up to max_retries (5)
|
||
D->>FS: retry
|
||
FS-->>D: still failing
|
||
end
|
||
D->>BJ: error("Failed to fetch data after 5 attempts")
|
||
BJ->>BJ: log_trace(correlation_id, "download retry exceeded")
|
||
```
|
||
|
||
> Traceability: `FR-010` (backoff retry), `NFR-202` (file server availability), `NFR-404` (alerting). Recovery: caller retries with a new upload; correlation id preserves traceability (`SD-008`).
|
||
|
||
**WT-015**: Unknown payload type throws `error("Unknown payload_type: $payload_type")` (`FR-007`).
|
||
|
||
---
|
||
|
||
## 3. Python (Backend)
|
||
|
||
This section traces the Python backend: it receives tabular data (DataFrame), analyzes it, and sends results back. Python supports `arrowtable` (desktop) and `dictionary` types.
|
||
|
||
> Cross-references: Python may publish to a topic that **Julia** and **Rust** also subscribe to (see [Section 2](#2-julia-backend-ground-truth), [Section 4](#4-rust-service)).
|
||
|
||
### 3.1 Session Start
|
||
|
||
**WT-020**: The Python backend imports `msghandler` and connects its transport consumer.
|
||
|
||
### 3.2 Send Tabular Data (Round-Trip)
|
||
|
||
**WT-021**: Python sends a DataFrame as `arrowtable`.
|
||
|
||
```python
|
||
# Python - WT-021
|
||
import pandas as pd
|
||
from msghandler import smartpack
|
||
|
||
df = pd.DataFrame({"id": [1, 2, 3], "name": ["Alice", "Bob", "Charlie"], "score": [95, 88, 92]})
|
||
env, msg_json = smartpack(
|
||
"/agent/wine/api/v1/analyze",
|
||
[("data", df, "arrowtable")],
|
||
broker_url="ws://localhost:4222",
|
||
receiver_name="agent-backend"
|
||
)
|
||
await transport_publisher.publish("/agent/wine/api/v1/analyze", msg_json)
|
||
```
|
||
|
||
> Traceability: `FR-002` (tabular data), `FR-012` (serialization), `SD-004`, `SD-005`, `SD-006`.
|
||
|
||
**WT-021.1**: Serialization to Arrow IPC preserves column types; the payload is `< threshold` so it travels direct with `encoding = "arrow-ipc"`.
|
||
|
||
> Traceability: `SD-002`, `SD-005`.
|
||
|
||
### 3.3 Receive Results
|
||
|
||
**WT-022**: Python receives the processed results and deserializes back to a DataFrame.
|
||
|
||
```python
|
||
# Python - WT-022
|
||
transport_msg = await transport_consumer.next()
|
||
env = await smartunpack(str(transport_msg.payload))
|
||
# env["payloads"] is now:
|
||
# [("results", <DataFrame>, "arrowtable"),
|
||
# ("summary", "Analysis complete: 1500 rows processed", "text")]
|
||
```
|
||
|
||
> Traceability: `FR-014`, `FR-002`, `FR-007`, `SD-004`, `SD-007`.
|
||
|
||
### 3.4 Error Paths (Python POV)
|
||
|
||
**WT-023**: Large `arrowtable` (≥ threshold) uses link transport; upload fails.
|
||
|
||
> Traceability: `FR-003`, `SD-001`, `SD-003`. Recovery: retry with a smaller chunk or direct transport.
|
||
|
||
**WT-024**: `arrowtable` sent but `pyarrow` not installed → `_serialize_data` raises `RuntimeError` (no Arrow support).
|
||
|
||
> Traceability: `FR-002`.
|
||
|
||
---
|
||
|
||
## 4. Rust (Service)
|
||
|
||
This section traces the Rust service, which leverages compile-time type safety via enums and `serde` for serialization, and a tokio async runtime for transport and HTTP. Rust supports the full feature set including `arrowtable` and link transport.
|
||
|
||
> Cross-references: Rust subscribes to the same topic consumed by **Julia** and **Python** (see [Section 2](#2-julia-backend-ground-truth), [Section 3](#3-python-backend)).
|
||
|
||
### 4.1 Session Start
|
||
|
||
**WT-030**: The Rust service connects and subscribes under a tokio runtime.
|
||
|
||
### 4.2 Receive & Dispatch by Type
|
||
|
||
**WT-031**: Rust receives and dispatches payloads by `payload_type`.
|
||
|
||
```rust
|
||
// Rust service - WT-031
|
||
use msghandler::{smartunpack, MsgEnvelopeV1};
|
||
use base64::Engine as _;
|
||
|
||
#[tokio::main]
|
||
async fn main() {
|
||
let mut sub = transport_client::connect("ws://localhost:4222")
|
||
.unwrap().subscribe("/agent/wine/api/v1/analyze");
|
||
|
||
for msg in sub.messages() {
|
||
let env = smartunpack(&String::from_utf8_lossy(&msg.payload), &Default::default()).await.unwrap();
|
||
for payload in &env.payloads {
|
||
match payload.payload_type.as_str() {
|
||
"arrowtable" => {
|
||
let arrow_bytes = BASE64.decode(&payload.data).unwrap();
|
||
println!("Received arrowtable payload ({} bytes)", arrow_bytes.len());
|
||
},
|
||
"text" => println!("Message: {}", payload.data),
|
||
"image" | "audio" | "video" | "binary" => {
|
||
let bytes = BASE64.decode(&payload.data).unwrap();
|
||
println!("Received {} bytes of {}", bytes.len(), payload.payload_type);
|
||
},
|
||
"dictionary" | "jsontable" => println!("Data: {}", payload.data),
|
||
_ => println!("Unknown payload type: {}", payload.payload_type),
|
||
}
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
> Traceability: `FR-007` (type preservation), `FR-006` (multi-payload), `SD-004`, `SD-005`. The `payload_type` field determines how to interpret the `data` string (type dispatch).
|
||
|
||
### 4.3 Send Type-Safe Results
|
||
|
||
**WT-032**: Rust sends mixed results using the enum-based `Payload` API.
|
||
|
||
```rust
|
||
// Rust service - WT-032
|
||
let (envelope, json_str) = smartpack(
|
||
"/agent/wine/api/v1/results",
|
||
&[
|
||
("results".to_string(), Payload::ArrowTable(result_bytes), "arrowtable".to_string()),
|
||
("summary".to_string(), Payload::Text("Analysis complete: 1500 rows".to_string()), "text".to_string()),
|
||
],
|
||
&smartpackOptions {
|
||
broker_url: DEFAULT_BROKER_URL.to_string(),
|
||
reply_to: "/python/worker/v1/results".to_string(),
|
||
msg_purpose: "chat".to_string(),
|
||
..Default::default()
|
||
},
|
||
).await?;
|
||
|
||
conn.publish("/agent/wine/api/v1/results", &json_str)?;
|
||
```
|
||
|
||
> Traceability: `FR-013` (publish), `FR-001`, `SD-004`, `SD-006`.
|
||
|
||
### 4.4 Large File via Link Transport
|
||
|
||
**WT-033**: Rust uploads a large binary file; `smartpack()` selects link transport and uploads via `reqwest`.
|
||
|
||
```rust
|
||
// Rust service - WT-033
|
||
let large_file_data: Vec<u8> = std::fs::read("/data/large_dataset.parquet")?;
|
||
let (envelope, json_str) = smartpack(
|
||
"/agent/wine/api/v1/upload",
|
||
&[("dataset".to_string(), Payload::Binary(large_file_data), "binary".to_string())],
|
||
&smartpackOptions {
|
||
fileserver_url: DEFAULT_FILESERVER_URL.to_string(),
|
||
size_threshold: DEFAULT_SIZE_THRESHOLD,
|
||
..Default::default()
|
||
},
|
||
).await?;
|
||
conn.publish("/agent/wine/api/v1/upload", &json_str)?;
|
||
```
|
||
|
||
> Traceability: `FR-003`, `SD-001`, `SD-002`, `SD-003`, `SD-006`.
|
||
|
||
### 4.5 Error Paths (Rust POV)
|
||
|
||
**WT-034**: Download exceeds retries → `MsgHandlerError::DownloadFailed { url, retries }` (`FR-010`, `SD-007`).
|
||
|
||
> Traceability: `FR-010`, `NFR-202`. Recovery: caller retries with a fresh upload; idiomatic `Result<T, E>`.
|
||
|
||
---
|
||
|
||
## 5. MicroPython (IoT Device)
|
||
|
||
This section traces the MicroPython sensor device. Due to memory constraints (256 KB–1 MB), MicroPython is limited to `text` and `dictionary` types and **direct transport only** (no link / file-server support).
|
||
|
||
> Cross-references: the device publishes readings that a **Python** backend consumes (see [Section 3](#3-python-backend)).
|
||
|
||
### 5.1 Session Start
|
||
|
||
**WT-040**: The device boots, connects to the MQTT broker, and publishes a sensor reading.
|
||
|
||
### 5.2 Send Sensor Data (Direct Only)
|
||
|
||
**WT-041**: MicroPython sends a `dictionary` payload under its lower threshold.
|
||
|
||
```python
|
||
# MicroPython - WT-041
|
||
from msghandler import smartpack
|
||
|
||
sensor_data = {"temperature": 25.5, "humidity": 60.0, "pressure": 1013.25}
|
||
env, msg_json = smartpack(
|
||
"/sensor/device/v1/readings",
|
||
[("data", sensor_data, "dictionary")],
|
||
broker_url="broker.local:1883",
|
||
size_threshold=100000 # 100 KB for MicroPython
|
||
)
|
||
mqtt_client.publish("/sensor/device/v1/readings", msg_json)
|
||
```
|
||
|
||
> Traceability: `FR-005` (MicroPython support), `FR-006` (multi-payload), `FR-012` (serialization), `SD-004`, `SD-005`, `SD-006`.
|
||
|
||
**WT-041.1**: Serialization is JSON → UTF-8 → base64 (no Arrow). Because the payload is under `size_threshold`, it travels direct.
|
||
|
||
> Traceability: `SD-002`, `SD-005`.
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
subgraph Device["MicroPython Device"]
|
||
D1["sensor_data = {temp, humidity}"]
|
||
D2["smartpack() -> JSON"]
|
||
D3["publish via MQTT"]
|
||
end
|
||
subgraph Broker["MQTT Broker"]
|
||
B["broker.local:1883"]
|
||
end
|
||
subgraph Backend["Python Backend"]
|
||
P1["subscribe via MQTT"]
|
||
P2["smartunpack(JSON)"]
|
||
P3["get dictionary"]
|
||
end
|
||
D1 --> D2 --> D3 --> B --> P1 --> P2 --> P3
|
||
```
|
||
|
||
### 5.3 Receive Acknowledgment
|
||
|
||
**WT-042**: The device receives an ACK/dictionary reply from the backend.
|
||
|
||
```python
|
||
# MicroPython - WT-042
|
||
env = smartunpack(str(transport_msg.payload))
|
||
# env["payloads"] is now:
|
||
# [("data", {"temperature": 25.5, ...}, "dictionary")]
|
||
```
|
||
|
||
> Traceability: `FR-014`, `FR-007`, `SD-004`.
|
||
|
||
### 5.4 Error Paths (MicroPython POV)
|
||
|
||
**WT-043**: Payload exceeds the 100 KB threshold → MicroPython cannot use link transport (no file-server support) and must reject or chunk the payload.
|
||
|
||
> Traceability: `FR-005`, `FR-003`. Recovery: split into multiple direct payloads below the threshold.
|
||
|
||
**WT-044**: `dictionary` type but data is not JSON-serializable → serialization error (memory constraint).
|
||
|
||
> Traceability: `FR-012`.
|
||
|
||
---
|
||
|
||
## 6. Cross-Participant Flow (all participants)
|
||
|
||
The following end-to-end flow shows a JavaScript client, a Julia backend, a Python worker, and a MicroPython device on a shared chat topic.
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant JS as JavaScript (Browser)
|
||
participant J as Julia (Backend)
|
||
participant P as Python (Worker)
|
||
participant MP as MicroPython (Device)
|
||
participant T as Message Broker
|
||
|
||
JS->>T: publish "/chat/user/v1/message" (WT-003)
|
||
T->>J: deliver to Julia
|
||
J->>J: smartunpack() (WT-011)
|
||
J->>J: process, build reply (WT-012)
|
||
J->>T: publish reply (SD-006)
|
||
T->>P: deliver to Python
|
||
P->>P: smartunpack() (WT-022)
|
||
T->>MP: deliver to MicroPython
|
||
MP->>MP: smartunpack() (WT-042)
|
||
T->>JS: deliver reply to browser (WT-005)
|
||
```
|
||
|
||
> Traceability: `FR-001`, `FR-006`, `FR-013`, `FR-014`, `SD-004`, `SD-006`, `SD-008`.
|
||
|
||
---
|
||
|
||
## 7. Specification Traceability
|
||
|
||
Each walkthrough step below maps to the exact `SD-XXX` / `FR-XXX` artifacts it implements or depends on.
|
||
|
||
| WT ID | Participant | Step | SD Ref | FR Ref | Description |
|
||
|-------|-------------|------|--------|--------|-------------|
|
||
| WT-001 | JavaScript | Session start / connect | SD-006 | FR-013 | Establish transport connection |
|
||
| WT-002 | JavaScript | Send mixed payloads (text + image) | SD-004, SD-005 | FR-001, FR-006, FR-007 | smartpack() with mixed types |
|
||
| WT-002.1 | JavaScript | Per-payload transport selection | SD-002 | FR-004 | direct transport for < threshold |
|
||
| WT-003 | JavaScript | Publish JSON via transport | SD-006 | FR-013 | caller publishes |
|
||
| WT-004 | JavaScript | Envelope shape crosses wire | SD-005, SD-008 | FR-011, FR-013 | msg_envelope_v1 JSON |
|
||
| WT-005 | JavaScript | Receive reply | SD-004, SD-007 | FR-007, FR-014 | smartunpack() on reply |
|
||
| WT-006 | JavaScript | Large file upload fails (link) | SD-001, SD-003 | FR-003, FR-008 | link transport error |
|
||
| WT-007 | JavaScript | Transport connection lost | SD-006 | NFR-203 | caller retries publish |
|
||
| WT-010 | Julia | Session start / subscribe | SD-006 | FR-014 | transport subscription |
|
||
| WT-011 | Julia | Receive & unpack chat | SD-004, SD-007 | FR-001, FR-006, FR-007 | smartunpack() |
|
||
| WT-011.1 | Julia | Direct path decode/deserialize | SD-004, SD-005 | FR-007 | _deserialize_data |
|
||
| WT-012 | Julia | Send reply | SD-004, SD-005, SD-008 | FR-001, FR-013, FR-011 | smartpack() + reply_to_msg_id |
|
||
| WT-013 | Julia | Receive large file (link) | SD-001, SD-002, SD-003 | FR-003, FR-008 | link download |
|
||
| WT-014 | Julia | Download fails after retries | SD-007 | FR-010, NFR-202 | _fetch_with_backoff |
|
||
| WT-015 | Julia | Unknown payload type | SD-004 | FR-007 | error throw |
|
||
| WT-020 | Python | Session start / connect | SD-006 | FR-013 | transport consumer |
|
||
| WT-021 | Python | Send arrowtable | SD-004, SD-005 | FR-002, FR-012 | Arrow IPC direct |
|
||
| WT-021.1 | Python | Arrow IPC serialization | SD-002, SD-005 | FR-012 | encoding arrow-ipc |
|
||
| WT-022 | Python | Receive results | SD-004, SD-007 | FR-002, FR-007, FR-014 | smartunpack() |
|
||
| WT-023 | Python | Large arrowtable upload fails | SD-001, SD-003 | FR-003 | link transport error |
|
||
| WT-024 | Python | pyarrow not available | SD-005 | FR-002 | RuntimeError |
|
||
| WT-030 | Rust | Session start / subscribe | SD-006 | FR-013 | tokio transport |
|
||
| WT-031 | Rust | Receive & dispatch by type | SD-004, SD-005 | FR-006, FR-007 | payload_type dispatch |
|
||
| WT-032 | Rust | Send type-safe results | SD-004 | FR-001, FR-013 | Payload enum |
|
||
| WT-033 | Rust | Large file via link | SD-001, SD-002, SD-003 | FR-003 | reqwest upload |
|
||
| WT-034 | Rust | Download exceeds retries | SD-007 | FR-010, NFR-202 | DownloadFailed |
|
||
| WT-040 | MicroPython | Session start / publish | SD-006 | FR-005 | MQTT connect |
|
||
| WT-041 | MicroPython | Send dictionary (direct) | SD-004, SD-005 | FR-005, FR-006 | JSON base64 |
|
||
| WT-041.1 | MicroPython | JSON serialization, direct transport | SD-002, SD-005 | FR-012 | JSON -> base64 |
|
||
| WT-042 | MicroPython | Receive ACK | SD-004 | FR-007, FR-014 | smartunpack() |
|
||
| WT-043 | MicroPython | Threshold exceeded (no link) | SD-002 | FR-003, FR-005 | direct-only constraint |
|
||
| WT-044 | MicroPython | Non-serializable dictionary | SD-005 | FR-012 | serialization error |
|
||
| WT-050 | — | Cross-participant chat broadcast | SD-004, SD-006 | FR-001, FR-006, FR-013, FR-014 | shared topic flow |
|
||
|
||
---
|
||
|
||
## 8. Gap-Check Validation (Walkthrough)
|
||
|
||
| Stage Transition | Gap-Check Question | Status |
|
||
|------------------|-------------------|--------|
|
||
| Solution Design → Walkthrough | Does every `SD-XXX` decision appear in at least one participant's walkthrough step with a `WT-XXX` id? Is there a section for each system participant? | ✅ Verified — SD-001..SD-008 traced across JS/Julia/Python/Rust/MicroPython sections with WT-XXX ids; all 5 participants covered |
|
||
| Walkthrough → Specification | Does every `WT-XXX` walkthrough step trace to at least one `SP-XXX` specification rule? | ✅ Verified — every `WT-XXX` step maps to `SP-XXX` in the Walkthrough → Specification Trace of [`4-specification.md`](4-specification.md) |
|
||
|
||
---
|
||
|
||
## 9. Change Log
|
||
|
||
| Date | Version | Changes | Specification Ref | Solution Design Ref |
|
||
|------|---------|---------|-------------------|---------------------|
|
||
| 2026-09-16 | 1.7.0 | Restructured per ASG Framework walkthrough guideline: per-participant sections (JS/Julia/Python/Rust/MicroPython), `WT-XXX` step ids, `SD-XXX`/`FR-XXX` citations, per-participant sequence diagrams, cross-participant flow, and error paths | All sections | SD-001 through SD-008 |
|
||
| 2026-09-16 | 1.6.0 | Aligned ground-truth API to `src/msghandler.jl` (smartpack returns tuple, smartunpack returns payloads as tuples); fixed Rust payload access | specification.md | SD-001 through SD-008 |
|
||
| 2026-05-22 | 1.5.0 | Updated to ASG Framework eight pillars — added solution-design traceability (SD-XXX) to the traceability table | All sections | SD-001 through SD-008 |
|
||
| 2026-05-15 | 1.4.0 | Made transport layer agnostic; removed NATS-specific references | All sections | SD-001 through SD-008 |
|
||
| 2026-05-13 | 1.2.0 | Added Rust support (tokio, serde, arrow2) | specification.md:11 | SD-001 through SD-008 |
|
||
| 2026-03-23 | 1.0.0 | Updated to ASG Framework walkthrough guidelines | All sections | SD-001 through SD-008 |
|
||
|
||
---
|
||
|
||
*This walkthrough document is versioned and maintained in git alongside the codebase. The technical details are grounded in [`src/msghandler.jl`](../src/msghandler.jl) (the reference implementation) and kept consistent with the Python, JavaScript (CSR), Rust, and MicroPython sources. All implementations must adhere to this documentation.*
|