session: an unreadable event batch is not the end of the stream

take_events kept returning Option<EventBatchView> and defaulting through the
question-mark operator, so a batch missing a field read as end of stream and the
ConsumerEvents enum added in the previous round described nothing. It returns
Batch or Unreadable now, and a test publishes a batch with no droppedBefore,
asserts it is reported as unreadable naming the field, and asserts the next real
batch still reads.

The checkpoint-envelope-v1 amendment cites the rule that lets a required
manifest field land with envelopeVersion still 1 while no production file
exists.

docs/design/session-framework/checkpoint-envelope-v1.md

@@ -111,7 +111,9 @@ Without it a replacement fly that built another graph -- the same dataset, the s
 count, another index -- passed the group check and was then published under its predecessor's
 `indexDigest`, which is a graph identity crossing a recovery and exactly what section 5's rules
 exist to prevent. It is recorded from the worker's attestation rather than recomputed from the
-dataset, because the point is that the two can disagree.
+dataset, because the point is that the two can disagree. `envelopeVersion` stays `1`, which the
+required-manifest-field rule below allows only while no production `FLYSESS1` file exists; once
+one does, adding a required manifest field must bump it.
 
 **Amendment, 2026-09-22 (STATE-01).** The table above names a holder for every payload except
 the environment's own, although section 6's fixture has one (`world`) and a group install has

services/flysim/crates/fly-session/src/publish.rs

@@ -1275,26 +1275,42 @@ impl PresentationConsumer {
     }
 
     /// Takes the next bounded event batch.
-    pub async fn take_events(&mut self) -> Option<EventBatchView> {
+    ///
+    /// `None` is the end of the stream and nothing else. A batch this consumer cannot read
+    /// comes back as [`ConsumerEvents::Unreadable`], because "there are no more events" and
+    /// "that one made no sense" are different facts and one value cannot carry both: a
+    /// consumer that saw the second as the first would stop reading a live stream.
+    pub async fn take_events(&mut self) -> Option<ConsumerEvents> {
         let message = self.events.next().await?;
         let payload = message.payload().clone();
         drop(message);
-        let epoch = payload.get("epoch").and_then(Value::as_str).map(id)?;
+        let unreadable = |what: &str| {
-        let dropped_before = payload
+            Some(ConsumerEvents::Unreadable {
+                detail: format!("an event batch has no readable {what}"),
+            })
+        };
+        let Some(epoch) = payload.get("epoch").and_then(Value::as_str).map(id) else {
+            return unreadable("epoch");
+        };
+        let Some(dropped_before) = payload
             .get("droppedBefore")
             .and_then(Value::as_str)
-            .and_then(|t| t.parse::<u64>().ok())?;
+            .and_then(|t| t.parse::<u64>().ok())
-        let event_ids = payload
+        else {
-            .get("events")
+            return unreadable("droppedBefore");
-            .and_then(Value::as_array)?
+        };
+        let Some(events) = payload.get("events").and_then(Value::as_array) else {
+            return unreadable("events array");
+        };
+        let event_ids = events
             .iter()
             .filter_map(|e| e.get("id").and_then(Value::as_str).map(str::to_owned))
             .collect();
-        Some(EventBatchView {
+        Some(ConsumerEvents::Batch(EventBatchView {
             epoch,
             dropped_before,
             event_ids,
-        })
+        }))
     }
 
     /// Records that this consumer has mapped geometry against the agents of `revision`.

services/flysim/crates/fly-session/tests/publishing.rs

@@ -14,8 +14,9 @@ use common::{Fixture, default_fixture, fixture, fly_a, fly_b, mode_fixture, with
 use fly_session::coordinator::{DESCRIPTOR_REVISION, Injections};
 use fly_session::harness::{AgentSpec, ExecutionMode, HarnessConfig, Via};
 use fly_session::publish::{
-    ApplicationChannel, ConsumerOutcome, Delivery, EVENT_BATCH_DEPTH, EventOutbox, GET_SNAPSHOT,
+    ApplicationChannel, ConsumerEvents, ConsumerOutcome, Delivery, EVENT_BATCH_DEPTH, EventOutbox,
-    PresentationConsumer, PublicationOutcome, TopicPolicy, check_publication, query_service,
+    GET_SNAPSHOT, PresentationConsumer, PublicationOutcome, TopicPolicy, check_publication,
+    query_service,
 };
 use fly_session::types::*;
 use serde_json::json;
@@ -38,6 +39,7 @@ both_transports!(
     the_query_service_answers_reads_and_nothing_else,
     a_refused_snapshot_is_still_what_the_repair_path_answers,
     a_replacement_that_built_another_index_cannot_install_the_checkpoint,
+    a_malformed_event_batch_is_unreadable_and_not_the_end_of_the_stream,
     the_published_descriptor_is_what_the_workers_attested_to,
     a_stimulus_kind_the_descriptor_does_not_declare_is_refused,
     a_restored_boundary_publishes_a_new_revision_and_no_transition,
@@ -1079,6 +1081,55 @@ async fn a_refused_snapshot_is_still_what_the_repair_path_answers(via: Via) {
     f.shutdown().await;
 }
 
+/// An event batch a consumer cannot read is reported as unreadable, and the stream goes on.
+///
+/// The two facts differ: a consumer that read "that one made no sense" as "there are no more
+/// events" would stop reading a live stream, which is the silent default this module exists to
+/// refuse. The malformed batch is published by a second publisher on the session's own address,
+/// because a well-behaved session cannot produce one.
+async fn a_malformed_event_batch_is_unreadable_and_not_the_end_of_the_stream(via: Via) {
+    let mut f = started(via).await;
+    let mut consumer = f.harness.consumer().await.expect("a consumer attaches");
+    let events_topic = f.harness.coordinator.topics().events.clone();
+    let intruder = f.harness.publisher().await.expect("a publishing client");
+
+    // A batch with no droppedBefore: readable JSON, unreadable as a batch.
+    intruder
+        .publish(
+            &events_topic,
+            object(json!({
+                "sessionId": f.harness.config.session_id.as_str(),
+                "epoch": f.harness.config.epoch.as_str(),
+                "events": [],
+            })),
+            &[],
+        )
+        .await
+        .expect("the malformed batch publishes");
+
+    match within("the malformed batch", consumer.take_events()).await {
+        Some(ConsumerEvents::Unreadable { detail }) => {
+            assert!(
+                detail.contains("droppedBefore"),
+                "the report names what it could not read: {detail}"
+            );
+        }
+        other => panic!("a malformed batch must not read as {other:?}"),
+    }
+
+    // The stream did not end: the next real batch still arrives and reads.
+    f.harness.coordinator.run(1).await.expect("a transition");
+    match within("the next batch", consumer.take_events()).await {
+        Some(ConsumerEvents::Batch(batch)) => {
+            assert_eq!(batch.epoch, f.harness.config.epoch);
+            assert!(!batch.event_ids.is_empty(), "the transition produced events");
+        }
+        other => panic!("expected a readable batch after the malformed one, got {other:?}"),
+    }
+    intruder.close().await;
+    f.shutdown().await;
+}
+
 /// Two reads and nothing else. There is no method on this service that could move anything.
 async fn the_query_service_answers_reads_and_nothing_else(via: Via) {
     let mut f = started(via).await;