diff --git a/docs/superpowers/plans/2026-10-18-tines-addressed-delivery.md b/docs/superpowers/plans/2026-10-18-tines-addressed-delivery.md new file mode 100644 index 0000000..0a38565 --- /dev/null +++ b/docs/superpowers/plans/2026-10-18-tines-addressed-delivery.md @@ -0,0 +1,283 @@ +# Addressed Delivery Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add addressed store-and-forward to tines — an observation can name a recipient (`to`), and any node carries it (multi-hop, epidemic, lifetime-bounded) until it reaches that recipient or expires. + +**Architecture:** Extend `Observation` with `to: Option` and `lifetime_ms: Option`; the content ID now includes `to`. Broadcast (`to = None`) keeps the existing bloom-summary gossip unchanged. Addressed (`to = Some(C)`) messages live in the same store, are excluded from blooms, and are forwarded on contact — delivered to the recipient, epidemic-forwarded to anyone else — and dropped once expired or over the store's `max_count`. + +**Tech Stack:** Rust (`#![no_std]` core + `std` simulator), postcard (wire), Blake2s-256 (identity), heapless, proptest. + +**Spec:** `docs/superpowers/specs/2026-10-18-tines-addressed-delivery.md` + +## Global Constraints + +- `tines-core` is `#![no_std]`; `alloc` is feature-gated (default on); only `MemStore` may require `alloc`. `--no-default-features` must build for `thumbv7em-none-eabihf`. +- No async, threads, I/O, or panics on untrusted wire input. +- Identity = `first16(Blake2s-256(observer ‖ seq_le ‖ topic_len_le(u16) ‖ topic ‖ payload_len_le(u16) ‖ payload ‖ to_marker))`, where `to_marker` is `0x00` for `None`, `0x01 ‖ node_id` for `Some`. `lifetime_ms` is excluded from identity. +- Invariant: `lifetime_ms.is_some() == to.is_some()`. +- Addressed messages (`to = Some(..)`) never appear in a bloom summary. +- Expired addressed messages are never stored (expiry checked before insertion). +- An addressed message is expired when `current_age > lifetime_ms` (age accumulates per hop). +- Constants unchanged: `MAX_TOPIC=64`, `MAX_PAYLOAD=1024`, `MAX_PEERS=8`, `SUMMARY_BYTES=512`, `BLOOM_K=3`, `SUMMARY_INTERVAL_MS=2000`, `REASSEMBLY_TIMEOUT_MS=5000`, `SEND_QUEUE_CAP=256`, `MAX_FRAME=1500`. + +## Review Focus + +1. A wire frame whose `to`/`lifetime_ms` violate the invariant (`to = Some` but `lifetime_ms = None`, or the reverse) is dropped, never stored, never panics. (Task 4) +2. An expired addressed message is rejected on both the `Obs` and the reassembled-`Frag` receive paths. (Task 4) +3. A message addressed to a peer whose `NodeId` is not yet known is still held and epidemic-forwarded, never dropped on sight. (Task 5) +4. A message never bounces back to the peer it was received from (loop prevention). (Task 5) +5. `max_count` trim removes the oldest addressed messages too, not only broadcast. (Task 2) + +--- + +### Task 1: Observation gains `to`/`lifetime_ms`; identity includes `to` + +**Files:** +- Modify: `tines-core/src/observation.rs` (fields, `id()`, `aged()`) +- Modify (add `to: None, lifetime_ms: None` to every literal): `tines-core/src/node.rs` (`observe`), and test helpers in `observation.rs`, `mem_store.rs`, `belief.rs`, `exchange.rs`, `frag.rs` +- Test: `tines-core/src/observation.rs` + +**Interfaces:** +- Produces: `Observation { observer: NodeId, seq: u64, age_ms: u64, observed_at: Option, topic: Topic, payload: Payload, to: Option, lifetime_ms: Option }`. +- `id()`: append `to_marker` to the hash — one byte `0x00` if `to.is_none()`, else `0x01` followed by the 16 `NodeId` bytes. +- `aged(elapsed_ms)`: clones `to` and `lifetime_ms` unchanged, advances `age_ms` (saturating) as before. + +- [ ] **Step 1: Write the failing tests** + +Add to `observation.rs`'s `tests` module (extend the `obs` helper with `to`/`lifetime_ms`, both defaulting to `None`): + +- `to_is_part_of_identity` — same `obs(7, 0, None)` with `to = None` vs `to = Some([2; 16])`: `assert_ne!` the ids. +- `lifetime_is_not_part_of_identity` — same payload, both `to = Some([2; 16])`, `lifetime_ms = 100` vs `200`: `assert_eq!` the ids. +- `broadcast_marker_distinct_from_zero_id` — `to = None` vs `to = Some([0; 16])`: `assert_ne!` the ids. +- `aged_preserves_addressing` — `obs` with `to = Some([2; 16])`, `lifetime_ms = Some(500)`; `.aged(10)` has `age_ms` +10 and unchanged `to`/`lifetime_ms`. +- Update `id_known_vector` (broadcast, `to = None`) to expect exactly: + `[0x4a, 0x33, 0x09, 0x92, 0xf1, 0x61, 0xbe, 0xc9, 0x19, 0x7b, 0x6e, 0x96, 0x11, 0x6f, 0xa4, 0xa2]`. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cargo test -p tines-core observation` — expected: compile errors (`to`, `lifetime_ms` unknown on `Observation`) plus the changed known-vector. + +- [ ] **Step 3: Implement** the two fields on `Observation`, the `to_marker` in `id()`, `aged()`'s field clones, and add `to: None, lifetime_ms: None` to every `Observation { .. }` literal across the crate and its test helpers. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cargo test --workspace` — expected: all pass (the known-vector now matches the new hash). + +- [ ] **Step 5: Commit** + +```bash +git add tines-core/src tines-core/tests +git commit -m "feat(core): Observation gains to/lifetime_ms; identity includes to" +``` + +--- + +### Task 2: `Store::for_each_broadcast_id` + +**Files:** +- Modify: `tines-core/src/store.rs` (trait method), `tines-core/src/mem_store.rs` (impl) +- Test: `tines-core/src/mem_store.rs` + +**Interfaces:** +- Produces: `fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId))` on `Store` — visits the ids of observations with `to == None`, in no particular order. + +- [ ] **Step 1: Write the failing test** + +`for_each_broadcast_id_skips_addressed` — insert one broadcast obs (`to = None`) and one addressed obs (`to = Some([2; 16])`); collect `for_each_broadcast_id` ids; `assert_eq!` the set to only the broadcast id. (Review Focus 5 lives here too: `trim_max_count_drops_oldest_addressed` — insert 10 addressed obs and one broadcast, trim `max_count = Some(4)`, assert the broadcast obs survives and only 4 remain.) + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cargo test -p tines-core mem_store` — expected: `no method named for_each_broadcast_id`. + +- [ ] **Step 3: Implement** the trait method and the `MemStore` impl (iterate the `BTreeMap` values, skip `held.obs.to.is_some()`). + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cargo test -p tines-core mem_store` — expected: pass. + +- [ ] **Step 5: Commit** + +```bash +git add tines-core/src/store.rs tines-core/src/mem_store.rs +git commit -m "feat(core): Store::for_each_broadcast_id" +``` + +--- + +### Task 3: Summaries carry only broadcast observations + +**Files:** +- Modify: `tines-core/src/exchange.rs` +- Test: `tines-core/src/exchange.rs` + +**Interfaces:** +- Consumes: `Store::for_each_broadcast_id` (Task 2). + +- [ ] **Step 1: Write the failing test** + +`addressed_obs_not_in_summary` — a store holding one broadcast obs `b` and one addressed obs `a` (`to = Some([2; 16])`); `Contact(1)`, then `next_frame` returns a `Summary`; decode it and assert `Bloom::from_parts(salt, &bits)` has `might_contain(b.id()) == true` and `might_contain(a.id()) == false`. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cargo test -p tines-core exchange::tests::addressed_obs_not_in_summary` — expected: the addressed id is still in the bloom (assert fails). + +- [ ] **Step 3: Implement** — in `next_frame`'s summary block, replace `store.for_each_id(..)` with `store.for_each_broadcast_id(..)`. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cargo test -p tines-core exchange` — expected: pass. + +- [ ] **Step 5: Commit** + +```bash +git add tines-core/src/exchange.rs +git commit -m "feat(core): summaries carry only broadcast observations" +``` + +--- + +### Task 4: Reject expired or malformed addressed messages + +**Files:** +- Modify: `tines-core/src/exchange.rs` (`on_frame`, both `Obs` and `Frag` arms) +- Test: `tines-core/src/exchange.rs` + +**Interfaces:** +- Consumes: `Observation` with `to`/`lifetime_ms`. +- Produces: a receive-side guard — an addressed observation is inserted only if it is not expired and its `to`/`lifetime_ms` satisfy the invariant. + +- [ ] **Step 1: Write the failing tests** + +- `expired_addressed_obs_is_rejected` — an `Obs` frame with `to = Some(ME)`, `lifetime_ms = Some(100)`, `age_ms = 200`: `on_event` returns `None` and `store.len() == 0`. +- `expired_addressed_frag_is_rejected` — a fragmented observation with the same expired addressing (build via `crate::frag::fragment` after setting `age_ms = 200`, `lifetime_ms = Some(100)`); feed the frags; `store.len() == 0`. (Review Focus 2) +- `addressing_invariant_violation_is_dropped` — `to = Some(ME)`, `lifetime_ms = None`: `on_event` returns `None`, `store.len() == 0`, no panic. (Review Focus 1) + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cargo test -p tines-core exchange` — expected: the expired/malformed messages are still stored (asserts fail). + +- [ ] **Step 3: Implement** — a private helper, e.g. `fn reject(obs: &Observation) -> bool { obs.to.is_some() && (obs.lifetime_ms.is_none() || obs.age_ms > obs.lifetime_ms.unwrap()) }` (the invariant check folds `lifetime_ms = None` into "reject"), called in both receive arms before `store.insert`. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cargo test -p tines-core exchange` — expected: pass. + +- [ ] **Step 5: Commit** + +```bash +git add tines-core/src/exchange.rs +git commit -m "feat(core): reject expired or malformed addressed messages" +``` + +--- + +### Task 5: Addressed store-and-forward in the exchange + +**Files:** +- Modify: `tines-core/src/exchange.rs` (forwarding + bounce state) +- Test: `tines-core/src/exchange.rs` + +**Interfaces:** +- Consumes: `Observation` with `to`/`lifetime_ms`, `Peer::id`. +- Produces: addressed forwarding in `next_frame` — when a peer's `id` is known, emit (one frame per poll, round-robin over peers and over the store's non-expired addressed observations): delivery when `obs.to == Some(peer.id)`, epidemic otherwise; skip a message's recorded source peer (bounce suppression); addressed forwarding must not starve the broadcast diff. + +- [ ] **Step 1: Write the failing tests** + +- `addressed_obs_delivered_to_recipient` — store has `to = Some([2; 16])`; `Contact(1)`, then a `Frame` from addr 1 with `from = [2; 16]` (so `peer.id = Some([2; 16])`); drain `next_frame` (skip the summary) and assert an `Obs` with `obs.to == Some([2; 16])` is emitted to addr 1. +- `addressed_obs_epidemic_forwards_to_non_recipient` — same setup but `to = Some([3; 16])`, peer id `[2; 16]`; assert an `Obs` with `to = Some([3; 16])` is emitted to addr 1 (Review Focus 3: the unknown recipient never drops the message). +- `addressed_obs_does_not_bounce_to_source` — store is empty; `on_event(Frame(addr 1, Obs { from: [2; 16], obs: to = Some([3; 16]) }))`; then `next_frame` emits no `Obs` addressed to `[3; 16]` back to addr 1 (only a summary). (Review Focus 4) + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cargo test -p tines-core exchange` — expected: no addressed frames are emitted (asserts fail). + +- [ ] **Step 3: Implement** — in `next_frame`, after the summary priority, add addressed forwarding: for the round-robin peer, enumerate the store's non-expired addressed observations and emit one (delivery before epidemic). Record the source addr of each inbound addressed message (e.g. a bounded `heapless::Vec<(ObservationId, A), SEND_QUEUE_CAP>` on `Exchange`) and skip re-forwarding an id to its recorded source. Interleave with the existing broadcast `emit_one` so broadcast work still runs. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cargo test -p tines-core exchange` — expected: pass. + +- [ ] **Step 5: Commit** + +```bash +git add tines-core/src/exchange.rs +git commit -m "feat(core): addressed store-and-forward in the exchange" +``` + +--- + +### Task 6: `Node::observe_to` and the app-facing view + +**Files:** +- Modify: `tines-core/src/node.rs` +- Test: `tines-core/tests/pair.rs` + +**Interfaces:** +- Produces: `pub fn observe_to(&mut self, to: NodeId, lifetime_ms: u64, topic: &[u8], payload: &[u8], observed_at: Option, now: Tick) -> Result` — creates an observation with `to = Some(to)`, `lifetime_ms = Some(lifetime_ms)`, `age_ms = 0`, incremented `next_seq`. +- `Node::observe` remains broadcast-only (`to = None`, `lifetime_ms = None`). +- `Node::belief` skips `to = Some(other != me)` observations (app view = broadcast + to-me). + +- [ ] **Step 1: Write the failing tests** (in `pair.rs`) + +- `observe_to_creates_addressed_message` — `node.observe_to([2; 16], 5000, b"task", b"go", None, 0)` returns an id; the stored observation has `to == Some([2; 16])` and `lifetime_ms == Some(5000)`. +- `observe_to_rejects_oversized_topic` — 65-byte topic → `Err(ObserveError::TooLarge)`. +- `belief_excludes_addressed_to_others` — store a broadcast obs and an addressed-to-other obs under the same topic prefix; `belief(prefix, &LatestWins, now)` returns the broadcast obs's payload (not the addressed one). + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cargo test -p tines-core --test pair` — expected: `no method named observe_to`. + +- [ ] **Step 3: Implement** `observe_to`, and add the `to = Some(other != me)` guard to `Node::belief`'s closure. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cargo test -p tines-core --test pair` — expected: pass. + +- [ ] **Step 5: Commit** + +```bash +git add tines-core/src/node.rs tines-core/tests/pair.rs +git commit -m "feat(core): Node::observe_to and belief excludes addressed-to-others" +``` + +--- + +### Task 7: Simulator relay scenario, property tests, convergence scoping + +**Files:** +- Modify: `tines-sim/src/sim.rs` (add `Sim::observe_to`), `tines-sim/tests/props.rs` (scope + new properties) +- Create: `tines-sim/tests/relay.rs` + +**Interfaces:** +- Consumes: `Node::observe_to` (Task 6), `SimWorld`/`SimTransport`. +- Produces: `pub fn observe_to(&mut self, node: usize, to: NodeId, lifetime_ms: u64, topic: &[u8], payload: &[u8]) -> ObservationId` on `Sim`. + +- [ ] **Step 1: Write the failing tests** + +`relay.rs` — `relayed_task_reaches_recipient`: three nodes, contacts `(0,1)` then `(1,2)`; node 0 `observe_to([2; 16], 60_000, b"task", b"investigate")`; `run_until(trace.end())`; assert `sim.nodes[2].store().contains(&id)`. Second test: `carrier_drops_expired` — a 10 ms lifetime with the same relay; after running well past expiry, assert node 2's store does not contain the id. + +`props.rs` — add `addressed_reaches_recipient_along_connected_path` (a 2..=6-node connected trace; one node tasks the max-id node; assert that node's store eventually contains the id). Do **not** assert `converged` for addressed scenarios (addressed messages diverge by design); the existing `connected_components_converge` and `store_is_independent_of_contact_order` keep using broadcast-only `observe` and are unchanged. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cargo test -p tines-sim` — expected: `no method named observe_to` on `Sim`. + +- [ ] **Step 3: Implement** `Sim::observe_to`, and the new tests' setup. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cargo test --workspace` — expected: all pass (including the pre-existing 85+ tests). + +- [ ] **Step 5: Commit** + +```bash +git add tines-sim +git commit -m "test(sim): addressed relay scenario and property tests" +``` + +--- + +## Final Gate + +After Task 7, run `sh scripts/ci.sh` (fmt, clippy `-D warnings`, doc, test, no_std) — expected: exit 0.