diff --git a/Cargo.lock b/Cargo.lock
index 47912b1..d1e1c1e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -937,6 +937,7 @@ dependencies = [
 name = "datastream"
 version = "0.1.0"
 dependencies = [
+ "crossbeam-channel",
  "iroh",
  "libc",
  "serde",
@@ -2104,6 +2105,7 @@ dependencies = [
 name = "iroh-driver"
 version = "0.1.0"
 dependencies = [
+ "crossbeam-channel",
  "datastream",
  "distribution",
  "iroh",
@@ -4342,6 +4344,7 @@ name = "swactor-process"
 version = "0.1.0"
 dependencies = [
  "crossbeam-queue",
+ "datastream",
  "libc",
  "proptest",
  "proptest-state-machine",
diff --git a/crates/dashboard/src/lib.rs b/crates/dashboard/src/lib.rs
index 6d9ecd5..bd5a37f 100644
--- a/crates/dashboard/src/lib.rs
+++ b/crates/dashboard/src/lib.rs
@@ -8,7 +8,6 @@ pub mod view;
 use std::sync::Arc;
 
 use datastream::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId};
-use parking_lot::Mutex;
 use serde::Serialize;
 use tokio::sync::broadcast;
 
@@ -86,10 +85,32 @@ pub struct DashboardHandle {
     store: Arc<DashboardStore>,
     views: Arc<ViewRegistry>,
     shutdown_notify: Arc<tokio::sync::Notify>,
-    standalone_rt: Mutex<Option<tokio::runtime::Runtime>>,
 }
 
 impl DashboardHandle {
+    /// Create the datastream dashboard state.
+    ///
+    /// The HTTP server is not started until `start_http` or `start_http_standalone`
+    /// is called.
+    pub fn new(config: DashboardConfig) -> Self {
+        let views = Arc::new(ViewRegistry::new());
+        views.register(Arc::new(live_explorer::LiveDatastreamExplorer::default()));
+        views.register(Arc::new(hardware_view::HardwareDashboardView::default()));
+        views.register(swactor::worker_view());
+        let store = Arc::new(DashboardStore::new(
+            config.raw_frame_history,
+            Arc::clone(&views),
+        ));
+        let (frames, _) = broadcast::channel(config.frame_buffer.max(1));
+        Self {
+            port: config.port,
+            frames,
+            store,
+            views,
+            shutdown_notify: Arc::new(tokio::sync::Notify::new()),
+        }
+    }
+
     /// Register a read-only view. External crates can keep their interpretation
     /// code beside their component and plug it into this registry.
     pub fn register_view(&self, view: Arc<dyn DashboardView>) {
@@ -113,8 +134,8 @@ impl DashboardHandle {
         self.shutdown_notify.notify_waiters();
     }
 
-    /// Start the HTTP server on an existing Tokio runtime.
-    pub fn start_http(&self, handle: tokio::runtime::Handle) {
+    /// Build the HTTP server future for an embedding runtime to poll directly.
+    pub fn http_server(&self) -> impl Future<Output = ()> + Send + 'static {
         let state = server::AppState {
             frames: self.frames.clone(),
             store: Arc::clone(&self.store),
@@ -122,44 +143,13 @@ impl DashboardHandle {
             shutdown_notify: Arc::clone(&self.shutdown_notify),
         };
         let port = self.port;
-        handle.spawn(async move {
+        async move {
             server::run_server(state, port).await;
-        });
+        }
     }
 
-    /// Start the HTTP server on a standalone Tokio runtime.
-    pub fn start_http_standalone(&self) {
-        let rt = tokio::runtime::Builder::new_multi_thread()
-            .worker_threads(1)
-            .enable_all()
-            .build()
-            .expect("failed to create tokio runtime for dashboard HTTP");
-        let handle = rt.handle().clone();
-        *self.standalone_rt.lock() = Some(rt);
-        self.start_http(handle);
-    }
-}
-
-/// Create the datastream dashboard state.
-///
-/// The HTTP server is not started until `start_http` or `start_http_standalone`
-/// is called.
-pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
-    let views = Arc::new(ViewRegistry::new());
-    views.register(Arc::new(live_explorer::LiveDatastreamExplorer::default()));
-    views.register(Arc::new(hardware_view::HardwareDashboardView::default()));
-    views.register(swactor::worker_view());
-    let store = Arc::new(DashboardStore::new(
-        config.raw_frame_history,
-        Arc::clone(&views),
-    ));
-    let (frames, _) = broadcast::channel(config.frame_buffer.max(1));
-    DashboardHandle {
-        port: config.port,
-        frames,
-        store,
-        views,
-        shutdown_notify: Arc::new(tokio::sync::Notify::new()),
-        standalone_rt: Mutex::new(None),
+    /// Spawn the HTTP server on an existing Tokio runtime and return its task handle.
+    pub fn spawn_http(&self, handle: &tokio::runtime::Handle) -> tokio::task::JoinHandle<()> {
+        handle.spawn(self.http_server())
     }
 }
diff --git a/crates/datastream/Cargo.toml b/crates/datastream/Cargo.toml
index 06198e8..c6510c1 100644
--- a/crates/datastream/Cargo.toml
+++ b/crates/datastream/Cargo.toml
@@ -9,6 +9,7 @@ swactor-transport = { path = "../transport" }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 iroh = "0.98"
+crossbeam-channel = "0.5"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"
diff --git a/crates/datastream/DATASTREAM_SPEC.md b/crates/datastream/DATASTREAM_SPEC.md
index 3e1b9c2..35266b8 100644
--- a/crates/datastream/DATASTREAM_SPEC.md
+++ b/crates/datastream/DATASTREAM_SPEC.md
@@ -1,462 +1,738 @@
 # The Datastream — Specification
 
->>**STALE** Stale and not the final advisor on current or future state of the datastream.
-
-> **Status.** This is the canonical specification for the `datastream` crate.
-> It supersedes the implicit spec the old `crates/distribution/src/datastream`
-> doc-comments referred to (`DATASTREAM_SPEC.md §4.1`, etc.). Those section
-> numbers are not preserved; updating the stale `(spec §…)` references is part
-> of the migration this document drives.
+> **Status.** Current-state specification for the `datastream` crate. This keeps
+> the original outline, but updates the model to the implementation that now
+> exists: stream-local numeric channel ids, a channel catalog, broadcast endpoint
+> subscriptions, and catalog-aware transport events.
 
 ---
 
 ## 1. What the datastream is
 
-The datastream is a **per-node, append-only telemetry pipe**. Every node
-produces exactly one stream of its own observations; one consumer reconstructs
-those streams and reads them.
+The datastream is a **per-node, append-only telemetry pipe**. Every stream is
+produced by one node incarnation, identified by `(node, life)`. A node can feed
+zero, one, or many local/remote consumers: the mux is drained once, then the
+endpoint fans out the resulting events to subscribers.
 
-Its entire value comes from one rule:
+Its core rule is unchanged:
 
-> **Nothing between a producer and a view ever interprets a payload.**
+> **Nothing between a producer and a view ever interprets a producer payload.**
 
-Producers tag bytes with a channel name and hand them off. A single per-node
-**mux** stamps each with a position and interleaves them into one ordered
-stream. A best-effort **transport** carries the stream. **Ingest** reconstructs
-each node's stream by position into a **store**. **Views** decode bytes back
-into meaning — and only here, at read time, does anything look inside a payload.
+The current implementation is catalog-aware. Producers register a channel name
+and content class with their stream owner, receive a stream-local numeric
+`ChannelId`, and submit opaque payload bytes on that id. A single per-stream
+**mux** queues accepted payloads and stamps each drained frame with a position,
+interleaving every channel into one ordered stream. A **catalog** maps numeric
+channel ids back to human-readable names and decode/display metadata.
+A best-effort **transport**
+carries catalog declarations and frame events. **Ingest** reconstructs streams
+by `StreamId` and position into a **store**. **Views** resolve channel names
+from metadata and decode bytes back into meaning — and only here, at read time,
+does anything look inside a producer payload.
 
 ```text
-   producers            tag bytes with a channel
+   producers                  register names, submit opaque bytes by ChannelId
         │
         ▼
-   per-node MUX          stamp position, interleave into one ordered stream
+   endpoint / catalog          allocate ids, declare stream/channel metadata
         │
         ▼
-   transport             best-effort: may drop / reorder / delay, never corrupt
+   per-stream MUX              queue payloads, assign positions on drain
         │
         ▼
-   ingest                reconstruct each node's stream by position
+   endpoint fanout             drain once, broadcast catalog-aware events
         │
         ▼
-   store (the truth)     whole, append-only, position-keyed
+   transport                   best-effort: may drop / reorder / delay / duplicate
         │
         ▼
-   views                 read-time projections — the only place bytes are decoded
+   ingest                      reconstruct each stream by position
+        │
+        ▼
+   store (truth for frames)    whole, append-only, position-keyed
+        │
+        ▼
+   views                       read-time projections; payload decoding lives here
 ```
 
-This collapses the usual telemetry zoo — counters, gauges, histograms, logs,
-events, traces — into **one** thing: positioned bytes on a named channel.
-Collection, transport, and storage are uniform because none of them know which
-of those a frame "is." That distinction does not exist in the pipe. It exists,
-if at all, in a view: "the latest frame on this channel" is a gauge, "every
-frame on this channel" is an event log, and they are the same bytes read two
-ways.
+This still collapses counters, gauges, histograms, logs, events, traces, and
+binary blobs into one mechanism: positioned bytes on named channels. The pipe
+stores and moves frames uniformly because it does not know what a payload
+"means." That distinction exists in a view: "the last frame on this channel" is
+a gauge, "every frame on this channel" is an event log, and both are projections
+over the same stored frames.
 
-### 1.1 Consequences that the rest of this spec just spells out
+### 1.1 Consequences that the rest of this spec spells out
 
-- There is **one way data enters** (§4). No per-channel "kind of producer."
-- A channel is **a name, not a resource** (§5). Nothing to allocate or free.
-- All **semantics live in views** (§8). The pipe is mechanism only.
+- There is **one data-entry shape** (§4): register or reuse a channel id, then
+  submit bytes on that id.
+- A channel name is not a queue, counter, or buffer, but it **is cataloged**
+  (§5): the stream owner allocates a stream-local numeric id and declares its
+  name/content metadata.
+- The **store is frame truth**, not catalog truth (§7). Durable name recovery
+  requires catalog descriptors alongside numeric frames.
+- All **producer-payload semantics live in views** (§8). Catalog content classes
+  help route and display; they do not let mux, transport, ingest, or store
+  inspect producer payloads.
 
 ---
 
 ## 2. The data model
 
-### 2.1 The frame is the only unit
+### 2.1 The frame remains the unit of ordered data
 
 ```rust
 pub struct Frame {
-    pub channel: ChannelId,   // the named lane these bytes belong to
-    pub position: Position,   // the mux-assigned order within the node's stream
+    pub channel: ChannelId,   // stream-local numeric lane
+    pub position: Position,   // mux-assigned order within the stream
     pub payload: Vec<u8>,     // opaque bytes — never interpreted by the pipe
 }
 ```
 
-A typed record, a log line, and a binary blob are the same kind of thing here:
-bytes on a channel. The `payload` is opaque to the mux, the transport, and the
-store.
+A typed record, a log line, and a binary blob are all frames: bytes on a
+stream-local channel id at a stream-local position. `payload` is opaque to the
+mux, transport, ingest, and store.
 
-There is **no per-frame wall-clock timestamp.** Frames are ordered and
-correlated by position alone. A producer that wants wall-clock time puts it
-*inside* the payload, as a field of its record — it is data, not a property of
-the pipe.
+There is still **no timestamp field on `Frame`**. Time, wall-clock correlation,
+latency, or tracing data is producer payload, not a datastream-owned sidecar or
+property on the frame envelope.
 
-### 2.2 The coordinate
-
-Every frame in the system has exactly one address:
-
-```text
-Frame  @  (node, life, channel, position)
-            └── stream ──┘  source   order
-```
-
-| Part       | Type         | Answers          |
-|------------|--------------|------------------|
-| `node`     | `NodeId`     | *who* produced it |
-| `life`     | `Lifetime`   | *which incarnation* of that node |
-| `channel`  | `ChannelId`  | *which source* within that node |
-| `position` | `Position`   | *where in order* within that stream |
-
-`(node, life)` together are the **stream**; `channel` is the **source** within
-it; `position` is the **order**. This four-tuple locates any datum the system
-has ever produced. Addressing (§3) is built entirely on it.
-
-### 2.3 Position: per-stream, monotonic, gap-free
-
-```rust
-pub struct Position(pub u64);
-```
-
-- **One sequence per stream**, not per channel. A node's mux holds a single
-  counter shared across every channel, so a given channel's frames carry
-  *non-contiguous* positions — a sparse projection of the node's one global
-  sequence, interleaved with every other channel.
-- **Monotonic and gap-free in assignment.** The mux never reuses a position and
-  never skips one when numbering. A position that is assigned but never
-  delivered surfaces downstream as a *missing* position — a detectable gap
-  (§4.4, §6.3).
-- **Not comparable across streams.** Positions order frames within one
-  `(node, life)` only. There is no global clock.
-
-**Why one counter and not one per channel.** It keeps the mux a single position
-authority and makes gap detection a whole-stream property: a hole means
-*something* was lost. The cost is that a drop **cannot be attributed to a
-specific channel** — you know a frame is missing, not which channel it carried.
-That is the right trade for telemetry.
-
-> **Decision of record — drop attribution.** If a particular source ever needs
-> guaranteed contiguous accounting, it
-> carries its *own* sequence number as a field in its record. The pipe's
-> position stays global and dumb; domain counting is domain data.
-
-### 2.4 Stream identity — incarnations never merge
+### 2.2 Stream identity — incarnations never merge
 
 ```rust
 pub struct StreamId { pub node: NodeId, pub life: Lifetime }
 ```
 
 `StreamId` is the ingest key. Two streams with the same `node` but different
-`life` are **different streams and must never merge.** A node that dies and is
-restarted (re-rented, re-scheduled) begins a new `life`, so its fresh stream
-does not append to — or collide with — its prior one. Restart is visible, not
-silently glued over.
+`life` are **different streams and must never merge**. A node that dies and is
+restarted (re-rented, re-scheduled, or bootstrapped for a new run) begins a new
+`life`, so the fresh stream does not append to or collide with the prior one.
+Restart is visible, not silently glued over.
+
+The stream descriptor declares where a stream came from:
+
+```rust
+pub enum StreamOrigin { Orchestrator, Bootstrap, RemoteNode }
+
+pub struct StreamDescriptor {
+    pub stream: StreamId,
+    pub label: Option<String>,
+    pub origin: StreamOrigin,
+}
+```
+
+`StreamOrigin` is subscription metadata. It is not a clock and does not order
+streams.
+
+### 2.3 Position: per-stream, monotonic when assigned
+
+```rust
+pub struct Position(pub u64);
+```
+
+- **One sequence per stream**, not per channel. A stream's mux holds a single
+  counter shared across every channel. A single channel therefore has sparse,
+  non-contiguous positions interleaved with every other channel.
+- **Assigned while draining.** `submit` only attempts to enqueue payload bytes.
+  `drain` assigns positions to accepted payloads. A queue-full rejection happens
+  before position assignment and therefore does not create a position gap.
+- **Not comparable across streams.** Positions order frames within one
+  `StreamId` only. There is no global stream order.
+
+**Why one counter and not one per channel.** It keeps the mux a single position
+authority once frames leave the queue and makes assigned-frame loss detection a
+whole-stream property: a hole means *something already assigned* was lost. A
+source needing contiguous domain accounting carries its own sequence number
+inside its payload.
+
+### 2.4 Channels: numeric ids plus catalog descriptors
+
+A raw frame contains a numeric channel id:
+
+```rust
+pub struct ChannelId(pub u32);
+```
+
+`ChannelId` values are **stream-local**. `ChannelId(7)` in one stream is not the
+same channel as `ChannelId(7)` in another stream unless their descriptors say
+so. The current endpoint allocates user channels starting at `ChannelId(1)`;
+`ChannelId(0)` is not allocated by the public registration path.
+
+The human-readable channel name and display/routing metadata live in a channel
+descriptor:
+
+```rust
+pub enum ChannelContentKind { Bytes, TextStream, JsonRecord }
+
+pub enum ChannelContent {
+    Bytes,
+    TextStream,
+    JsonRecord { schema: Option<String> },
+}
+
+pub struct ChannelDescriptor {
+    pub stream: StreamId,
+    pub id: ChannelId,
+    pub name: String,
+    pub label: Option<String>,
+    pub content: ChannelContent,
+}
+
+pub struct ChannelRef {
+    pub stream: StreamId,
+    pub channel: ChannelId,
+}
+```
+
+The globally resolved raw channel identity is `ChannelRef`, not a bare
+`ChannelId`. A view usually needs the `ChannelDescriptor` for that `ChannelRef`
+to recover the channel name and choose a decode/display path.
+
+### 2.5 Datastream events
+
+The live endpoint/subscription path carries catalog-aware events:
+
+```rust
+pub struct FrameDelivery {
+    pub channel: ChannelRef,
+    pub position: Position,
+    pub payload: Vec<u8>,
+}
+
+pub enum DatastreamEvent {
+    StreamDeclared(StreamDescriptor),
+    ChannelDeclared(ChannelDescriptor),
+    Frame(FrameDelivery),
+    StreamEnded(StreamId),
+}
+```
+
+`FrameDelivery` is the live-event form of a frame: stream and channel are
+resolved into a `ChannelRef`, then position and payload follow. The older
+`Delivery { stream, frame }` shape still exists for ingest, store tests, and
+legacy adapters (§6.1, §7.1).
+
+`StreamDeclared` is part of the event vocabulary, but current QUIC transport
+puts the stream descriptor in the stream header and does not emit a separate
+`StreamDeclared` event. `StreamEnded` is also part of the event vocabulary, but
+current `DatastreamEndpoint` exposes no public `end_stream` method; terminal
+source records remain a producer convention until a stream-ending API is added
+(§5.7).
+
 
 ---
 
 ## 3. Addressing
 
-### 3.1 The address is the coordinate
+### 3.1 Raw and resolved coordinates
 
-A frame is addressed by `(node, life, channel, position)`. A *source* is
-addressed by the prefix `(node, life, channel)` — drop `position` and you are
-naming a lane rather than a single datum. A *node's whole stream* is
-`(node, life)`. Nothing else is needed; there is no separate registry of
-producer identities.
+A stored raw frame is addressed by:
 
-### 3.2 Channel paths: opaque on the wire, structured at the edges
-
-```rust
-pub struct ChannelId(Arc<str>);   // an opaque token to the pipe
+```text
+raw-frame @ (stream, numeric-channel, position)
+             └ node/life ┘  ChannelId       order
 ```
 
-To the mux, transport, ingest, and store a `ChannelId` is an uninterpreted
-string. They never split it, match it, or validate it. This is what keeps the
-pipe dumb and lets a brand-new channel flow end to end with zero pipe changes.
+A named channel source is addressed by resolving that raw channel through the
+catalog:
 
-Its **structure is a read-side convention** — known only to producers (to mint
-paths they own) and to consumers/classifiers (to match and decode them). It
-is therefore *both*: a flat string on the wire, a structured path at the edges.
-There is no conflict, because the two views never meet inside the pipe.
+```text
+source @ (stream, channel-name)
+          └ node/life ┘ descriptor.name
+```
 
-### 3.3 The path grammar
+A payload datum is therefore either:
 
-A channel path is a dotted sequence of segments:
+- raw: `(StreamId, ChannelId, Position)`, sufficient for storage and ordering;
+- resolved: `(StreamId, ChannelDescriptor.name, Position)`, required for
+  human-facing selection and decoding.
+
+No bare `ChannelId` is globally meaningful. A consumer that receives a frame
+before its descriptor can store it raw, but cannot render a stable name until the
+catalog descriptor arrives or is recovered from durable catalog state.
+
+### 3.2 Channel names: structured catalog paths, not wire ids
+
+Channel names remain dotted paths, but they are catalog descriptor fields rather
+than the frame's wire identity. The mux, ingest, and store do not split names;
+subscription matching and views may match names through descriptors.
+
+A channel name is a dotted sequence of segments:
 
 ```text
 channel := namespace ( "." qualifier )*
-namespace := the owning subsystem        e.g. host, transport, swim, runtime,
-                                              dist, proc, identity
-qualifier := instance-key | leaf         instance-key identifies *which* of a
-                                              dynamic source; leaf names the signal
+namespace := owning subsystem          e.g. datastream, host, runtime, proc,
+                                            mvp, transport, dist, identity
+qualifier := instance-key | leaf       instance-key identifies a dynamic source;
+                                            leaf names the signal
 ```
 
-Examples, current and proposed:
+Current names and families seen in code include:
 
-| Path                          | Namespace   | Instance      | Leaf       |
-|-------------------------------|-------------|---------------|------------|
-| `identity`                    | identity    | —             | —          |
-| `host.resource`               | host        | —             | resource   |
-| `transport.internals`         | transport   | —             | internals  |
-| `proc.trainer.stdout`         | proc        | `trainer`     | stdout     |
-| `transport.peer.<id>.conn` *(proposed)* | transport | `peer/<id>` | conn |
+| Path or family                                      | Owner / meaning |
+|-----------------------------------------------------|-----------------|
+| `datastream.health`                                 | datastream self-health record |
+| `host.cpu`, `host.gpu`, `host.net`                  | host hardware samples |
+| `runtime.actors`                                    | swactor runtime actor stats |
+| `proc.<label>.stdout`, `proc.<label>.stderr`        | managed process output streams |
+| `mvp.lifecycle`                                     | MVP lifecycle record |
+| `mvp.provisioning.events`                           | MVP provisioning events |
+| `mvp.provisioning.logs.node.<id>.<stdout|stderr|provider>` | MVP provisioning logs |
 
-A single-segment path (`identity`, `membership`) is the degenerate case: a
-namespace with one global signal and no multiplicity. Instance segments appear
-only where a source has many instances (one per process, per peer, per actor).
+Single-segment names are allowed, but most current code uses namespaced paths.
+Instance segments appear only where a source has many instances: a process
+label, node id, actor id, peer id, run id, or similar domain-owned key.
 
-### 3.4 Source selectors
+### 3.3 Namespacing and ownership
 
-A consumer addresses sources by a **stream scope plus a channel prefix**:
+Collisions are prevented by namespace ownership plus catalog conflict checks:
 
-```text
-selector := (node?, life?)  channel-prefix [ ".*" ]
+- Each subsystem owns its first path segment. Only that subsystem should mint
+  names below it.
+- Dynamic instance segments must be stable identifiers already unique in that
+  subsystem's domain.
+- Within one stream catalog, registering the same name with the same content
+  returns the existing id; registering the same name with conflicting content is
+  an error.
+- Two streams may allocate different numeric ids for the same channel name.
+
+### 3.4 Source and channel selectors
+
+Subscriptions use structured filters rather than a single textual selector:
+
+```rust
+pub struct SubscriptionRequest {
+    pub sources: SourceFilter,
+    pub channels: ChannelFilter,
+}
+
+pub enum SourceFilter {
+    All,
+    Origin(StreamOrigin),
+    Node(NodeId),
+    Stream(StreamId),
+}
+
+pub enum ChannelFilter {
+    All,
+    Name(String),
+    Prefix(String),
+    Content(ChannelContentKind),
+}
 ```
 
-- `(*, *)  membership`            — every node's membership signal.
-- `(X, *)  proc.trainer.*`        — everything the trainer process on node `X`
-                                     ever emitted, across that node's restarts.
-- `(X, Y)  host.resource`         — one exact source.
-- `(*, *)  proc.*`                — all process output, fleet-wide.
+`SourceFilter::Node(node)` means every known life of that node. `Stream(stream)`
+means exactly one `(node, life)`. `Origin(origin)` matches stream descriptor
+metadata. `ChannelFilter::Prefix(prefix)` is a plain string `starts_with` over
+`ChannelDescriptor.name`; the old `.*` notation is user-interface sugar, not the
+internal API. `ChannelFilter::Content(kind)` matches catalog content metadata,
+not payload inspection.
 
-Prefix matching over the path **is** the query model. Because instances live in
-the path, "all processes" or "one peer's view" is a prefix, never a side table
-of what exists. This is the whole reason the path is structured.
+Examples:
+
+| Request | Meaning |
+|---------|---------|
+| `{ sources: All, channels: Prefix("proc.") }` | all process-output channels with descriptors visible to the subscriber |
+| `{ sources: Node(X), channels: Prefix("proc.trainer.") }` | trainer process output from every life of node `X` |
+| `{ sources: Stream(S), channels: Name("host.cpu") }` | exact host CPU channel in one stream |
+| `{ sources: Origin(Bootstrap), channels: Content(JsonRecord) }` | JSON-record channels from bootstrap streams |
 
 ---
 
 ## 4. Population — the one way in
 
-### 4.1 A producer holds a submission handle
+### 4.1 The endpoint owns stream-local allocation and fanout
 
-There is exactly one means of population. A producer holds a **submission
-handle** to its node's mux and calls:
+The current public owner of a stream is `DatastreamEndpoint`:
 
 ```rust
-fn submit(&self, channel: impl Into<ChannelId>, payload: Vec<u8>);
+pub struct DatastreamEndpoint { /* stream, mux, catalog, fanout, counters */ }
 ```
 
-Everything else is sugar over this. A typed record encodes itself and submits
-(`record.emit_to(&handle)`); a text source submits a line; a binary source
-submits bytes. There is no second mechanism, no "event vs sample vs one-shot"
-path. *When* a producer calls `submit` — on a timer, in a callback, once at boot
-— is the producer's own business and is invisible to the stream.
+An endpoint owns:
 
-The handle is cheaply cloneable and `Send`, so any number of producers, on any
-threads, feed the same mux. The mux serializes them into one order (§4.3).
+- the stream id;
+- the mux that assigns positions;
+- the catalog that maps names to stream-local channel ids;
+- local subscriber fanout;
+- counters for assigned, drained, mux-dropped, and bitbucketed frames.
 
-### 4.2 One observer per subsystem
-
-Every subsystem that produces telemetry exposes **one observer trait**, and the
-node wires that observer to a submission handle. This is the uniform "means of
-population" applied across the codebase:
+Code that emits telemetry normally asks the endpoint for a cloneable
+`DatastreamProducer`:
 
 ```rust
-// e.g. process output
-trait ProcessOutputObserver { fn on_output(&self, label: &str, is_stderr: bool, data: &[u8]); }
-// the node installs an observer whose body is `handle.submit(...)`
+let producer = endpoint.producer();
 ```
 
-This replaces today's inconsistent wiring (some subsystems install a sink, some
-install `None`, some route through the node as an adapter). The rule: **a
-subsystem emits by handing its observer a submission handle — nothing else.**
-Existing observer-shaped seams such as `ProcessOutputObserver` and the unused
-`SwimObserver` conform to this one shape.
+Legacy `DatastreamEmitter` / `DatastreamEventSink` APIs still exist for old
+call sites. New code should register channels on `DatastreamEndpoint` or
+`DatastreamProducer` and submit through `DatastreamProducer`.
 
-### 4.3 The mux: single position authority
+### 4.2 Register or reuse a channel, then submit bytes
+
+A producer registers a channel name and content descriptor, receiving a numeric
+`ChannelId`:
 
 ```rust
-pub struct Mux { stream: StreamId, next: AtomicU64, buffer: Mutex<VecDeque<Frame>> }
+fn register_channel(name: impl Into<String>, content: ChannelContent) -> ChannelId;
+fn try_register_channel(
+    name: impl Into<String>,
+    content: ChannelContent,
+) -> Result<ChannelId, ChannelRegistrationError>;
+fn register_record<R: Record>() -> ChannelId;
 ```
 
-One mux per node. It is the sole assigner of positions and the outgoing buffer:
+Then it submits payload bytes on that id:
 
-1. On `submit`, it atomically takes the next position **before** buffering the
-   frame. Positions are consumed in submission order and never reused.
-2. The frame goes into a bounded buffer.
-3. The transport `drain`s the buffer each tick, emptying it in insertion order.
+```rust
+fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> bool;
+fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> bool;
+fn submit_text_owned(&self, channel: ChannelId, text: String) -> bool;
+fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> bool;
+```
 
-Because the position is taken before buffering, the numbering is gap-free even
-when the buffer is not.
+Everything entering the stream goes through the same shape: numeric channel id
+and payload bytes. The submit APIs return `true` when the payload entered the
+mux queue and `false` when it was dropped before position assignment. Text and
+typed records are API sugar over bytes. *When* a producer emits — timer,
+callback, process output, once at boot — is the producer's business.
 
-### 4.4 Drops are gaps (the back-pressure policy)
+### 4.3 Observers and subsystem adapters
 
-The buffer is **bounded**. On overflow the mux **drops the frame but keeps its
-position consumed.** The producer never blocks. Downstream, the dropped position
-is simply missing — a detectable gap, never a silent renumber.
+Subsystems should expose observer/hook seams that can be wired to a producer.
+The current endpoint includes adapters for process output and runtime stats:
 
-> **Decision of record — lossy, not blocking.** Telemetry must never apply
-> back-pressure to the work it observes. A datastream is allowed to lose frames;
-> it is not allowed to stall a producer or to hide that a loss happened. Anything
-> that cannot tolerate loss does not belong on the datastream (§9).
+```rust
+trait ProcessOutputObserver {
+    fn on_output(&self, label: &str, is_stderr: bool, data: &[u8]);
+}
+
+trait StatsHook {
+    fn on_tick(&self, worker_id: usize, snapshots: &[ActorSnapshot]);
+}
+```
+
+Process output uses a caller-provided closure from `(label, is_stderr)` to an
+already registered `ChannelId`. Runtime stats default to `runtime.actors` as a
+JSON record channel. Other subsystems follow the same rule: register or reuse a
+channel id, encode their own payload, submit bytes.
+
+### 4.4 The mux: single position authority
+
+```rust
+struct PendingFrame {
+    channel: ChannelId,
+    payload: Vec<u8>,
+}
+
+pub struct Mux {
+    stream: StreamId,
+    next: AtomicU64,
+    dropped: AtomicU64,
+    tx: crossbeam_channel::Sender<PendingFrame>,
+    rx: crossbeam_channel::Receiver<PendingFrame>,
+}
+```
+
+The receiver is stored directly; no outer receiver mutex is used for `drain()`.
+
+One mux owns one stream's position sequence. On `submit`:
+
+1. The mux attempts a nonblocking send of `PendingFrame { channel, payload }`
+   into a bounded queue.
+2. If the send succeeds, `submit` returns `true`.
+3. If the queue is full or disconnected, the mux increments `dropped` and
+   returns `false`; no position has been consumed.
+4. `drain` takes all currently queued pending frames, assigns each one the next
+   stream position, and returns frames in queue-drain order.
+
+The queue capacity is bounded and clamped by the implementation. Position is the
+canonical store/view ordering key after drain; the drain/fanout batch itself is
+not specified as a sorted-position replay guarantee.
+
+### 4.5 Queue drops are pre-position; producers do not block
+
+Telemetry must not apply back-pressure to the work it observes. A full mux queue
+causes `try_send` failure; the producer does not block and no position is
+assigned to the rejected payload.
+
+The `dropped` mux counter counts submissions lost to mux overflow or
+receiver-disconnect before they enter the queue. Once a frame has been drained
+and assigned a position, later transport or store loss can still surface as an
+interior gap if bracketing positions arrive.
+
+> **Decision of record — lossy, not blocking.** A datastream is allowed to lose
+> frames; it is not allowed to stall a producer or silently renumber around a
+> loss. Anything that cannot tolerate loss does not belong on the datastream as
+> its sole source of truth.
 
 ---
 
 ## 5. Channels
 
-### 5.1 Channels are names, not resources
+### 5.1 Channels are catalog entries, not independent resources
 
-This is the core of channel management, and it makes the rest of this section
-short. A channel is **a name attached to frames** — nothing more. There is:
+A channel is a cataloged name/content descriptor mapped to a stream-local
+numeric id. It is **not** a separate queue, counter, lifecycle object, task, or
+storage partition.
 
-- no channel object, no `open`/`close`,
-- no per-channel counter (positions are per-stream, §2.3),
-- no per-channel buffer (one buffer per mux, §4.3),
-- no registration call at runtime.
+There is:
 
-A channel **exists** the instant a frame bears its name, and not before.
-Everything below follows from that.
+- one mux queue per stream, not per channel;
+- one position counter per stream, not per channel;
+- a stream-local catalog mapping `name <-> ChannelId`;
+- a conflict check for duplicate names with different content metadata;
+- no per-channel close or garbage collection operation.
 
-### 5.2 Meaning is caller-owned
+A channel can be declared before its first frame. A frame can be stored with
+only its numeric channel id, but human-readable decoding requires its descriptor.
 
-`datastream` defines the vocabulary for classification, not the global list of
-classified channels:
+### 5.2 Meaning is split: catalog content vs view classifier
+
+There are two related but distinct classification layers.
+
+Catalog metadata describes how a channel is expected to be routed/displayed:
+
+```rust
+pub enum ChannelContent {
+    Bytes,
+    TextStream,
+    JsonRecord { schema: Option<String> },
+}
+```
+
+This metadata is allowed in the endpoint, subscription snapshot matcher, and
+transport catalog. It is not payload inspection.
+
+View classification remains caller-owned:
 
 ```rust
 pub enum ChannelKind { Typed, Text, Opaque }
+
 pub trait ChannelClassifier {
-    fn classify(&self, channel: &ChannelId) -> ChannelKind;
+    fn classify(&self, channel_name: &str) -> ChannelKind;
 }
 ```
 
-Producers and consumers own the records and channel constants for their domains.
-A plugin or application may compose a registry of the channels it understands,
-but the pipe never consults that registry. Adding a channel or
-teaching a view a new codec changes nothing in the mux, transport, ingest, or
-store.
+A `ChannelRegistry` can classify exact typed/text names and text prefixes, but
+unknown names default to `Opaque`. The pipe may route by `ChannelContentKind`; a
+view decides how far to decode a payload by `ChannelKind` and the caller's
+registry.
 
 ### 5.3 Static channels and dynamic families
 
-- **Static channel** — a fully literal path known at compile time
-  (`host.resource`). Defined by the crate that owns that signal.
-- **Dynamic family** — a path *template* with parameter segments
-  (`proc.{label}.stdout`). The owning crate defines the helper; concrete
-  instances are minted at runtime by binding parameters:
+- **Static channel** — a fully literal path known at compile time, such as
+  `host.cpu`, `host.net`, `datastream.health`, or `runtime.actors`.
+- **Dynamic family** — a path template with domain-owned parameter segments,
+  such as `proc.<label>.stdout` or
+  `mvp.provisioning.logs.node.<id>.<stream>`.
 
-  ```rust
-  pub fn process_output(label: &str, stream: ProcStream) -> ChannelId; // proc.<label>.<stream>
-  ```
+Dynamic helpers should return channel **names** or registration descriptors, not
+bare `ChannelId`s, unless they also have access to the endpoint/producer that
+allocates ids. Current code registers each concrete dynamic name, then submits
+on the allocated numeric id.
 
 A caller-owned classifier can know a family shape without knowing every concrete
-instance: `proc.*` may classify as `Text` while `proc.trainer.stdout` first
-appears only when the trainer emits.
+instance: for example, `proc.` may classify as text while
+`proc.trainer.stdout` first appears only when the trainer emits and registers.
 
-### 5.4 Typed / Text / Opaque, and the Opaque fallback
+### 5.4 Bytes / TextStream / JsonRecord, and the raw fallback
 
-- **Typed** — decodes to a structured record under a JSON codec (§5.5).
-- **Text** — opaque text; a view treats it as lines (`proc.*`).
-- **Opaque** — *unknown to this consumer.* Retained whole, shown as raw bytes.
+Catalog content classes are:
 
-> **Decision of record — forward-compatible fallback.** An unrecognized path is
-> `Opaque`, never an error. A newer producer may introduce a channel an older
-> consumer has never heard of; that consumer carries it, stores it, and renders
-> it as raw bytes rather than rejecting it. A consumer never drops a frame it
-> does not understand.
+- **Bytes** — arbitrary bytes. Display as raw unless a view knows more.
+- **TextStream** — UTF-8-ish stream chunks. A view may render valid UTF-8 as
+  text and invalid bytes as raw.
+- **JsonRecord** — payloads encoded with serde JSON for a record type. The
+  optional `schema` string is catalog metadata, not a versioned wire envelope.
+
+View decode results are:
+
+- **Record** — a typed/JSON channel decoded to a structured JSON value.
+- **Text** — a text channel decoded as UTF-8.
+- **Raw** — unknown, invalid, or intentionally opaque bytes.
+
+> **Decision of record — forward-compatible fallback.** An unrecognized channel
+> name is `Opaque` to the caller's view, never a pipe error. A newer producer may
+> introduce a channel an older consumer has never heard of; that consumer stores
+> the frame and renders raw bytes rather than rejecting it.
 
 ### 5.5 Record ↔ channel binding
 
-A typed channel's schema is a Rust type that knows its own channel and
-round-trips through the codec:
+A typed record binds to a channel name, not directly to a numeric id:
 
 ```rust
-pub trait Record: Serialize + DeserializeOwned + Sized {
-    const CHANNEL: &'static str;            // or a template, for dynamic families
-    fn channel() -> ChannelId;              // the concrete path
-    fn encode(&self) -> Vec<u8>;            // serde_json by default
-    fn decode(bytes: &[u8]) -> Result<Self>;
+pub trait Record: Serialize + for<'de> Deserialize<'de> + Sized {
+    const CHANNEL: &'static str;
+
+    fn channel_name() -> &'static str { Self::CHANNEL }
+    fn encode(&self) -> Vec<u8> { serde_json::to_vec(self).unwrap() }
+    fn decode(payload: &[u8]) -> Result<Self, serde_json::Error>;
 }
 ```
 
-The codec contract is `decode(encode(r)) == r`. In practice most dynamic
-families are `Text`/`Opaque` (process output), and typed records are mostly
-static; typed-dynamic is supported but uncommon and deliberately un-elaborated.
+To emit a record, register the record's channel name on a producer/endpoint to
+obtain a `ChannelId`, then call `submit_record(channel, &record)`. The codec
+contract is still `decode(encode(r)) == r` for compatible producer/consumer
+versions.
 
-### 5.6 Schema evolution
+### 5.6 Schema evolution and schema metadata
 
-Schema evolution is serde discipline, stated as a rule rather than a mechanism:
+Schema evolution is still serde discipline:
 
 - **Add fields freely.** New fields are emitted by new producers.
-- **Tolerate missing.** `#[serde(default)]` fills a field an older producer did
+- **Tolerate missing.** `#[serde(default)]` fills fields an older producer did
   not send.
 - **Ignore unknown.** A consumer drops fields it does not recognize.
 - **Never remove or repurpose.** Retire a field by leaving it unused; introduce
-  meaning as a new field.
+  new meaning as a new field.
 
-There is **no schema-version number** in the path or the frame. Append-only
-field evolution makes one unnecessary, and absent is simpler than present.
+Current `JsonRecord` descriptors carry `schema: Option<String>`. In the current
+implementation, `register_record<R>()` uses `Some(R::CHANNEL.to_owned())` as the
+schema string. Treat this as catalog metadata for display/subscription tooling,
+not as a frame-level schema version. There is still no schema version field in
+`Frame` itself.
 
 ### 5.7 Lifecycle and liveness
 
-Because the stream is append-only, **a source is never deleted.** Its lifecycle
-is "first frame → silence":
+A channel source's data lifecycle remains append-only: first frame, then more
+frames, then silence. Prior frames remain in the store as history. There is no
+channel teardown or channel garbage collection.
 
-- A source comes into existence by emitting.
-- When it ends (a process exits, an actor stops), it emits a **terminal frame**
-  if it has something to say (`proc.<label>.exit` with the code) and then stops.
-- Its prior frames remain in the store as history. There is no teardown and no
-  channel GC.
+A producer may emit a terminal **data record** if the domain has something to
+say, such as a process exit record. Separately, the event vocabulary includes
+`DatastreamEvent::StreamEnded(StreamId)` for stream-level control, and QUIC can
+encode it. Current `DatastreamEndpoint` does not expose a public method to emit
+`StreamEnded`, so it is a defined control event whose emission policy is not yet
+wired through the endpoint API.
 
-**Liveness is a view concern, never a pipe concern.** A consumer decides a
-source is dead by TTL-since-last-frame or by observing its terminal frame. The
-pipe has no notion of "alive."
-
-Bounding, if any, is on the **stored frames** (the store has finite capacity),
-never on the channel identity.
+Liveness remains a view/subscriber concern. A consumer decides a source is dead
+by TTL-since-last-frame, a terminal data record, or a future stream-ended event.
+The store does not infer liveness.
 
 ### 5.8 Discovery
 
-Discovery splits cleanly along the classifier/store line:
+Discovery has moved from store-only scanning to catalog metadata:
 
-- A **caller-owned classifier** answers *meaning* — given a path, how should this
-  consumer decode it?
-- The **store** answers *existence* — which paths have actually been seen.
-  Dynamic paths are discovered by scanning stored frame keys at read time.
+- The **catalog** answers which names and content classes have been declared for
+  a stream.
+- The **store** answers which numeric frames were delivered and stored.
+- A **view** joins the two when it wants named projections.
 
-So "list every live process channel" is a **view over the store** (distinct
-paths matching `proc.*`), not a registry lookup. A classifier says what this
-consumer understands; the store says what is *real*.
+A live subscriber receives an initial `DatastreamSnapshot { streams, channels }`
+and future `ChannelDeclared` events. A durable store that must render names
+after restart must persist or reconstruct catalog descriptors alongside frames;
+frames alone contain only numeric channel ids.
 
-### 5.9 Namespacing and ownership
+### 5.9 Namespaces currently in use
 
-Collisions are prevented by construction, not by a central allocator:
+The spec does not reserve every name below, but these current code paths should
+not be contradicted:
 
-- Each subsystem **owns its namespace** (the first segment): only the `proc`
-  code mints `proc.*`, only the `swim` code mints `swim.*`. One module is the
-  single source of a namespace's paths.
-- An **instance segment must be a stable identifier already unique within that
-  namespace's domain**: an actor address, a peer `NodeId`, a process label.
-  Uniqueness is inherited from the domain; no coordination is required.
+| Namespace/path | Current meaning |
+|----------------|-----------------|
+| `datastream.health` | datastream self-health counters |
+| `host.cpu`, `host.gpu`, `host.net` | host hardware samples |
+| `runtime.actors` | runtime actor stats hook output |
+| `proc.<label>.stdout`, `proc.<label>.stderr` | managed process output |
+| `mvp.lifecycle` | MVP lifecycle facts |
+| `mvp.provisioning.events` | MVP provisioning lifecycle facts |
+| `mvp.provisioning.logs.node.<id>.<stream>` | MVP provisioning stdout/stderr/provider lines |
 
 ---
 
 ## 6. Transport
 
-### 6.1 Best-effort carriage
+### 6.1 Two transport-facing shapes
+
+The crate currently has two related seams.
+
+The older store/test seam is `Delivery`:
 
 ```rust
-pub trait FrameSink { fn deliver(&self, frames: &[Frame]); }
-pub struct Delivery { pub stream: StreamId, pub frame: Frame }
+pub struct Delivery {
+    pub stream: StreamId,
+    pub frame: Frame,
+}
 ```
 
-The transport drains the mux and carries `(StreamId, Frame)` pairs to the
-consumer. Implementations are pluggable: in-process, UDP datagrams, the cluster
-actor transport, or a scripted transport that injects faults for tests. None of
-them changes anything above or below.
+`Consumer::accept` ingests `Delivery`. Scripted transport and the legacy actor
+wire envelope still use this shape.
 
-### 6.2 Fan-out: drain once, distribute
+The live endpoint/subscription seam is `DatastreamEvent` (§2.5). It carries
+catalog declarations and `FrameDelivery` events. A transitional helper converts
+`DatastreamEvent::Frame` back to `Delivery` when a stored-stream test or adapter
+needs the old shape.
 
-A node may feed more than one consumer (for example, a local renderer and a
-remote collector). Draining is destructive — the buffer can be drained once —
-so fan-out **cannot** be "each sink drains."
+### 6.2 Local fanout: drain once, publish to subscribers
 
-> **Decision of record — fan-out shape.** The driver drains the mux **once** per
-> tick into a batch, then distributes that batch to each registered
-> `FrameSink`. Fan-out is the driver's responsibility, not the mux's and not a
-> sink-chaining trick. The mux stays a single drain; sinks stay independent and
-> side-effect-only.
+Draining the mux is destructive, so the endpoint drains once per tick:
 
-### 6.3 What the transport may and may not do
+```rust
+pub struct EndpointTick {
+    pub drained: usize,
+    pub delivered: usize,
+    pub dropped_for_subscribers: usize,
+    pub subscribers: usize,
+}
+```
 
-| May (consumer must tolerate) | May **not** |
-|------------------------------|-------------|
-| **Drop** a frame → a gap     | **Corrupt** a payload, channel, or position |
-| **Reorder** frames           | **Renumber** a frame |
-| **Delay** a frame            | **Fabricate** a frame |
-| **Duplicate** a frame        | Merge two streams' frames |
+`DatastreamEndpoint::tick` drains queued mux frames, converts them to
+`DatastreamEvent::Frame`, and publishes the batch to `DeliveryFanout`
+subscribers. Future-event fanout is broadcast-only: every current subscriber is
+offered every future event, regardless of its `SubscriptionRequest`. A slow
+subscriber drops only its own copies; other subscribers can still receive the
+same batch. If no subscribers exist, drained frames are bitbucketed and counted.
 
-The envelope is self-describing (length-prefixed fields) and decodes a delivery
-or fails cleanly; it never yields a *plausible but wrong* frame. So ingest must
-handle drop / reorder / delay / duplicate, and may assume away corruption.
+Subscriptions are **future-event streams plus an initial snapshot**. A new
+subscriber receives catalog metadata filtered by its `SubscriptionRequest`; the
+same request remains useful to subscribers and downstream filters, but the
+endpoint does not filter future fanout events. It does not replay prior frames
+unless a caller separately reads a store.
+
+### 6.3 What transport may and may not do
+
+A consumer/store must tolerate:
+
+| May happen | Required response |
+|------------|-------------------|
+| Drop an assigned frame | surface an interior gap when bracketing positions exist |
+| Reorder frames | assemble by position |
+| Delay frames | same as reorder from the store's perspective |
+| Duplicate frames | keep one frame for the position; first stored frame wins |
+| Drop subscriber copies | count the subscriber drop; do not block producer or other subscribers |
+
+A conforming transport may **not** corrupt payload bytes, renumber frames,
+fabricate plausible frames, or merge two streams' frames under one `StreamId`.
+Decode failures drop/abort the malformed record instead of yielding a plausible
+but wrong frame.
+
+Current `ScriptedTransport` deliberately models drops and reorders/delays only;
+it does not inject duplicates even though ingest/store remain idempotent.
+
+### 6.4 Wire formats in current use
+
+The legacy delivery envelope encodes:
+
+```text
+node_len/node, life, position, channel_id_u32, payload_len/payload
+```
+
+This supports old actor transport and real-I/O envelope tests.
+
+The current QUIC adapter uses a catalog-aware stream:
+
+1. A header: magic, flow id, token, stream descriptor, channel descriptors.
+2. Tagged records:
+   - `ChannelDeclared` as JSON descriptor.
+   - `Frame` as numeric channel id, position, and payload bytes.
+   - `StreamEnded` as a tag.
+   - `StreamDeclared` is skipped because the stream descriptor is already in the
+     header.
+
+The new spec treats the event stream as the live transport shape and the legacy
+`Delivery` envelope as a compatibility/test seam unless a caller explicitly uses
+it.
 
 ---
 
@@ -469,8 +745,12 @@ pub struct Consumer { store: Store }
 impl Consumer { fn accept(&mut self, d: Delivery) -> bool; }
 ```
 
-Ingest is stateless routing: route each delivery to its stream by `StreamId`,
-record the frame at its position. No decoding, no thinning, no aggregation.
+Ingest routes each `Delivery` to its `StreamId` and records the frame at its
+position. It does not decode, thin, aggregate, or inspect payload bytes.
+
+Live `DatastreamEvent::Frame` values can be converted to `Delivery` when feeding
+the store. Catalog events are not stored by `Store`; callers that need durable
+name resolution must persist catalog metadata elsewhere or extend storage.
 
 ### 7.2 Out-of-order, idempotent assembly
 
@@ -482,74 +762,134 @@ pub struct StoredStream { frames: BTreeMap<u64, Frame> }
   position key, so it lands in order regardless of arrival order.
 - **Duplicates collapse:** a position delivered twice is recorded once; the
   first frame wins. Ingest is idempotent.
+- **Payloads stay whole:** a frame on an unknown channel id is stored exactly
+  like any other frame.
 
-### 7.3 The store is the truth; gaps are derived
+### 7.3 The store is frame truth; gaps are derived
 
 ```rust
 pub struct Store { streams: BTreeMap<StreamId, StoredStream> }
 ```
 
-The store holds each node's frames whole and append-only, keyed by `StreamId`.
-**Gaps are not stored — they are derived** at read time by walking adjacent
-stored positions. An interior hole (a position between two stored frames that is
-itself absent) is a gap; trailing absence is just "nothing yet," not a gap.
+The store holds each stream's frames whole and append-only, keyed by `StreamId`.
+Gaps are not stored. They are derived by walking adjacent stored positions. An
+interior hole — a missing position between two stored frames — is a gap.
+Trailing absence after the last delivered frame is not knowable as a gap; it is
+just the stream ending or going silent from the store's perspective.
+
+Gap derivation is proportional to the number of stored frames, not the size of
+the missing position span.
+
+### 7.4 Store/catalog boundary
+
+A stored `Frame` contains `ChannelId`, `Position`, and payload bytes, but not the
+channel name or `ChannelContent`. Therefore:
+
+- the store can reconstruct order and gaps without catalog metadata;
+- a raw view can render payload bytes without catalog metadata;
+- a named/typed view needs a `ChannelId -> channel name` resolver and usually a
+  classifier/registry;
+- durable systems that want named views after restart must persist catalog
+  descriptors together with or near the frame store.
 
 ---
 
-## 8. Views — where all semantics live
+## 8. Views — where payload semantics live
 
 ### 8.1 Read-time projections
 
-A view is the only thing that interprets a payload. Everything displayed is
-*computed at read time* from the stored stream; nothing is precomputed in the
-pipe. This is where the telemetry zoo of §1 re-enters — as queries, not as types.
+A view is the only place producer payload bytes are interpreted. Everything
+displayed is computed at read time from stored frames plus optional catalog
+metadata/classifiers; nothing is precomputed in the mux, transport, ingest, or
+store.
 
-### 8.2 Standard views
+The current view API operates over `StoredStream`, whose frames carry numeric
+channels. Named decoding therefore needs a resolver from `ChannelId` to channel
+name and a caller-owned `ChannelClassifier`.
 
-| View              | Meaning |
-|-------------------|---------|
-| `latest(channel)` | the most recent frame on a channel — i.e. a *gauge* |
-| `merged_log()`    | all frames in position order, gaps surfaced as `LogEntry::Gap` — i.e. an *event log* |
-| `metric_series<R>(channel)` | a typed channel decoded to `Vec<(Position, R)>` — a *time series* |
-| `tail(n)` / `grep(pat)` / `filter(pred)` | windowed and predicate-restricted views |
+### 8.2 Implemented standard views
 
-`latest` vs `merged_log` over the *same channel* is exactly the gauge-vs-log
-distinction — chosen per query by the consumer, never bound to the channel.
+| View/API | Meaning |
+|----------|---------|
+| `merged_log(stream)` | all stored frames in position order, gaps surfaced as `LogEntry::Gap`, bodies raw |
+| `merged_log_with_names(stream, classifier, resolve_name)` | merged log with `ChannelId -> name` resolution and caller-owned body decoding |
+| `replay(stream)` | iterator over the raw merged timeline |
+| `metric_series_on<R>(stream, channel)` | decode frames on one numeric channel as `R` |
+| `metric_series<R>(stream)` | attempt to decode every frame as `R`, regardless of channel |
+| `tail(n)` / `grep(needle)` / `filter(pred)` | windowed and predicate-restricted raw frame views |
+
+`latest(channel)` was in the old spec but is not currently implemented in
+`views.rs`; do not treat it as a current conformance requirement until an API is
+added.
 
 ### 8.3 Graceful degradation
 
-A view never fails on data it cannot interpret:
+A view never fails the whole query because one payload is unknown or invalid:
 
-- An unknown channel decodes to `Body::Raw(bytes)` (§5.4).
-- A typed frame that will not parse (version skew, truncation upstream) is
-  skipped or shown raw, not fatal.
-- A gap is rendered as a gap, not as missing context that silently corrupts a
-  series.
+- Unknown channel name or missing resolver result decodes to `Body::Raw(bytes)`.
+- A typed/JSON payload that will not parse is shown raw or skipped by the typed
+  series API, depending on the view.
+- Invalid UTF-8 on a text channel is shown raw.
+- A gap is rendered as a gap, not silently collapsed.
+
+### 8.4 Gauge, log, and time series are projections
+
+Gauge-vs-log-vs-series is still a query choice, not a pipe type:
+
+- "latest value" means a view selects the last stored frame for a channel.
+- "event log" means a view walks all frames, usually with gaps surfaced.
+- "time series" means a view decodes selected frames into records keyed by
+  `Position`.
+
+The same stored bytes can participate in multiple projections.
 
 ---
 
-## 9. Invariants (conformance checklist)
+## 9. Invariants and conformance notes
 
-A conforming implementation upholds all of these. Each is a property a test can
-assert without reaching into internals (scenario / property / contract style).
+### 9.1 Invariants
 
-1. **Opaque payloads.** No code between a producer's `submit` and a view's
-   decode inspects, parses, or branches on payload bytes.
-2. **Gap-free numbering.** Within a stream, assigned positions are monotonic and
-   never reused; a dropped frame leaves its position permanently absent.
-3. **Drops are detectable, not silent.** A lost frame appears as a gap, never as
-   a renumber or a backfill.
-4. **No back-pressure.** A full buffer drops; it never blocks or stalls a
+A conforming implementation upholds these properties. Each should be testable
+without reaching into private internals.
+
+1. **Opaque producer payloads.** No code between producer submission and a view's
+   decode inspects, parses, or branches on producer payload bytes.
+2. **Stream-local numeric channel ids.** A raw `ChannelId` is meaningful only
+   with its `StreamId`; named rendering requires a channel descriptor.
+3. **Catalog consistency.** Within one stream, duplicate registration with the
+   same name/content returns the existing channel id; duplicate registration
+   with conflicting content errors.
+4. **Drain-time position assignment.** Within a stream, drain-assigned positions
+   are monotonic and never reused.
+5. **Queue drops are pre-position.** A full mux queue drops before assignment and
+   does not create a position hole.
+6. **Assigned-frame drops are detectable when bracketed.** A lost interior frame
+   appears as a derived gap, never as renumbering or backfill.
+7. **No producer back-pressure.** A full mux queue drops instead of blocking a
    producer.
-5. **Incarnations never merge.** Frames from the same `node` but different `life`
-   are stored as distinct streams.
-6. **Idempotent ingest.** Delivering the same `(StreamId, position)` twice
-   yields one stored frame.
-7. **Order independence.** Reordered or delayed deliveries reconstruct the same
-   stored stream as in-order delivery.
-8. **Unknown channels survive.** A frame on an unrecognized channel is stored
-   whole and rendered as raw bytes, never dropped or errored.
-9. **Version skew tolerance.** A typed record missing or carrying extra fields
-   decodes (defaults fill, extras ignore) rather than failing.
-10. **Semantics only in views.** Removing every view leaves a pipe that still
-    transports and stores every frame correctly.
+8. **Position order is canonical for store/views.** Store and view replay use
+   mux positions; endpoint drain/fanout order is queue order, not a
+   sorted-position guarantee.
+9. **Incarnations never merge.** Frames from the same `node` but different
+   `life` are stored as distinct streams.
+10. **Idempotent ingest.** Delivering the same `(StreamId, position)` twice
+    yields one stored frame; first stored frame wins.
+11. **Order independence.** Reordered or delayed deliveries reconstruct the same
+    stored stream as in-order delivery, modulo drops and duplicate collapse.
+12. **Unknown channels survive.** A frame with an unknown numeric id/name is
+    stored whole and rendered raw unless metadata later enables decoding.
+13. **Version skew tolerance.** Typed JSON records use serde-compatible
+    evolution: defaults for missing fields, ignored unknown fields, no field
+    repurposing.
+14. **Subscriber isolation.** A slow subscriber can lose its own event copies
+    without blocking the endpoint or other subscribers.
+15. **Semantics only in views.** Removing every view leaves a pipe that still
+    allocates channels, orders frames, transports events, ingests deliveries,
+    and stores frames correctly.
+
+### 9.2 Compatibility notes
+
+The old `DatastreamEmitter`, `DatastreamEventSink`, legacy actor envelope, and
+`Delivery` test seam are compatibility surfaces. New producer code should prefer
+`DatastreamEndpoint` and `DatastreamProducer`; new live transports should prefer
+catalog-aware `DatastreamEvent` streams.
diff --git a/crates/datastream/src/emit.rs b/crates/datastream/src/emit.rs
index d91248e..290b22b 100644
--- a/crates/datastream/src/emit.rs
+++ b/crates/datastream/src/emit.rs
@@ -7,14 +7,14 @@ use swactor::actor::ActorAddress;
 use swactor::process_observer::ProcessOutputObserver;
 use swactor::runtime::Runtime;
 
-use super::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId};
+use super::frame::{ChannelId, Frame, Lifetime, NodeId, StreamId};
 use super::mux::Mux;
 use super::record::Record;
 use super::wire::{DatastreamFrame, encode_delivery};
 
-/// Where assembled frames go once the mux has ordered them.
+/// Legacy sink for frames after mux drain has assigned positions.
 pub trait FrameSink: Send {
-    /// Ship one ordered frame for `stream`. Best-effort: a sink may drop.
+    /// Ship one positioned frame for `stream`. Best-effort: a sink may drop.
     fn ship(&mut self, stream: &StreamId, frame: &Frame);
 }
 
@@ -73,26 +73,22 @@ impl DatastreamEmitter {
         self.mux.dropped()
     }
 
-    pub fn set_frame_timing_enabled(&self, enabled: bool) {
-        self.mux.set_frame_timing_enabled(enabled);
-    }
-
-    pub fn frame_timing_enabled(&self) -> bool {
-        self.mux.frame_timing_enabled()
-    }
-
-    pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> Position {
+    pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> bool {
         self.mux.submit(channel, record.encode())
     }
 
-    pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> Position {
+    pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> bool {
         self.mux.submit(channel, text.as_ref().to_vec())
     }
 
-    pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> Position {
+    pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> bool {
         self.mux.submit(channel, bytes)
     }
 
+    pub fn submit_text_owned(&self, channel: ChannelId, text: String) -> bool {
+        self.mux.submit(channel, text.into_bytes())
+    }
+
     pub fn process_observer_with<F>(&self, channel_for: F) -> Arc<dyn ProcessOutputObserver>
     where
         F: Fn(&str, bool) -> ChannelId + Send + Sync + 'static,
@@ -123,17 +119,21 @@ pub struct DatastreamEventSink {
 }
 
 impl DatastreamEventSink {
-    pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> Position {
+    pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> bool {
         self.mux.submit(channel, record.encode())
     }
 
-    pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> Position {
+    pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> bool {
         self.mux.submit(channel, text.as_ref().to_vec())
     }
 
-    pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> Position {
+    pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> bool {
         self.mux.submit(channel, bytes)
     }
+
+    pub fn submit_text_owned(&self, channel: ChannelId, text: String) -> bool {
+        self.mux.submit(channel, text.into_bytes())
+    }
 }
 
 /// A sink that drops everything.
diff --git a/crates/datastream/src/endpoint.rs b/crates/datastream/src/endpoint.rs
index 565f195..c317402 100644
--- a/crates/datastream/src/endpoint.rs
+++ b/crates/datastream/src/endpoint.rs
@@ -6,7 +6,11 @@
 
 use std::collections::BTreeMap;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Arc, Mutex, mpsc};
+use std::sync::{Arc, Mutex};
+
+use crossbeam_channel::{
+    Receiver, RecvError, RecvTimeoutError, Sender, TryRecvError, TrySendError, bounded,
+};
 
 use serde::{Deserialize, Serialize};
 use swactor::process_observer::ProcessOutputObserver;
@@ -14,12 +18,10 @@ use swactor::stats::{ActorSnapshot, StatsHook};
 
 use crate::frame::{
     ChannelContent, ChannelDescriptor, ChannelFilter, ChannelId, ChannelRef, DatastreamEvent,
-    FrameDelivery, Position, SourceFilter, StreamDescriptor, StreamId, StreamOrigin,
-    SubscriptionRequest,
+    FrameDelivery, SourceFilter, StreamDescriptor, StreamId, StreamOrigin, SubscriptionRequest,
 };
 use crate::mux::Mux;
 use crate::record::Record;
-use crate::timing::{FRAME_TIME_CHANNEL, FRAME_TIME_CHANNEL_ID};
 use crate::transport::Delivery;
 
 const DEFAULT_MUX_CAPACITY: usize = 4096;
@@ -51,7 +53,7 @@ pub struct SubscriberSnapshot {
     pub dropped: u64,
 }
 
-/// Current catalog snapshot delivered at subscription time.
+/// Current catalog snapshot delivered at subscription time, filtered by request.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct DatastreamSnapshot {
     pub streams: Vec<StreamDescriptor>,
@@ -70,7 +72,7 @@ pub struct DatastreamSubscription {
     name: String,
     request: SubscriptionRequest,
     snapshot: DatastreamSnapshot,
-    rx: mpsc::Receiver<DatastreamEvent>,
+    rx: Receiver<DatastreamEvent>,
 }
 
 impl DatastreamSubscription {
@@ -90,18 +92,18 @@ impl DatastreamSubscription {
         &self.snapshot
     }
 
-    pub fn try_recv(&self) -> Result<DatastreamEvent, mpsc::TryRecvError> {
+    pub fn try_recv(&self) -> Result<DatastreamEvent, TryRecvError> {
         self.rx.try_recv()
     }
 
-    pub fn recv(&self) -> Result<DatastreamEvent, mpsc::RecvError> {
+    pub fn recv(&self) -> Result<DatastreamEvent, RecvError> {
         self.rx.recv()
     }
 
     pub fn recv_timeout(
         &self,
         timeout: std::time::Duration,
-    ) -> Result<DatastreamEvent, mpsc::RecvTimeoutError> {
+    ) -> Result<DatastreamEvent, RecvTimeoutError> {
         self.rx.recv_timeout(timeout)
     }
 
@@ -116,17 +118,27 @@ impl DatastreamSubscription {
 
 struct SubscriberSlot {
     name: String,
-    request: SubscriptionRequest,
-    tx: mpsc::SyncSender<DatastreamEvent>,
+    tx: Sender<DatastreamEvent>,
     dropped: u64,
 }
 
+struct FanoutTarget {
+    id: SubscriptionId,
+    tx: Sender<DatastreamEvent>,
+}
+
+struct FanoutReport {
+    id: SubscriptionId,
+    dropped: u64,
+    disconnected: bool,
+}
+
 struct FanoutState {
     next_id: u64,
     subscribers: BTreeMap<SubscriptionId, SubscriberSlot>,
 }
 
-/// Local event fanout used by endpoints and collectors.
+/// Local fanout; future events are broadcast to every subscriber without request filtering.
 pub struct DeliveryFanout {
     default_capacity: usize,
     state: Mutex<FanoutState>,
@@ -168,7 +180,7 @@ impl DeliveryFanout {
         capacity: usize,
     ) -> DatastreamSubscription {
         let name = name.into();
-        let (tx, rx) = mpsc::sync_channel(capacity.max(1));
+        let (tx, rx) = bounded(capacity.max(1));
         let mut state = self.state.lock().expect("datastream fanout poisoned");
         let id = SubscriptionId(state.next_id);
         state.next_id = state.next_id.wrapping_add(1).max(1);
@@ -176,7 +188,6 @@ impl DeliveryFanout {
             id,
             SubscriberSlot {
                 name: name.clone(),
-                request: request.clone(),
                 tx,
                 dropped: 0,
             },
@@ -212,23 +223,34 @@ impl DeliveryFanout {
             .collect()
     }
 
-    pub fn publish(&self, event: DatastreamEvent, catalog: &CatalogSnapshot) -> EndpointTick {
-        self.publish_batch(std::iter::once(event), catalog)
+    pub fn publish(&self, event: DatastreamEvent) -> EndpointTick {
+        self.publish_batch(std::iter::once(event))
     }
 
-    pub fn publish_batch(
-        &self,
-        events: impl IntoIterator<Item = DatastreamEvent>,
-        catalog: &CatalogSnapshot,
-    ) -> EndpointTick {
+    pub fn publish_batch(&self, events: impl IntoIterator<Item = DatastreamEvent>) -> EndpointTick {
         let events: Vec<DatastreamEvent> = events.into_iter().collect();
         if events.is_empty() {
             return EndpointTick::default();
         }
 
-        let mut state = self.state.lock().expect("datastream fanout poisoned");
-        let subscribers = state.subscribers.len();
-        if subscribers == 0 {
+        // Snapshot sender handles while holding the subscriber map lock, then
+        // deliver outside the lock so large batches or slow subscribers do not
+        // block subscribe/snapshot control-plane operations.
+        let (targets, subscribers) = {
+            let state = self.state.lock().expect("datastream fanout poisoned");
+            let subscribers = state.subscribers.len();
+            let targets = state
+                .subscribers
+                .iter()
+                .map(|(id, slot)| FanoutTarget {
+                    id: *id,
+                    tx: slot.tx.clone(),
+                })
+                .collect::<Vec<_>>();
+            (targets, subscribers)
+        };
+
+        if targets.is_empty() {
             return EndpointTick {
                 drained: events.len(),
                 subscribers: 0,
@@ -237,36 +259,48 @@ impl DeliveryFanout {
         }
 
         let mut delivered = 0;
-        let mut dropped = 0;
-        let mut disconnected = Vec::new();
-        for (id, slot) in state.subscribers.iter_mut() {
+        let mut reports = Vec::new();
+        for target in targets {
+            let mut dropped = 0;
+            let mut disconnected = false;
             for event in &events {
-                if !event_matches_request(event, &slot.request, catalog) {
-                    continue;
-                }
-                match slot.tx.try_send(event.clone()) {
+                match target.tx.try_send(event.clone()) {
                     Ok(()) => delivered += 1,
-                    Err(mpsc::TrySendError::Full(_)) => {
-                        slot.dropped = slot.dropped.saturating_add(1);
+                    Err(TrySendError::Full(_)) => dropped += 1,
+                    Err(TrySendError::Disconnected(_)) => {
                         dropped += 1;
-                    }
-                    Err(mpsc::TrySendError::Disconnected(_)) => {
-                        slot.dropped = slot.dropped.saturating_add(1);
-                        dropped += 1;
-                        disconnected.push(*id);
-                        break;
+                        disconnected = true;
                     }
                 }
             }
+            if dropped > 0 || disconnected {
+                reports.push(FanoutReport {
+                    id: target.id,
+                    dropped,
+                    disconnected,
+                });
+            }
         }
-        for id in disconnected {
-            state.subscribers.remove(&id);
+
+        let dropped_for_subscribers = reports.iter().map(|report| report.dropped).sum::<u64>();
+        if !reports.is_empty() {
+            let mut state = self.state.lock().expect("datastream fanout poisoned");
+            for report in &reports {
+                if let Some(slot) = state.subscribers.get_mut(&report.id) {
+                    slot.dropped = slot.dropped.saturating_add(report.dropped);
+                }
+            }
+            for report in &reports {
+                if report.disconnected {
+                    state.subscribers.remove(&report.id);
+                }
+            }
         }
 
         EndpointTick {
             drained: events.len(),
             delivered,
-            dropped_for_subscribers: dropped,
+            dropped_for_subscribers: usize::try_from(dropped_for_subscribers).unwrap_or(usize::MAX),
             subscribers,
         }
     }
@@ -279,6 +313,7 @@ pub struct CatalogSnapshot {
 }
 
 impl CatalogSnapshot {
+    /// Apply a subscription request to the initial metadata snapshot only.
     pub fn datastream_snapshot(&self, request: &SubscriptionRequest) -> DatastreamSnapshot {
         let channels: Vec<ChannelDescriptor> = self
             .channels
@@ -320,24 +355,12 @@ struct ChannelCatalogState {
 
 impl ChannelCatalogState {
     fn new(stream: StreamDescriptor) -> Self {
-        let mut state = Self {
-            stream: stream.clone(),
+        Self {
+            stream,
             by_id: BTreeMap::new(),
             by_name: BTreeMap::new(),
             next_channel: 1,
-        };
-        let timing = ChannelDescriptor {
-            stream: stream.stream.clone(),
-            id: FRAME_TIME_CHANNEL_ID,
-            name: FRAME_TIME_CHANNEL.to_owned(),
-            label: Some("frame construction time".to_owned()),
-            content: ChannelContent::JsonRecord {
-                schema: Some("datastream.frame_time.v1".to_owned()),
-            },
-        };
-        state.by_name.insert(timing.name.clone(), timing.id);
-        state.by_id.insert(timing.id, timing);
-        state
+        }
     }
 
     fn snapshot(&self) -> CatalogSnapshot {
@@ -456,14 +479,6 @@ impl DatastreamEndpoint {
             .snapshot()
     }
 
-    pub fn set_frame_timing_enabled(&self, enabled: bool) {
-        self.mux.set_frame_timing_enabled(enabled);
-    }
-
-    pub fn frame_timing_enabled(&self) -> bool {
-        self.mux.frame_timing_enabled()
-    }
-
     pub fn producer(&self) -> DatastreamProducer {
         DatastreamProducer {
             mux: Arc::clone(&self.mux),
@@ -530,26 +545,36 @@ impl DatastreamEndpoint {
         self.fanout.subscriber_snapshots()
     }
 
-    /// Drain the mux and fan out catalog-aware future events.
+    /// Drain the mux once and broadcast future frame events; with no subscribers, drained frames are bitbucketed.
     pub fn tick(&self) -> EndpointTick {
-        let frames = self.mux.drain();
-        if frames.is_empty() {
+        let events = self.drain_events();
+        if events.is_empty() {
             return EndpointTick::default();
         }
-        let drained = frames.len();
-        self.drained.fetch_add(drained as u64, Ordering::Relaxed);
-        let events = frames.into_iter().map(|frame| {
-            DatastreamEvent::Frame(FrameDelivery {
-                channel: ChannelRef {
-                    stream: self.stream.clone(),
-                    channel: frame.channel,
-                },
-                position: frame.position,
-                payload: frame.payload,
+        self.publish_events(events)
+    }
+
+    fn drain_events(&self) -> Vec<DatastreamEvent> {
+        self.mux
+            .drain()
+            .into_iter()
+            .map(|frame| {
+                DatastreamEvent::Frame(FrameDelivery {
+                    channel: ChannelRef {
+                        stream: self.stream.clone(),
+                        channel: frame.channel,
+                    },
+                    position: frame.position,
+                    payload: frame.payload,
+                })
             })
-        });
-        let catalog = self.catalog_snapshot();
-        let tick = self.fanout.publish_batch(events, &catalog);
+            .collect()
+    }
+
+    fn publish_events(&self, events: Vec<DatastreamEvent>) -> EndpointTick {
+        let drained = events.len();
+        self.drained.fetch_add(drained as u64, Ordering::Relaxed);
+        let tick = self.fanout.publish_batch(events);
         if tick.subscribers == 0 {
             self.bitbucketed
                 .fetch_add(drained as u64, Ordering::Relaxed);
@@ -620,24 +645,20 @@ impl DatastreamProducer {
             .id_for_name(name)
     }
 
-    pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> Position {
+    pub fn submit_record<R: Record>(&self, channel: ChannelId, record: &R) -> bool {
         self.mux.submit(channel, record.encode())
     }
 
-    pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> Position {
+    pub fn submit_text(&self, channel: ChannelId, text: impl AsRef<[u8]>) -> bool {
         self.mux.submit(channel, text.as_ref().to_vec())
     }
 
-    pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> Position {
+    pub fn submit_bytes(&self, channel: ChannelId, bytes: Vec<u8>) -> bool {
         self.mux.submit(channel, bytes)
     }
 
-    pub fn set_frame_timing_enabled(&self, enabled: bool) {
-        self.mux.set_frame_timing_enabled(enabled);
-    }
-
-    pub fn frame_timing_enabled(&self) -> bool {
-        self.mux.frame_timing_enabled()
+    pub fn submit_text_owned(&self, channel: ChannelId, text: String) -> bool {
+        self.mux.submit(channel, text.into_bytes())
     }
 
     pub fn process_observer_with<F>(&self, channel_for: F) -> Arc<dyn ProcessOutputObserver>
@@ -674,29 +695,25 @@ fn register_channel(
     name: String,
     content: ChannelContent,
 ) -> Result<ChannelId, ChannelRegistrationError> {
-    let (id, event, snapshot) = {
+    // New channel declarations publish a future event immediately; existing-name
+    // reuse only returns the prior id.
+    let (id, event) = {
         let mut catalog = catalog.lock().expect("datastream catalog poisoned");
         match catalog.try_register_channel(name.clone(), content)? {
-            Some(descriptor) => {
-                let id = descriptor.id;
-                let snapshot = catalog.snapshot();
-                (
-                    id,
-                    Some(DatastreamEvent::ChannelDeclared(descriptor)),
-                    snapshot,
-                )
-            }
+            Some(descriptor) => (
+                descriptor.id,
+                Some(DatastreamEvent::ChannelDeclared(descriptor)),
+            ),
             None => {
                 let id = catalog
                     .id_for_name(&name)
                     .expect("duplicate channel name remains registered");
-                let snapshot = catalog.snapshot();
-                (id, None, snapshot)
+                (id, None)
             }
         }
     };
     if let Some(event) = event {
-        let _ = fanout.publish(event, &snapshot);
+        let _ = fanout.publish(event);
     }
     Ok(id)
 }
@@ -773,33 +790,6 @@ struct RuntimeMessageTypeCount<'a> {
     count: u64,
 }
 
-fn event_matches_request(
-    event: &DatastreamEvent,
-    request: &SubscriptionRequest,
-    catalog: &CatalogSnapshot,
-) -> bool {
-    match event {
-        DatastreamEvent::StreamDeclared(descriptor) => {
-            source_matches(&descriptor.stream, Some(descriptor), &request.sources)
-                && (matches!(request.channels, ChannelFilter::All)
-                    || catalog.channels.values().any(|channel| {
-                        channel.stream == descriptor.stream
-                            && descriptor_matches_request(channel, request, catalog)
-                    }))
-        }
-        DatastreamEvent::ChannelDeclared(descriptor) => {
-            descriptor_matches_request(descriptor, request, catalog)
-        }
-        DatastreamEvent::Frame(delivery) => catalog
-            .descriptor_for(&delivery.channel)
-            .map(|descriptor| descriptor_matches_request(descriptor, request, catalog))
-            .unwrap_or_else(|| matches!(request.channels, ChannelFilter::All)),
-        DatastreamEvent::StreamEnded(stream) => {
-            source_matches(stream, catalog.stream_descriptor(stream), &request.sources)
-        }
-    }
-}
-
 fn descriptor_matches_request(
     descriptor: &ChannelDescriptor,
     request: &SubscriptionRequest,
diff --git a/crates/datastream/src/frame.rs b/crates/datastream/src/frame.rs
index 2a73b93..26d2b5f 100644
--- a/crates/datastream/src/frame.rs
+++ b/crates/datastream/src/frame.rs
@@ -1,4 +1,4 @@
-//! Core data model: the framed, channel-multiplexed stream (spec §4).
+//! Core data model: the framed, channel-multiplexed stream (spec §2).
 //!
 //! A stream is identified by the producing node and lifetime. Frames carry a
 //! stream-local numeric channel id plus the mux-assigned position and opaque
@@ -10,12 +10,12 @@ use std::sync::Arc;
 
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 
-/// A position assigned by a node's mux (spec §5.2).
+/// A position assigned by a node's mux during drain (spec §2.3).
 ///
-/// Positions are **monotonic** and **gap-free** within a single node's stream:
-/// the mux never reuses one and never skips one in its numbering. A position
-/// that is assigned but never delivered surfaces downstream as a missing
-/// position — a detectable gap (spec §5.3, §7.5).
+/// Assignment is monotonic and gap-free for accepted frames: the mux never
+/// reuses one and never skips one while draining. A frame assigned a position
+/// can still be lost by transport and later surface downstream as a detectable
+/// gap.
 #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Serialize, Deserialize)]
 pub struct Position(pub u64);
 
@@ -27,9 +27,9 @@ impl fmt::Display for Position {
 
 /// Stream-local numeric channel id.
 ///
-/// `ChannelId(0)` is reserved for the datastream frame-timing sidecar. Every
-/// other id is allocated by the stream owner and is meaningful only with the
-/// corresponding [`StreamId`]. Consumers resolve frames by `(stream, channel)`.
+/// A raw `ChannelId` is meaningful only together with its [`StreamId`]. The
+/// public endpoint allocator currently starts at `ChannelId(1)`, leaving
+/// `ChannelId(0)` unallocated by normal registration.
 #[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, Serialize, Deserialize)]
 pub struct ChannelId(pub u32);
 
@@ -73,7 +73,7 @@ pub enum StreamOrigin {
     RemoteNode,
 }
 
-/// The stable identity of a node that produces a stream (spec §4.4, §8.1).
+/// The stable identity of a node that produces a stream (spec §2.2, §7.1).
 #[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
 pub struct NodeId(Arc<str>);
 
@@ -131,7 +131,7 @@ impl<'de> Deserialize<'de> for NodeId {
     }
 }
 
-/// A lifetime discriminator distinguishing a node's incarnations (spec §8.4).
+/// A lifetime discriminator distinguishing a node's incarnations (spec §2.2).
 #[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Debug, Serialize, Deserialize)]
 pub struct Lifetime(pub u64);
 
@@ -168,7 +168,7 @@ pub struct StreamDescriptor {
     pub origin: StreamOrigin,
 }
 
-/// Channel metadata declared by the stream owner.
+/// Catalog metadata declared by the stream owner; frames store only `id` and payload.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ChannelDescriptor {
     pub stream: StreamId,
@@ -236,7 +236,7 @@ impl SubscriptionRequest {
     }
 }
 
-/// The unit the mux emits (spec §4.1): bytes tagged with a channel and a position.
+/// The unit the mux emits (spec §2.1): bytes tagged with a channel and a position.
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Frame {
     /// The stream-local lane these bytes belong to.
diff --git a/crates/datastream/src/ingest.rs b/crates/datastream/src/ingest.rs
index a86979d..6799d30 100644
--- a/crates/datastream/src/ingest.rs
+++ b/crates/datastream/src/ingest.rs
@@ -1,24 +1,23 @@
-//! Consumer ingest: reconstruct each node's stream from deliveries (spec
-//! §8.1).
+//! Consumer ingest: reconstruct each stream from deliveries (spec §7.1).
 //!
 //! The consumer receives frames from many nodes, in any order, some never
-//! arriving, and reconstructs each node's stream — keyed by stream id,
-//! ordered by position — into the [`Store`]. Ingest is deliberately thin:
-//! it routes a delivery to its stream and records the frame whole. It MUST
-//! NOT thin, aggregate, decode-and-discard, or truncate (spec §8.2), and it
-//! never inspects a channel or a payload, so a channel it cannot decode is
-//! retained exactly like any other (spec §8.3).
+//! arriving, and reconstructs each stream — keyed by stream id, ordered by
+//! position — into the [`Store`]. Ingest is deliberately thin: it routes a
+//! delivery to its stream and records the frame whole. It MUST NOT thin,
+//! aggregate, decode-and-discard, or truncate (spec §7.1, §7.2), and it never
+//! inspects a channel or payload, so an undecoded channel is retained exactly
+//! like any other (spec §7.2, §8.3).
 //!
-//! Reconstruction is by position, not arrival: out-of-order deliveries land
-//! in order in the store, and a position delivered twice collapses to one
-//! (spec §7.5). Two lives of one node are different stream ids and never
-//! merge (spec §8.4).
+//! Reconstruction is by position, not arrival: out-of-order deliveries land in
+//! order in the store, and a position delivered twice collapses to one (spec
+//! §7.2). Two lives of one node are different stream ids and never merge (spec
+//! §2.2).
 
 use super::store::Store;
 use super::transport::Delivery;
 
-/// The single consumer toward which all telemetry flows (spec §2). It owns
-/// the stored streams and grows them as deliveries arrive.
+/// A store-owning ingest fold for deliveries. It grows stored streams as
+/// deliveries arrive.
 #[derive(Debug, Default)]
 pub struct Consumer {
     store: Store,
@@ -33,7 +32,7 @@ impl Consumer {
     /// Accept one delivery: route it to its stream and record the frame.
     /// Returns `true` if the frame was new (a duplicate position is
     /// ignored, keeping the first — the carrier cannot fabricate content,
-    /// spec §9).
+    /// spec §9.1).
     pub fn accept(&mut self, delivery: Delivery) -> bool {
         let Delivery { stream, frame } = delivery;
         self.store.stream_mut(&stream).record(frame)
@@ -46,7 +45,7 @@ impl Consumer {
         }
     }
 
-    /// The stored streams — the source of truth for every view (spec §8.2).
+    /// The stored streams — frame truth for raw storage and view projections (spec §7.3, §8.1).
     pub fn store(&self) -> &Store {
         &self.store
     }
diff --git a/crates/datastream/src/lib.rs b/crates/datastream/src/lib.rs
index dfcd941..b105a62 100644
--- a/crates/datastream/src/lib.rs
+++ b/crates/datastream/src/lib.rs
@@ -1,34 +1,35 @@
 //! The per-node telemetry **datastream** (see `DATASTREAM_SPEC.md`).
 //!
-//! A deliberately dumb pipe: producers dump bytes tagged by channel, a
-//! single per-node mux interleaves them into one ordered stream, a
-//! best-effort transport carries that stream to the one consumer, ingest
-//! reconstructs each node's stream by position, and views are read-time
-//! projections over the stored stream. Nothing between a producer and a
+//! A deliberately dumb pipe: producers dump bytes tagged by stream-local
+//! channel id, a single per-node mux accepts those bytes and assigns canonical
+//! positions during drain, the endpoint broadcasts catalog-aware events to
+//! subscribers, ingest reconstructs streams by position, and views are
+//! read-time projections over stored frames. Nothing between a producer and a
 //! view interprets the payload.
 //!
 //! ```text
 //!    producers (caller-owned records + text)
-//!             │  bytes tagged by channel        → [`record::Record`]
+//!             │  bytes tagged by registered ChannelId
 //!             ▼
-//!         per-node MUX                          → [`mux::Mux`]
-//!             │  one ordered stream of [`Frame`]s
+//!    endpoint / catalog                         → [`endpoint`]
+//!             │  channel metadata + producer handles
 //!             ▼
-//!    best-effort transport                      → [`transport`]
-//!             │  delivery: frames, maybe dropped/reordered/delayed
+//!    per-node MUX                               → [`mux::Mux`]
+//!             │  positioned [`frame::Frame`]s
 //!             ▼
-//!       consumer INGEST                         → [`ingest::Consumer`]
-//!             │  complete stream, stored whole
+//!    endpoint fanout                            → [`endpoint::DeliveryFanout`]
+//!             │  catalog-aware events, maybe dropped per subscriber
 //!             ▼
-//!      stored STREAM (truth)                    → [`store`]
-//!             │  read-time only
+//!    ingest / store                             → [`ingest`], [`store`]
+//!             │  position-keyed frame truth
 //!             ▼
-//!         VIEWS                                 → [`views`]
+//!    views                                      → [`views`]
 //! ```
 //!
-//! The data model ([`frame`]), extension contract ([`record`]), and wire
-//! envelope ([`wire`]) are the seams a test observes. Channel meanings live in
-//! producer/consumer crates, not in a datastream-wide catalog.
+//! The data model ([`frame`]), extension contract ([`record`]), endpoint/fanout
+//! seam ([`endpoint`]), and compatibility wire helpers ([`wire`]) are the seams
+//! tests observe. Channel meanings live in producer/consumer crates, not in a
+//! datastream-wide global registry.
 
 pub mod emit;
 pub mod endpoint;
@@ -41,7 +42,6 @@ pub mod publisher_actor;
 pub mod record;
 pub mod sink_actor;
 pub mod store;
-pub mod timing;
 pub mod transport;
 pub mod views;
 pub mod wire;
@@ -65,6 +65,5 @@ pub use publisher_actor::{
 pub use record::{ChannelKind, ChannelRegistry, Record};
 pub use sink_actor::{DATASTREAM_SINK_NAME, DatastreamSink};
 pub use store::{GapSpan, Store, StoredStream};
-pub use timing::{FRAME_TIME_CHANNEL, FRAME_TIME_CHANNEL_ID, FrameTimeSample};
 pub use transport::{Delivery, Reorder, ScriptedTransport, StreamScript};
 pub use views::{Body, LogEntry, MergedFrame};
diff --git a/crates/datastream/src/mux.rs b/crates/datastream/src/mux.rs
index bd56c74..101c642 100644
--- a/crates/datastream/src/mux.rs
+++ b/crates/datastream/src/mux.rs
@@ -1,42 +1,39 @@
-//! The per-node mux: the single ordering authority (spec §5).
+//! The per-stream mux: the single ordering authority (spec §4.4).
 //!
-//! Every producer on a node submits bytes tagged with a stream-local channel id
-//! to one mux, and the mux assigns a single monotonic position sequence across
-//! all channels. A drop consumes a position and is therefore visible downstream
-//! as a gap.
+//! Producers submit bytes tagged with a stream-local channel id. The mux accepts
+//! payloads into a bounded queue first, then assigns a single monotonic position
+//! sequence while draining accepted payloads.
 
-use std::sync::Mutex;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::mpsc::{Receiver, SyncSender, TryRecvError, TrySendError, sync_channel};
-use std::time::{SystemTime, UNIX_EPOCH};
+use crossbeam_channel::{Receiver, Sender, TryRecvError, TrySendError, bounded};
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use super::frame::{ChannelId, Frame, Position, StreamId};
-use super::record::Record;
-use super::timing::FRAME_TIME_CHANNEL_ID;
-use super::timing::FrameTimeSample;
+
+struct PendingFrame {
+    channel: ChannelId,
+    payload: Vec<u8>,
+}
 
 /// A node's single position authority and outgoing telemetry queue.
 pub struct Mux {
     stream: StreamId,
     next: AtomicU64,
     dropped: AtomicU64,
-    frame_timing_enabled: AtomicBool,
-    tx: SyncSender<Frame>,
-    rx: Mutex<Receiver<Frame>>,
+    tx: Sender<PendingFrame>,
+    rx: Receiver<PendingFrame>,
 }
 
 impl Mux {
     /// Create a mux for `stream` with a bounded outgoing queue.
     pub fn new(stream: StreamId, capacity: usize) -> Self {
         let capacity = capacity.max(1).min(1_048_576);
-        let (tx, rx) = sync_channel(capacity);
+        let (tx, rx) = bounded(capacity);
         Mux {
             stream,
             next: AtomicU64::new(0),
             dropped: AtomicU64::new(0),
-            frame_timing_enabled: AtomicBool::new(true),
             tx,
-            rx: Mutex::new(rx),
+            rx,
         }
     }
 
@@ -45,88 +42,50 @@ impl Mux {
         Mux::new(stream, usize::MAX)
     }
 
-    /// The stream this mux produces (spec §8.4 ingest key).
+    /// The stream this mux produces (spec §2.2, §7.1 ingest key).
     pub fn stream_id(&self) -> &StreamId {
         &self.stream
     }
 
     /// Submit opaque bytes on a registered channel id.
-    pub fn submit(&self, channel: ChannelId, payload: Vec<u8>) -> Position {
-        let position = Position(self.next.fetch_add(1, Ordering::Relaxed));
-        let timing_sample = self
-            .frame_timing_enabled
-            .load(Ordering::Relaxed)
-            .then(|| now_unix_ns())
-            .filter(|_| channel != FRAME_TIME_CHANNEL_ID)
-            .map(|created_at_unix_ns| FrameTimeSample::new(position, created_at_unix_ns));
-        let frame = Frame {
-            channel,
-            position,
-            payload,
-        };
-
-        match self.tx.try_send(frame) {
-            Ok(()) => {
-                if let Some(sample) = timing_sample {
-                    self.push_timing_sample_if_room(sample);
-                }
-            }
+    pub fn submit(&self, channel: ChannelId, payload: Vec<u8>) -> bool {
+        match self.tx.try_send(PendingFrame { channel, payload }) {
+            Ok(()) => true,
             Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
                 self.dropped.fetch_add(1, Ordering::Relaxed);
+                false
             }
         }
-        position
     }
 
-    /// Pull all currently queued frames, sorted by mux position.
+    /// Pull all currently queued frames in mux queue order.
     pub fn drain(&self) -> Vec<Frame> {
-        let rx = self.rx.lock().expect("mux receiver poisoned");
         let mut frames = Vec::new();
         loop {
-            match rx.try_recv() {
-                Ok(frame) => frames.push(frame),
-                Err(TryRecvError::Empty) => break,
-                Err(TryRecvError::Disconnected) => break,
+            match self.rx.try_recv() {
+                Ok(pending) => {
+                    // Position is consumed only after a pending frame has left
+                    // the queue; failed submit never reaches this point.
+                    let position = Position(self.next.fetch_add(1, Ordering::Relaxed));
+                    frames.push(Frame {
+                        channel: pending.channel,
+                        position,
+                        payload: pending.payload,
+                    });
+                }
+                Err(TryRecvError::Empty) | Err(TryRecvError::Disconnected) => break,
             }
         }
-        frames.sort_by_key(|frame| frame.position);
         frames
     }
 
-    /// Enable or disable optional sidecar timing samples for newly submitted frames.
-    pub fn set_frame_timing_enabled(&self, enabled: bool) {
-        self.frame_timing_enabled.store(enabled, Ordering::Relaxed);
-    }
-
-    /// Whether this mux currently emits sidecar frame timing samples.
-    pub fn frame_timing_enabled(&self) -> bool {
-        self.frame_timing_enabled.load(Ordering::Relaxed)
-    }
-
-    fn push_timing_sample_if_room(&self, sample: FrameTimeSample) {
-        let position = Position(self.next.fetch_add(1, Ordering::Relaxed));
-        let frame = Frame {
-            channel: FRAME_TIME_CHANNEL_ID,
-            position,
-            payload: sample.encode(),
-        };
-        let _ = self.tx.try_send(frame);
-    }
-
-    /// How many positions have been assigned — the gap-free high-water mark.
+    /// How many positions have been assigned while draining accepted frames.
     pub fn assigned(&self) -> u64 {
         self.next.load(Ordering::Relaxed)
     }
 
-    /// How many data frames have been dropped on overflow.
+    /// How many submissions have been dropped before entering the mux.
     pub fn dropped(&self) -> u64 {
         self.dropped.load(Ordering::Relaxed)
     }
 }
-
-fn now_unix_ns() -> u64 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map(|duration| u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX))
-        .unwrap_or(0)
-}
diff --git a/crates/datastream/src/sink_actor.rs b/crates/datastream/src/sink_actor.rs
index 86960dd..d5d610a 100644
--- a/crates/datastream/src/sink_actor.rs
+++ b/crates/datastream/src/sink_actor.rs
@@ -1,15 +1,15 @@
-//! `DatastreamSink` — the cluster-side consumer of [`DatastreamFrame`] messages.
+//! Legacy `DatastreamSink` for [`DatastreamFrame`] actor messages.
 //!
-//! This is the counterpart to [`ClusterFrameSink`](super::emit::ClusterFrameSink):
-//! a node ships its ordered telemetry as `DatastreamFrame` actor messages over the
-//! regular swactor transport, and this actor — registered under a well-known name
-//! on the collector (e.g. the orchestrator) — receives them, decodes each back
-//! into a `(StreamId, Frame)` delivery, and hands it to a caller-supplied fold.
+//! New live transports should prefer catalog-aware [`crate::DatastreamEvent`]
+//! streams. This actor remains as the counterpart to
+//! [`ClusterFrameSink`](super::emit::ClusterFrameSink): a node ships positioned
+//! telemetry as legacy `DatastreamFrame` actor messages over regular Swactor
+//! transport, and this actor receives them, decodes each back into a
+//! `(StreamId, Frame)` delivery, and hands it to a caller-supplied fold.
 //!
 //! It deliberately knows nothing about any view layer. The actor owns an opaque
-//! callback so binaries can wire the decoded deliveries into whichever fold they
-//! need. Malformed payloads are dropped silently — the same best-effort tolerance
-//! the UDP ingest had.
+//! callback so binaries can wire decoded deliveries into whichever fold they
+//! need. Malformed payloads are dropped silently.
 
 use swactor::actor::ActorInterface;
 use swactor::runtime::Ctx;
@@ -17,9 +17,9 @@ use swactor::runtime::Ctx;
 use super::frame::{Frame, StreamId};
 use super::wire::{DatastreamFrame, decode_delivery};
 
-/// Receives [`DatastreamFrame`] cluster messages and folds each decoded delivery
-/// through `on_frame`. Spawn it, then publish its address under
-/// [`DATASTREAM_SINK_NAME`] so emitters can resolve and ship to it.
+/// Receives legacy [`DatastreamFrame`] cluster messages and folds each decoded
+/// delivery through `on_frame`. Spawn it, then publish its address under
+/// [`DATASTREAM_SINK_NAME`] so legacy emitters can resolve and ship to it.
 pub struct DatastreamSink {
     on_frame: Box<dyn FnMut(StreamId, Frame) + Send>,
 }
diff --git a/crates/datastream/src/store.rs b/crates/datastream/src/store.rs
index 063008f..aa85292 100644
--- a/crates/datastream/src/store.rs
+++ b/crates/datastream/src/store.rs
@@ -1,15 +1,15 @@
-//! The stored stream — the consumer's source of truth (spec §8).
+//! The stored stream — the consumer's frame source of truth (spec §7).
 //!
-//! Storage holds each node's **complete** stream, whole and append-only.
-//! Nothing is thinned, aggregated, decoded-and-discarded, or truncated at
-//! ingest (spec §8.2); everything a view ever shows is derived from here
-//! (spec §9.1). Frames on channels the consumer cannot decode are kept as
-//! opaque bytes, in order, alongside the rest (spec §8.3) — the store never
-//! looks at a channel or a payload.
+//! Storage holds each node's complete stream, whole and append-only. Nothing is
+//! thinned, aggregated, decoded-and-discarded, or truncated at ingest (spec
+//! §7.1, §7.2); everything a view shows is derived from here (spec §8.1).
+//! Frames on channels the consumer cannot decode are kept as opaque bytes,
+//! alongside the rest (spec §7.2, §8.3) — the store never looks at a channel or
+//! payload.
 //!
 //! A [`StoredStream`] is keyed in the [`Store`] by [`StreamId`] — node plus
-//! lifetime — so a re-incarnated node does not append to its prior life
-//! (spec §8.4).
+//! lifetime — so a re-incarnated node does not append to its prior life (spec
+//! §2.2, §7.1).
 
 use std::collections::BTreeMap;
 
@@ -19,8 +19,8 @@ use super::frame::{Frame, Position, StreamId};
 ///
 /// Backed by a position-keyed map so out-of-order arrivals land in order
 /// and a position seen twice collapses to one (the carrier may not
-/// fabricate content, spec §9). Gaps are not stored — they are *derived*
-/// at read time from the positions that are present (spec §9.1).
+/// fabricate content, spec §9.1). Gaps are not stored — they are *derived*
+/// at read time from the positions that are present (spec §7.3, §8.1).
 #[derive(Debug, Clone, Default)]
 pub struct StoredStream {
     frames: BTreeMap<u64, Frame>,
@@ -34,7 +34,7 @@ impl StoredStream {
 
     /// Record a delivered frame. Idempotent by position: the first frame
     /// seen for a position wins and is never mutated (append-only,
-    /// spec §8.2). Returns `true` if this was the first time the position
+    /// spec §7.2). Returns `true` if this was the first time the position
     /// was seen.
     pub fn record(&mut self, frame: Frame) -> bool {
         match self.frames.entry(frame.position.0) {
@@ -73,19 +73,19 @@ impl StoredStream {
     }
 
     /// The **interior** gaps — runs of positions assigned between the first
-    /// and last delivered frame but never delivered (spec §7.5, §8), each as
-    /// one [`GapSpan`].
+    /// and last delivered frame but never delivered (spec §7.3), each as one
+    /// [`GapSpan`].
     ///
     /// Cost is O(stored frames), never O(gap size): it walks adjacent stored
     /// positions and reads each span's endpoints from them, rather than
     /// enumerating the (possibly enormous) range in between. A stream that
     /// brackets a huge interior gap — what a long consumer outage produces
-    /// (spec §7.4), or a single wild position from a corrupt datagram — still
-    /// surfaces in work proportional to the frames held, not to `u64::MAX`.
+    /// (spec §6.3, §7.3), or a single wild position from a corrupt datagram —
+    /// still surfaces in work proportional to the frames held, not to `u64::MAX`.
     ///
-    /// Only interior gaps are knowable: a position lost *after* the last
-    /// delivered frame leaves no bracketing frame to reveal it, so it shows
-    /// up as the stream simply ending (spec §7.4 node death), not a gap.
+    /// Leading and trailing losses are not derivable from stored frames because
+    /// no bracketing position exists. A position lost after the last delivered
+    /// frame shows up as the stream simply ending (spec §5.7, §7.3), not a gap.
     pub fn gap_spans(&self) -> Vec<GapSpan> {
         let mut spans = Vec::new();
         let mut prev: Option<u64> = None;
@@ -105,7 +105,7 @@ impl StoredStream {
 }
 
 /// A contiguous run of missing positions surfaced in a stored stream
-/// (spec §7.5). Inclusive on both ends.
+/// (spec §7.3). Inclusive on both ends.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct GapSpan {
     /// First missing position.
@@ -121,7 +121,7 @@ impl GapSpan {
     }
 }
 
-/// All stored streams at the consumer, keyed by [`StreamId`] (spec §8.4).
+/// All stored streams at the consumer, keyed by [`StreamId`] (spec §2.2, §7.1).
 ///
 /// Two streams with the same node but different lifetime are distinct keys
 /// and never merge.
diff --git a/crates/datastream/src/timing.rs b/crates/datastream/src/timing.rs
deleted file mode 100644
index 15802ee..0000000
--- a/crates/datastream/src/timing.rs
+++ /dev/null
@@ -1,32 +0,0 @@
-//! Optional frame-construction timing sidecar records.
-
-use serde::{Deserialize, Serialize};
-
-use crate::frame::ChannelId;
-
-use crate::frame::Position;
-use crate::record::Record;
-
-/// Reserved channel carrying optional timing samples for frames in the same stream.
-pub const FRAME_TIME_CHANNEL: &str = "datastream.frame_time";
-pub const FRAME_TIME_CHANNEL_ID: ChannelId = ChannelId(0);
-
-/// Sidecar timing sample keyed by the target frame's stream-local position.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct FrameTimeSample {
-    pub target_position: u64,
-    pub created_at_unix_ns: u64,
-}
-
-impl FrameTimeSample {
-    pub fn new(target_position: Position, created_at_unix_ns: u64) -> Self {
-        Self {
-            target_position: target_position.0,
-            created_at_unix_ns,
-        }
-    }
-}
-
-impl Record for FrameTimeSample {
-    const CHANNEL: &'static str = FRAME_TIME_CHANNEL;
-}
diff --git a/crates/datastream/src/transport.rs b/crates/datastream/src/transport.rs
index 43a29a0..aec4203 100644
--- a/crates/datastream/src/transport.rs
+++ b/crates/datastream/src/transport.rs
@@ -1,28 +1,26 @@
-//! Transport: best-effort carriage of a node's stream to the one consumer
-//! (spec §7), and a scripted in-process carrier for offline tests (testing
-//! spec §2, §9).
+//! Legacy transport/test seam for carrying positioned frames into ingest.
 //!
-//! A [`Delivery`] is the value on the transport→ingest seam: which stream a
-//! frame belongs to, and the frame. A real carrier rides the connections
-//! the system already maintains (spec §7.1); a test replaces it with the
-//! [`ScriptedTransport`] here, whose faults are chosen by the scenario and
-//! stay inside the **envelope** (testing spec §9): a carrier may *deliver*,
-//! *drop*, *reorder*, or *delay*, and it MUST NOT corrupt a payload,
-//! fabricate a frame, or alter a position.
+//! The live endpoint path now fans out catalog-aware [`DatastreamEvent`] values;
+//! this module keeps the older [`Delivery`] shape used by ingest, storage tests,
+//! and scripted conformance checks.
 //!
-//! Under position-ordering a *delay* is indistinguishable from a *reorder*
-//! (a delayed frame simply arrives later), so the envelope's delay is
-//! covered by [`Reorder`]. Everything the scripted carrier produces is a
-//! reordered subsequence of what was sent — never a superset, never a
-//! mutation — which is exactly the property the real-transport conformance
-//! check pins (testing spec §9).
+//! A [`Delivery`] pairs the producing stream id with one frame. A real carrier
+//! rides connections the system already maintains; tests can replace it with
+//! [`ScriptedTransport`], whose faults stay inside the transport envelope: it
+//! may *deliver*, *drop*, *reorder*, or *delay*, and it MUST NOT corrupt a
+//! payload, fabricate a frame, or alter a position.
+//!
+//! Under position-ordering a *delay* is indistinguishable from a *reorder*:
+//! a delayed frame simply arrives later. Everything the scripted carrier
+//! produces is a reordered subsequence of what was sent — never a superset and
+//! never a mutation.
 
 use std::collections::BTreeSet;
 
 use super::frame::{Frame, StreamId};
 
-/// A frame as the consumer receives it from the transport (testing spec §2
-/// seam): tagged with the stream it belongs to.
+/// A frame as the legacy transport/ingest seam receives it: tagged with the
+/// stream it belongs to.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Delivery {
     /// Which node's life produced the frame (spec §8.1 ingest key).
@@ -38,9 +36,9 @@ impl Delivery {
     }
 }
 
-/// How the surviving frames of a stream are reordered on arrival. This is
-/// the envelope's *reorder* (and *delay*) axis (testing spec §9); each
-/// variant is a permutation of the survivors, never adding or dropping.
+/// How the surviving frames of a stream are reordered on arrival. This is the
+/// envelope's *reorder* and *delay* axis (spec §6.3, §9.1); each variant is a
+/// permutation of the survivors, never adding or dropping.
 #[derive(Debug, Clone, Default)]
 pub enum Reorder {
     /// Delivered in the order sent.
@@ -57,7 +55,7 @@ pub enum Reorder {
     Permutation(Vec<usize>),
 }
 
-/// The faults a scripted carrier applies to one stream (testing spec §9
+/// The faults a scripted carrier applies to one stream (spec §6.3, §9.1
 /// envelope). Drops and reorders only — payloads and positions are never
 /// touched.
 #[derive(Debug, Clone, Default)]
@@ -91,10 +89,9 @@ impl StreamScript {
     }
 }
 
-/// A scripted, in-process transport (testing spec §2). It is a pure,
-/// deterministic transform from what a node *sent* to what the consumer is
-/// *delivered* — the entanglement of real wires replaced by a script so a
-/// run completes in microseconds and returns the same result every time.
+/// A scripted, in-process transport for conformance scenarios (spec §9.1). It
+/// is a pure, deterministic transform from what a node sent to what the
+/// consumer receives, replacing real wire behavior with a fast test script.
 pub struct ScriptedTransport;
 
 impl ScriptedTransport {
diff --git a/crates/datastream/src/views.rs b/crates/datastream/src/views.rs
index f7361b7..3e3880e 100644
--- a/crates/datastream/src/views.rs
+++ b/crates/datastream/src/views.rs
@@ -1,4 +1,4 @@
-//! Views: read-time projections over a stored stream (spec §9).
+//! Views: read-time projections over a stored stream (spec §8).
 
 use std::fmt;
 
@@ -70,6 +70,7 @@ where
     C: ChannelClassifier + ?Sized,
     R: Fn(ChannelId) -> Option<String>,
 {
+    // Capacity covers stored frames; surfaced gaps may add extra log entries.
     let mut out = Vec::with_capacity(stream.len());
     let mut prev: Option<u64> = None;
     for frame in stream.frames() {
@@ -114,7 +115,8 @@ where
     timeline_with_resolver(stream, classifier, resolve_name)
 }
 
-/// Transitional merged log using numeric channel ids as strings for classifier lookup.
+/// Transitional helper for callers that still classify by rendered numeric ids;
+/// named decoding should use [`merged_log_with_names`].
 pub fn merged_log_with<C: ChannelClassifier + ?Sized>(
     stream: &StoredStream,
     classifier: &C,
diff --git a/crates/datastream/src/wire.rs b/crates/datastream/src/wire.rs
index a7252fa..ad2e0f0 100644
--- a/crates/datastream/src/wire.rs
+++ b/crates/datastream/src/wire.rs
@@ -59,7 +59,7 @@ pub fn decode_delivery(buf: &[u8]) -> Result<(StreamId, Frame), WireError> {
     Ok((stream, frame))
 }
 
-/// One datastream event, addressed to a local/cluster datastream actor.
+/// Legacy actor-message payload wrapper for datastream bytes.
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct DatastreamFrame {
     pub payload: Vec<u8>,
diff --git a/crates/datastream/tests/datastream_support/mod.rs b/crates/datastream/tests/datastream_support/mod.rs
deleted file mode 100644
index 194963f..0000000
--- a/crates/datastream/tests/datastream_support/mod.rs
+++ /dev/null
@@ -1,567 +0,0 @@
-//! Shared support for the datastream tests (testing spec §2, §3).
-//!
-//! Two things live here, both deliberately separate from the system under
-//! test so a test never asserts the code against itself:
-//!
-//! * the **payload library** — realistic record shapes and log lines, the
-//!   bytes the pipe will actually carry (testing spec §5: "never
-//!   placeholder text"); and
-//! * the **reference model** — the spec's rules restated as small, total
-//!   functions over sequences (testing spec §3). It is the trusted oracle:
-//!   every test's `expected` is derived from it, never captured from a run.
-//!   It is written naively on purpose (collect, sort, scan) so a reader can
-//!   confirm it against the spec by eye, while the pipe computes the same
-//!   answers the long way.
-//!
-//! This module is `#[path]`-included into more than one test binary, so
-//! some items are unused in some of them.
-#![allow(dead_code)]
-
-use datastream::mux::Mux;
-use datastream::store::GapSpan;
-pub use datastream::transport::Delivery;
-use datastream::{ChannelId, Frame, Position, Record, StreamId};
-
-pub mod schema {
-    use datastream::{ChannelId, Record};
-    use serde::{Deserialize, Serialize};
-
-    pub const IDENTITY: &str = "identity";
-    pub const HOST_RESOURCE: &str = "host.resource";
-    pub const TRANSPORT_INTERNALS: &str = "transport.internals";
-    pub const MEMBERSHIP: &str = "membership";
-    pub const RUNTIME_STATS: &str = "runtime.stats";
-    pub const DIST_STATE: &str = "dist.state";
-    pub const RUNTIME_ACTORS: &str = "runtime.actors";
-    pub const RUNTIME_WORKERS: &str = "runtime.workers";
-    pub const DATASTREAM_HEALTH: &str = "datastream.health";
-
-    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
-    pub enum ProcStream {
-        Stdout,
-        Stderr,
-    }
-
-    impl ProcStream {
-        pub fn as_str(self) -> &'static str {
-            match self {
-                ProcStream::Stdout => "stdout",
-                ProcStream::Stderr => "stderr",
-            }
-        }
-    }
-
-    pub fn process_output(label: &str, stream: ProcStream) -> ChannelId {
-        ChannelId::new(format!("proc.{label}.{}", stream.as_str()))
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct IdentityRecord {
-        pub node: String,
-        #[serde(default)]
-        pub life: u64,
-        #[serde(default)]
-        pub node_name: String,
-        #[serde(default)]
-        pub listen_addr: String,
-        #[serde(default)]
-        pub relay_url: String,
-        #[serde(default)]
-        pub version: String,
-    }
-
-    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-    pub struct ResourceSample {
-        #[serde(default)]
-        pub cpu_pct: f32,
-        #[serde(default)]
-        pub mem_used_mb: u32,
-        #[serde(default)]
-        pub mem_total_mb: u32,
-        #[serde(default)]
-        pub gpu_pct: f32,
-        #[serde(default)]
-        pub disk_used_gb: u32,
-        #[serde(default)]
-        pub net_rx_kbps: u32,
-        #[serde(default)]
-        pub net_tx_kbps: u32,
-    }
-
-    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-    pub struct TransportInternals {
-        #[serde(default)]
-        pub relay_connected: bool,
-        #[serde(default)]
-        pub direct_peers: u32,
-        #[serde(default)]
-        pub relay_peers: u32,
-        #[serde(default)]
-        pub rtt_ms_p50: u32,
-    }
-
-    #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct MembershipTransition {
-        pub peer: String,
-        pub from: String,
-        pub to: String,
-        #[serde(default)]
-        pub reason: String,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct RuntimeStats {
-        #[serde(default)]
-        pub actors_live: u32,
-        #[serde(default)]
-        pub mailbox_depth: u32,
-        #[serde(default)]
-        pub scheduled_tasks: u32,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct DistributionState {
-        #[serde(default)]
-        pub cache_size: u32,
-        #[serde(default)]
-        pub cache_entries: Vec<CacheEntryRec>,
-        #[serde(default)]
-        pub directory_route_count: u32,
-        #[serde(default)]
-        pub registry_size: u32,
-        #[serde(default)]
-        pub registry_tombstones: u32,
-        #[serde(default)]
-        pub registry_entries: Vec<RegistryEntryRec>,
-        #[serde(default)]
-        pub recent_probe_targets: Vec<String>,
-        #[serde(default)]
-        pub peer_auth_mode: String,
-        #[serde(default)]
-        pub authorized_peer_count: u32,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct CacheEntryRec {
-        #[serde(default)]
-        pub actor_addr: String,
-        #[serde(default)]
-        pub node_id: String,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct RegistryEntryRec {
-        #[serde(default)]
-        pub name: String,
-        #[serde(default)]
-        pub actor_addr: String,
-        #[serde(default)]
-        pub node_id: String,
-        #[serde(default)]
-        pub tombstone: bool,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct WorkerCounters {
-        #[serde(default)]
-        pub num_workers: u32,
-        #[serde(default)]
-        pub scheduled_tasks: u32,
-        #[serde(default)]
-        pub local_sends: u64,
-        #[serde(default)]
-        pub cross_sends: u64,
-        #[serde(default)]
-        pub inbox_sends: u64,
-        #[serde(default)]
-        pub type_mismatches: u64,
-        #[serde(default)]
-        pub panics: u64,
-        #[serde(default)]
-        pub messages_dropped: u64,
-        #[serde(default)]
-        pub restarts: u64,
-        #[serde(default)]
-        pub stops: u64,
-        #[serde(default)]
-        pub messages_processed: u64,
-        #[serde(default)]
-        pub tick_p50_us: u64,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct DatastreamHealth {
-        #[serde(default)]
-        pub assigned: u64,
-        #[serde(default)]
-        pub dropped: u64,
-        #[serde(default)]
-        pub loss_rate_ppm: u32,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct ActorRuntimeDetail {
-        #[serde(default)]
-        pub actors: Vec<ActorRec>,
-    }
-
-    #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
-    pub struct ActorRec {
-        #[serde(default)]
-        pub address: String,
-        #[serde(default)]
-        pub name: String,
-        #[serde(default)]
-        pub mailbox_depth: u32,
-        #[serde(default)]
-        pub messages_processed: u64,
-        #[serde(default)]
-        pub last_msg_type: String,
-        #[serde(default)]
-        pub poisoned: bool,
-        #[serde(default)]
-        pub message_type_counts: Vec<(String, u64)>,
-    }
-
-    impl Record for IdentityRecord {
-        const CHANNEL: &'static str = IDENTITY;
-    }
-    impl Record for ResourceSample {
-        const CHANNEL: &'static str = HOST_RESOURCE;
-    }
-    impl Record for TransportInternals {
-        const CHANNEL: &'static str = TRANSPORT_INTERNALS;
-    }
-    impl Record for MembershipTransition {
-        const CHANNEL: &'static str = MEMBERSHIP;
-    }
-    impl Record for RuntimeStats {
-        const CHANNEL: &'static str = RUNTIME_STATS;
-    }
-    impl Record for DistributionState {
-        const CHANNEL: &'static str = DIST_STATE;
-    }
-    impl Record for ActorRuntimeDetail {
-        const CHANNEL: &'static str = RUNTIME_ACTORS;
-    }
-    impl Record for WorkerCounters {
-        const CHANNEL: &'static str = RUNTIME_WORKERS;
-    }
-    impl Record for DatastreamHealth {
-        const CHANNEL: &'static str = DATASTREAM_HEALTH;
-    }
-}
-
-use schema::*;
-
-/// An in-process node (testing spec §2 — the faked machine boundary): a
-/// *real* mux plus the producers that feed it. Producers push realistic
-/// bytes in at the producer seam; [`Node::sent`] takes the mux's ordered
-/// output stream — the value on the mux→transport seam. Only the machine
-/// boundary is faked; the mux is the production code.
-pub struct Node {
-    stream: StreamId,
-    mux: Mux,
-}
-
-impl Node {
-    /// Stand up a node for a given stream (node identity + lifetime).
-    pub fn new(stream: StreamId) -> Self {
-        Node {
-            mux: Mux::unbounded(stream.clone()),
-            stream,
-        }
-    }
-
-    /// The stream this node produces.
-    pub fn stream_id(&self) -> &StreamId {
-        &self.stream
-    }
-
-    /// A producer emits a typed record on its own channel (spec §6.1).
-    pub fn emit<R: Record>(&self, record: &R) -> Position {
-        self.mux.submit(R::channel(), record.encode())
-    }
-
-    /// A producer emits a line of raw process output (spec §6.2).
-    pub fn emit_text(&self, label: &str, stream: ProcStream, line: &str) -> Position {
-        self.mux
-            .submit(process_output(label, stream), line.as_bytes().to_vec())
-    }
-
-    /// A producer emits bytes on a channel the consumer may not know
-    /// (spec §6.3) — opaque to everything until a view learns the channel.
-    pub fn emit_opaque(&self, channel: &str, bytes: &[u8]) -> Position {
-        self.mux.submit(ChannelId::new(channel), bytes.to_vec())
-    }
-
-    /// Take the node's ordered output stream (drains the mux).
-    pub fn sent(&self) -> Vec<Frame> {
-        self.mux.drain()
-    }
-}
-
-/// Realistic payloads, drawn on by both the verified vectors (testing spec
-/// §5) and the deployment scenario (testing spec §8). Nothing here is
-/// placeholder text.
-pub mod payloads {
-    use super::*;
-
-    /// A boot/identity record for a node, including the descriptive fields a
-    /// node fills once known (name, listen addr, embedded relay, build version).
-    pub fn identity(node: &str, life: u64) -> IdentityRecord {
-        IdentityRecord {
-            node: node.to_string(),
-            life,
-            node_name: format!("swift-{node}"),
-            listen_addr: format!("{node}.iroh:4242"),
-            relay_url: "https://relay.example:4443/".to_string(),
-            version: "ds-inference @ abc1234".to_string(),
-        }
-    }
-
-    /// A consolidated distribution-subsystem state; `tick` nudges the values so
-    /// a series is not constant.
-    pub fn dist_state(tick: u64) -> DistributionState {
-        DistributionState {
-            cache_size: 3 + (tick % 4) as u32,
-            cache_entries: vec![CacheEntryRec {
-                actor_addr: format!("actor-{}", tick % 5),
-                node_id: format!("node-{}", tick % 3),
-            }],
-            directory_route_count: 5 + (tick % 7) as u32,
-            registry_size: 8 + (tick % 3) as u32,
-            registry_tombstones: (tick % 2) as u32,
-            registry_entries: vec![RegistryEntryRec {
-                name: format!("svc-{}", tick % 4),
-                actor_addr: format!("actor-{}", tick % 5),
-                node_id: format!("node-{}", tick % 3),
-                tombstone: tick.is_multiple_of(2),
-            }],
-            recent_probe_targets: vec![format!("peer-{}", tick % 6)],
-            peer_auth_mode: if tick.is_multiple_of(2) {
-                "open".into()
-            } else {
-                "allow-list".into()
-            },
-            authorized_peer_count: (tick % 5) as u32,
-        }
-    }
-
-    /// A per-actor runtime-detail record (the real actor table).
-    pub fn actor_detail(tick: u64) -> ActorRuntimeDetail {
-        ActorRuntimeDetail {
-            actors: vec![
-                ActorRec {
-                    address: format!("{:064x}", tick),
-                    name: "SwimActor".to_string(),
-                    mailbox_depth: (tick % 5) as u32,
-                    messages_processed: 100 + tick,
-                    last_msg_type: "swactor_dist::Ping".to_string(),
-                    poisoned: false,
-                    message_type_counts: vec![("swactor_dist::Ping".to_string(), 40 + tick)],
-                },
-                ActorRec {
-                    address: format!("{:064x}", tick + 1),
-                    name: "RegistryActor".to_string(),
-                    mailbox_depth: 0,
-                    messages_processed: 10 + tick % 3,
-                    last_msg_type: "Tick".to_string(),
-                    poisoned: false,
-                    message_type_counts: vec![("Tick".to_string(), 10 + tick % 3)],
-                },
-            ],
-        }
-    }
-
-    /// A plausible resource sample; `tick` nudges the values so a series is
-    /// not constant.
-    pub fn resource(tick: u64) -> ResourceSample {
-        ResourceSample {
-            cpu_pct: 12.5 + (tick % 7) as f32 * 3.0,
-            mem_used_mb: 2048 + (tick % 5) as u32 * 128,
-            mem_total_mb: 16384,
-            gpu_pct: (tick % 4) as f32 * 25.0,
-            disk_used_gb: 40 + (tick % 3) as u32,
-            net_rx_kbps: 900 + (tick % 11) as u32 * 30,
-            net_tx_kbps: 300 + (tick % 13) as u32 * 20,
-        }
-    }
-
-    /// A transport-internals snapshot.
-    pub fn transport(tick: u64) -> TransportInternals {
-        TransportInternals {
-            relay_connected: !tick.is_multiple_of(9),
-            direct_peers: 2 + (tick % 3) as u32,
-            relay_peers: 1,
-            rtt_ms_p50: 18 + (tick % 5) as u32 * 4,
-        }
-    }
-
-    /// A membership transition between two peers' states.
-    pub fn membership(peer: &str, from: &str, to: &str) -> MembershipTransition {
-        MembershipTransition {
-            peer: peer.to_string(),
-            from: from.to_string(),
-            to: to.to_string(),
-            reason: "probe timeout".to_string(),
-        }
-    }
-
-    /// A runtime-stats record.
-    pub fn runtime(tick: u64) -> RuntimeStats {
-        RuntimeStats {
-            actors_live: 30 + (tick % 6) as u32,
-            mailbox_depth: (tick % 17) as u32,
-            scheduled_tasks: 4 + (tick % 3) as u32,
-        }
-    }
-
-    /// Aggregated worker-runtime counters (the `runtime.workers` channel).
-    pub fn worker_counters(tick: u64) -> WorkerCounters {
-        WorkerCounters {
-            num_workers: 4,
-            scheduled_tasks: 4 + (tick % 3) as u32,
-            local_sends: 100 + tick,
-            cross_sends: 20 + tick,
-            inbox_sends: tick,
-            messages_processed: 1000 + tick * 7,
-            tick_p50_us: 50 + tick,
-            ..Default::default()
-        }
-    }
-
-    /// Datastream self-health (the `datastream.health` channel). `assigned`
-    /// tracks the seed directly so a scenario can pick a frame out by its value.
-    pub fn datastream_health(tick: u64) -> DatastreamHealth {
-        DatastreamHealth {
-            assigned: tick,
-            dropped: tick % 4,
-            loss_rate_ppm: (tick % 4) as u32,
-        }
-    }
-
-    /// A realistic line of process output (without trailing newline).
-    pub fn log_line(label: &str, tick: u64) -> String {
-        format!(
-            "[{label}] step {tick} loss=0.{:03} lr=3e-4",
-            250 - (tick % 200)
-        )
-    }
-}
-
-/// The reference model: the spec's rules as plain total functions over
-/// sequences. No transport, no storage, no concurrency, no time.
-pub mod reference {
-    use super::*;
-
-    /// The frames a consumer was delivered for one stream, in arrival
-    /// order — the raw material reconstruction works over.
-    pub fn delivered_frames(stream: &StreamId, deliveries: &[Delivery]) -> Vec<Frame> {
-        deliveries
-            .iter()
-            .filter(|d| &d.stream == stream)
-            .map(|d| d.frame.clone())
-            .collect()
-    }
-
-    /// Reconstruction (spec §8.1, testing spec §3): "keep the frames that
-    /// were delivered, in position order." Duplicates of a position
-    /// collapse to one (the carrier may not fabricate content, spec §9, so
-    /// a repeat carries identical bytes).
-    pub fn reconstruct(delivered: &[Frame]) -> Vec<Frame> {
-        let mut frames: Vec<Frame> = Vec::new();
-        for f in delivered {
-            if !frames.iter().any(|seen| seen.position == f.position) {
-                frames.push(f.clone());
-            }
-        }
-        frames.sort_by_key(|f| f.position);
-        frames
-    }
-
-    /// The surfaced gaps as spans: the **interior** runs of positions missing
-    /// between the first and last delivered position (spec §7.5, §8). A
-    /// consumer can only detect gaps it has bracketing frames for; positions
-    /// lost after the last delivered frame are invisible and manifest as the
-    /// stream ending (spec §7.4 node death = truncation, not a gap).
-    ///
-    /// Walks the sorted delivered positions — O(frames), never the gap size —
-    /// so the oracle agrees with the store on the cheap path even across a
-    /// near-`u64::MAX` gap.
-    pub fn gap_spans(delivered: &[Frame]) -> Vec<GapSpan> {
-        let recon = reconstruct(delivered);
-        let mut spans = Vec::new();
-        let mut prev: Option<u64> = None;
-        for f in &recon {
-            let p = f.position.0;
-            if let Some(q) = prev
-                && p > q + 1
-            {
-                spans.push(GapSpan {
-                    start: q + 1,
-                    end: p - 1,
-                });
-            }
-            prev = Some(p);
-        }
-        spans
-    }
-
-    /// One structural item on the merged timeline (testing spec §3: "all
-    /// stored frames in position order, channels interleaved"). This is the
-    /// *structure* of the merged log — order and surfaced gaps — decoupled
-    /// from how each payload is rendered for display, which is a separate
-    /// §9.3 concern the tests assert on its own.
-    #[derive(Debug, Clone, PartialEq, Eq)]
-    pub enum TimelineItem {
-        Frame { position: u64, channel: String },
-        Gap { start: u64, end: u64 },
-    }
-
-    /// The merged log oracle (spec §9.2): every delivered frame in position
-    /// order, channels interleaved, with an interior gap surfaced wherever a
-    /// position is missing. Written naively so it is obviously the spec.
-    pub fn merged_log(delivered: &[Frame]) -> Vec<TimelineItem> {
-        let frames = reconstruct(delivered);
-        let mut out = Vec::new();
-        let mut prev: Option<u64> = None;
-        for f in &frames {
-            let pos = f.position.0;
-            if let Some(p) = prev
-                && pos > p + 1
-            {
-                out.push(TimelineItem::Gap {
-                    start: p + 1,
-                    end: pos - 1,
-                });
-            }
-            out.push(TimelineItem::Frame {
-                position: pos,
-                channel: f.channel.to_string(),
-            });
-            prev = Some(pos);
-        }
-        out
-    }
-}
-
-/// Convenience: build a frame on a typed channel from a record.
-pub fn typed_frame<R: Record>(record: &R, position: u64) -> Frame {
-    Frame::new(R::channel(), Position(position), record.encode())
-}
-
-/// Convenience: build a frame on a raw-text process-output channel.
-pub fn text_frame(label: &str, stream: ProcStream, line: &str, position: u64) -> Frame {
-    Frame::new(
-        process_output(label, stream),
-        Position(position),
-        line.as_bytes().to_vec(),
-    )
-}
-
-/// Convenience: a frame on a channel id the consumer does not know — an
-/// opaque channel (spec §6.3).
-pub fn opaque_frame(id: &str, payload: &[u8], position: u64) -> Frame {
-    Frame::new(ChannelId::new(id), Position(position), payload.to_vec())
-}
diff --git a/crates/datastream/tests/t_datastream.rs b/crates/datastream/tests/t_datastream.rs
index ff637b3..67830e1 100644
--- a/crates/datastream/tests/t_datastream.rs
+++ b/crates/datastream/tests/t_datastream.rs
@@ -7,8 +7,7 @@ use datastream::transport::{Delivery, Reorder, ScriptedTransport, StreamScript};
 use datastream::views::{self, Body, LogEntry};
 use datastream::wire::{decode_delivery, encode_delivery};
 use datastream::{
-    ChannelId, ChannelKind, ChannelRegistry, FRAME_TIME_CHANNEL_ID, Frame, FrameTimeSample,
-    Lifetime, NodeId, Position, Record, StreamId,
+    ChannelId, ChannelKind, ChannelRegistry, Frame, Lifetime, NodeId, Position, Record, StreamId,
 };
 use serde::{Deserialize, Serialize};
 
@@ -58,67 +57,81 @@ fn record_codecs_round_trip_without_global_catalog() {
 }
 
 #[test]
-fn mux_numbers_monotonic_and_gap_free() {
+fn mux_assigns_positions_when_drained() {
     let mux = Mux::unbounded(stream());
-    mux.set_frame_timing_enabled(false);
 
     for i in 0..64u64 {
-        let position = mux.submit(RESOURCE_CHANNEL, resource(i as u32).encode());
-        assert_eq!(position, Position(i));
+        assert!(mux.submit(RESOURCE_CHANNEL, resource(i as u32).encode()));
     }
 
-    assert_eq!(mux.assigned(), 64);
+    assert_eq!(mux.assigned(), 0);
     assert_eq!(mux.dropped(), 0);
-    let positions: Vec<u64> = mux.drain().iter().map(|frame| frame.position.0).collect();
+    let mut positions: Vec<u64> = mux.drain().iter().map(|frame| frame.position.0).collect();
+    positions.sort_unstable();
     assert_eq!(positions, (0..64).collect::<Vec<_>>());
+    assert_eq!(mux.assigned(), 64);
 }
 
 #[test]
-fn mux_queue_preserves_position_order_across_producers() {
+fn mux_concurrent_producers_assign_unique_positions_on_drain() {
     let mux = Arc::new(Mux::unbounded(stream()));
-    mux.set_frame_timing_enabled(false);
 
     let mut threads = Vec::new();
     for producer in 0..4u8 {
         let mux = Arc::clone(&mux);
         threads.push(thread::spawn(move || {
+            let mut accepted = 0;
             for seq in 0..32u8 {
-                mux.submit(LOG_CHANNEL, vec![producer, seq]);
+                if mux.submit(LOG_CHANNEL, vec![producer, seq]) {
+                    accepted += 1;
+                }
             }
+            accepted
         }));
     }
-    for thread in threads {
-        thread.join().expect("producer thread completes");
-    }
+    let accepted: usize = threads
+        .into_iter()
+        .map(|thread| thread.join().expect("producer thread completes"))
+        .sum();
 
+    assert_eq!(accepted, 128);
+    assert_eq!(mux.assigned(), 0);
     let frames = mux.drain();
     assert_eq!(frames.len(), 128);
-    for (expected, frame) in frames.iter().enumerate() {
-        assert_eq!(frame.position, Position(expected as u64));
-    }
+    let mut positions: Vec<u64> = frames.iter().map(|frame| frame.position.0).collect();
+    positions.sort_unstable();
+    assert_eq!(positions, (0..128).collect::<Vec<_>>());
+    assert_eq!(mux.assigned(), 128);
 }
 
 #[test]
-fn mux_timing_sidecar_uses_reserved_numeric_channel_and_does_not_recurse() {
-    let mux = Mux::unbounded(stream());
-    assert!(mux.frame_timing_enabled());
+fn mux_full_queue_drops_without_consuming_position() {
+    let mux = Mux::new(stream(), 1);
 
-    let data_position = mux.submit(RESOURCE_CHANNEL, resource(0).encode());
-    let timing_position = mux.submit(
-        FRAME_TIME_CHANNEL_ID,
-        FrameTimeSample::new(Position(42), 123).encode(),
-    );
+    assert!(mux.submit(LOG_CHANNEL, b"first".to_vec()));
+    assert!(!mux.submit(LOG_CHANNEL, b"second".to_vec()));
+    assert_eq!(mux.dropped(), 1);
+    assert_eq!(mux.assigned(), 0);
+
+    let frames = mux.drain();
+    assert_eq!(frames.len(), 1);
+    assert_eq!(frames[0].channel, LOG_CHANNEL);
+    assert_eq!(frames[0].position, Position(0));
+    assert_eq!(frames[0].payload, b"first");
+    assert_eq!(mux.assigned(), 1);
+}
+
+#[test]
+fn mux_one_submit_one_drained_frame_without_timing_sidecar() {
+    let mux = Mux::unbounded(stream());
+
+    assert!(mux.submit(RESOURCE_CHANNEL, resource(0).encode()));
     let frames = mux.drain();
 
-    assert_eq!(data_position, Position(0));
-    assert_eq!(timing_position, Position(2));
-    assert_eq!(mux.assigned(), 3);
-    assert_eq!(frames.len(), 3);
+    assert_eq!(frames.len(), 1);
     assert_eq!(frames[0].channel, RESOURCE_CHANNEL);
-    assert_eq!(frames[1].channel, FRAME_TIME_CHANNEL_ID);
-    assert_eq!(frames[2].channel, FRAME_TIME_CHANNEL_ID);
-    let sample = FrameTimeSample::decode(&frames[1].payload).expect("timing decodes");
-    assert_eq!(sample.target_position, data_position.0);
+    assert_eq!(frames[0].position, Position(0));
+    assert_eq!(frames[0].payload, resource(0).encode());
 }
 
 #[test]
diff --git a/crates/datastream/tests/t_datastream_endpoint.rs b/crates/datastream/tests/t_datastream_endpoint.rs
index 0602df1..9d5d7fc 100644
--- a/crates/datastream/tests/t_datastream_endpoint.rs
+++ b/crates/datastream/tests/t_datastream_endpoint.rs
@@ -2,8 +2,8 @@ use std::time::Duration;
 
 use datastream::{
     ChannelContent, ChannelContentKind, ChannelFilter, ChannelId, DatastreamEndpoint,
-    DatastreamEvent, FRAME_TIME_CHANNEL_ID, FrameDelivery, FrameTimeSample, Lifetime, NodeId,
-    Position, Record, SourceFilter, StreamId, SubscriptionRequest,
+    DatastreamEvent, FrameDelivery, Lifetime, NodeId, Position, Record, SourceFilter, StreamId,
+    SubscriptionRequest,
 };
 use serde_json::Value;
 use swactor::actor::ActorAddress;
@@ -37,7 +37,6 @@ fn frame_event(event: &DatastreamEvent) -> &FrameDelivery {
 fn endpoint_without_subscribers_drains_to_bitbucket() {
     let endpoint = endpoint();
     let producer = endpoint.producer();
-    producer.set_frame_timing_enabled(false);
     let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
 
     producer.submit_text(log, "before");
@@ -63,12 +62,12 @@ fn channel_registration_allocates_numeric_ids() {
     assert_eq!(stderr, ChannelId(2));
     assert_eq!(runtime, ChannelId(3));
     let catalog = endpoint.catalog_snapshot();
+    let removed_timing_name = ["datastream", "frame_time"].join(".");
     assert!(
-        catalog
+        !catalog
             .channels
             .values()
-            .any(|descriptor| descriptor.id == FRAME_TIME_CHANNEL_ID
-                && descriptor.name == "datastream.frame_time")
+            .any(|descriptor| descriptor.name == removed_timing_name)
     );
 }
 
@@ -118,7 +117,6 @@ fn subscription_snapshot_contains_stream_and_channel_metadata() {
 fn subscription_receives_only_future_matching_frames() {
     let endpoint = endpoint();
     let producer = endpoint.producer();
-    producer.set_frame_timing_enabled(false);
     let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
 
     producer.submit_text(log, "pre-subscription");
@@ -138,10 +136,9 @@ fn subscription_receives_only_future_matching_frames() {
 }
 
 #[test]
-fn subscription_filters_textstream_channels() {
+fn subscription_snapshot_filters_but_future_fanout_broadcasts() {
     let endpoint = endpoint();
     let producer = endpoint.producer();
-    producer.set_frame_timing_enabled(false);
     let stdout = producer.register_channel("stdout", ChannelContent::TextStream);
     let json = producer.register_record::<RuntimeRecord>();
     let text_subscription = endpoint.subscribe(
@@ -152,22 +149,26 @@ fn subscription_filters_textstream_channels() {
         },
     );
 
+    assert_eq!(text_subscription.snapshot().channels.len(), 1);
+    assert_eq!(text_subscription.snapshot().channels[0].id, stdout);
+
     producer.submit_text(stdout, "line");
     producer.submit_record(json, &RuntimeRecord { value: 5 });
     endpoint.tick();
 
     let events = text_subscription.drain_available();
-    assert_eq!(events.len(), 1);
-    let delivery = frame_event(&events[0]);
-    assert_eq!(delivery.channel.channel, stdout);
-    assert_eq!(delivery.payload, b"line");
+    assert_eq!(events.len(), 2);
+    let channels: Vec<ChannelId> = events
+        .iter()
+        .map(|event| frame_event(event).channel.channel)
+        .collect();
+    assert_eq!(channels, vec![stdout, json]);
 }
 
 #[test]
 fn endpoint_fans_out_ordered_frames_to_multiple_subscribers() {
     let endpoint = endpoint();
     let producer = endpoint.producer();
-    producer.set_frame_timing_enabled(false);
     let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
     let left = endpoint.subscribe_all("left");
     let right = endpoint.subscribe_all("right");
@@ -188,7 +189,6 @@ fn endpoint_fans_out_ordered_frames_to_multiple_subscribers() {
 fn slow_subscriber_drops_without_blocking_fast_subscriber() {
     let endpoint = endpoint();
     let producer = endpoint.producer();
-    producer.set_frame_timing_enabled(false);
     let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
     let slow = endpoint.subscribe_all_with_capacity("slow", 1);
     let fast = endpoint.subscribe_all_with_capacity("fast", 8);
@@ -212,35 +212,56 @@ fn slow_subscriber_drops_without_blocking_fast_subscriber() {
 }
 
 #[test]
-fn endpoint_producer_timing_sidecars_are_fanned_out_when_enabled() {
+fn channel_declared_is_broadcast_to_filtered_subscribers() {
+    let endpoint = endpoint();
+    let subscription = endpoint.subscribe(
+        "text-only",
+        SubscriptionRequest {
+            sources: SourceFilter::All,
+            channels: ChannelFilter::Content(ChannelContentKind::TextStream),
+        },
+    );
+
+    let channel = endpoint.register_channel(
+        "runtime.json",
+        ChannelContent::JsonRecord {
+            schema: Some("runtime.json".to_owned()),
+        },
+    );
+
+    let events = subscription.drain_available();
+    assert_eq!(events.len(), 1);
+    match &events[0] {
+        DatastreamEvent::ChannelDeclared(descriptor) => {
+            assert_eq!(descriptor.id, channel);
+            assert_eq!(descriptor.name, "runtime.json");
+        }
+        other => panic!("expected channel declaration, got {other:?}"),
+    }
+}
+
+#[test]
+fn submit_text_owned_queues_owned_string() {
     let endpoint = endpoint();
     let producer = endpoint.producer();
     let log = producer.register_channel("runtime.log", ChannelContent::TextStream);
     let sub = endpoint.subscribe_all("test");
 
-    producer.set_frame_timing_enabled(true);
-    let data_position = producer.submit_text(log, "visible");
+    assert!(producer.submit_text_owned(log, String::from("hello")));
     let tick = endpoint.tick();
 
-    assert_eq!(data_position, Position(0));
-    assert_eq!(tick.drained, 2);
-    let events = sub.drain_available();
-    assert_eq!(events.len(), 2);
-    assert_eq!(frame_event(&events[0]).channel.channel, log);
-    assert_eq!(frame_event(&events[0]).payload, b"visible");
-    assert_eq!(
-        frame_event(&events[1]).channel.channel,
-        FRAME_TIME_CHANNEL_ID
-    );
-    let sample = FrameTimeSample::decode(&frame_event(&events[1]).payload).expect("timing decodes");
-    assert_eq!(sample.target_position, data_position.0);
+    assert_eq!(tick.drained, 1);
+    assert_eq!(tick.delivered, 1);
+    let event = sub.recv_timeout(Duration::from_millis(50)).unwrap();
+    let delivery = frame_event(&event);
+    assert_eq!(delivery.channel.channel, log);
+    assert_eq!(delivery.payload, b"hello");
 }
 
 #[test]
 fn process_observer_adapter_submits_configured_channels() {
     let endpoint = endpoint();
     let producer = endpoint.producer();
-    producer.set_frame_timing_enabled(false);
     let stdout = producer.register_channel("proc.trainer.stdout", ChannelContent::TextStream);
     let stderr = producer.register_channel("proc.trainer.stderr", ChannelContent::TextStream);
     let observer = producer.process_observer_with(
@@ -300,8 +321,10 @@ fn stats_hook_adapter_submits_worker_snapshot_json() {
 }
 
 fn positions(events: &[DatastreamEvent]) -> Vec<u64> {
-    events
+    let mut positions: Vec<u64> = events
         .iter()
         .map(|event| frame_event(event).position.0)
-        .collect()
+        .collect();
+    positions.sort_unstable();
+    positions
 }
diff --git a/crates/datastream/tests/t_datastream_realio.rs b/crates/datastream/tests/t_datastream_realio.rs
index 15c478f..8df84e5 100644
--- a/crates/datastream/tests/t_datastream_realio.rs
+++ b/crates/datastream/tests/t_datastream_realio.rs
@@ -28,7 +28,6 @@ impl Record for ResourceSample {
 fn build_stream() -> (StreamId, Vec<Frame>) {
     let id = StreamId::new(NodeId::new("node-real"), Lifetime(1));
     let mux = Mux::unbounded(id.clone());
-    mux.set_frame_timing_enabled(false);
     for tick in 0..20 {
         mux.submit(
             RESOURCE_CHANNEL,
diff --git a/crates/distribution/tests/telemetry.rs b/crates/distribution/tests/telemetry.rs
index edec249..6ede5a9 100644
--- a/crates/distribution/tests/telemetry.rs
+++ b/crates/distribution/tests/telemetry.rs
@@ -14,7 +14,7 @@ mod datastream_records {
     //! the mux.
 
     use datastream::frame::{Lifetime, NodeId, StreamId};
-    use datastream::{Mux, Record};
+    use datastream::{ChannelId, Mux, Position, Record};
     use distribution::telemetry::{
         CacheEntryRec, DIST_STATE, DistributionState, MembershipTransition, RegistryEntryRec,
     };
@@ -49,18 +49,18 @@ mod datastream_records {
     fn distribution_emits_owned_channel_through_datastream_mux() {
         let stream = StreamId::new(NodeId::new("dist-node"), Lifetime(1));
         let mux = Mux::unbounded(stream);
-        mux.set_frame_timing_enabled(false);
         let state = DistributionState {
             registry_size: 9,
             ..Default::default()
         };
 
-        let pos = mux.submit(DistributionState::channel(), state.encode());
+        let channel = ChannelId(1);
+        assert!(mux.submit(channel, state.encode()));
         let frames = mux.drain();
 
-        assert_eq!(pos.0, 0);
         assert_eq!(frames.len(), 1);
-        assert_eq!(frames[0].channel.as_str(), DIST_STATE);
+        assert_eq!(frames[0].channel, channel);
+        assert_eq!(frames[0].position, Position(0));
         assert_eq!(
             DistributionState::decode(&frames[0].payload)
                 .unwrap()