fix(conversation): Keep chat responses a newer jp wrote

.config/rustqual/baseline-dry.json

@@ -48,32 +48,32 @@
     {
       "name": "ConversationStream::pop_while",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 816
+      "line": 827
     },
     {
       "name": "ConversationStream::remove_orphaned_inquiry_responses",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 995
+      "line": 1018
     },
     {
       "name": "ConversationStream::remove_orphaned_inquiry_requests",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1037
+      "line": 1060
     },
     {
       "name": "ConversationStream::normalize_turn_starts",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1084
+      "line": 1107
     },
     {
       "name": "ConversationStream::sanitize_orphaned_tool_calls",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1171
+      "line": 1194
     },
     {
       "name": "ConversationStream::from_legacy_events",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1580
+      "line": 1603
     },
     {
       "name": "strip_schema",
@@ -153,12 +153,12 @@
     {
       "name": "EventPayload::deserialize",
       "file": "crates/jp_conversation/src/stream/entry.rs",
-      "line": 152
+      "line": 161
     },
     {
       "name": "StoredEvent::deserialize",
       "file": "crates/jp_conversation/src/stream/entry.rs",
-      "line": 305
+      "line": 346
     },
     {
       "name": "EventId::random",

.config/rustqual/baseline.json

@@ -1,13 +1,13 @@
 {
   "version": 2,
-  "quality_score": 0.7058772822297027,
-  "iosp_score": 0.9146377525874815,
+  "quality_score": 0.7060539191242928,
+  "iosp_score": 0.9147302087435999,
   "violations": 866,
-  "total": 13997,
+  "total": 14012,
   "complexity_warnings": 163,
   "magic_number_warnings": 238,
   "nesting_depth_warnings": 27,
-  "function_length_warnings": 433,
+  "function_length_warnings": 434,
   "unsafe_warnings": 74,
   "error_handling_warnings": 194,
   "duplicate_groups": 0,
@@ -22,13 +22,13 @@
   "coupling_warnings": 0,
   "coupling_cycles": 0,
   "tq_no_assertion_warnings": 0,
-  "tq_no_sut_warnings": 19,
+  "tq_no_sut_warnings": 20,
   "tq_untested_warnings": 95,
   "tq_uncovered_warnings": 0,
   "tq_untested_logic_warnings": 0,
   "structural_srp_warnings": 48,
   "structural_coupling_warnings": 8,
-  "total_findings": 2819,
+  "total_findings": 2821,
   "violation_details": [
     {
       "name": "ConversationStream::config_unsets",
@@ -48,32 +48,32 @@
     {
       "name": "ConversationStream::pop_while",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 816
+      "line": 827
     },
     {
       "name": "ConversationStream::remove_orphaned_inquiry_responses",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 995
+      "line": 1018
     },
     {
       "name": "ConversationStream::remove_orphaned_inquiry_requests",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1037
+      "line": 1060
     },
     {
       "name": "ConversationStream::normalize_turn_starts",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1084
+      "line": 1107
     },
     {
       "name": "ConversationStream::sanitize_orphaned_tool_calls",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1171
+      "line": 1194
     },
     {
       "name": "ConversationStream::from_legacy_events",
       "file": "crates/jp_conversation/src/stream.rs",
-      "line": 1580
+      "line": 1603
     },
     {
       "name": "strip_schema",
@@ -153,12 +153,12 @@
     {
       "name": "EventPayload::deserialize",
       "file": "crates/jp_conversation/src/stream/entry.rs",
-      "line": 152
+      "line": 161
     },
     {
       "name": "StoredEvent::deserialize",
       "file": "crates/jp_conversation/src/stream/entry.rs",
-      "line": 305
+      "line": 346
     },
     {
       "name": "EventId::random",

crates/jp_cli/src/cmd/query_tests.rs

@@ -3176,6 +3176,52 @@ fn pending_trim_pop_request_removes_stale_turn_start() {
     assert_no_adjacent_turn_starts(&stream);
 }
 
+/// `--replay` on a turn holding a chat response this build cannot read (one a
+/// newer jp wrote) removes it with the rest of the turn, instead of leaving it
+/// behind to be read as part of the turn before.
+#[test]
+fn pending_trim_replay_removes_an_unreadable_chat_response_with_its_turn() {
+    let mut stream = ConversationStream::new_test();
+    stream.start_turn("first question");
+    stream.push_event(ConversationEvent::now(ChatResponse::message(
+        "first answer",
+    )));
+    stream.start_turn("second question");
+    stream.push_event(ConversationEvent::now(ChatResponse::message(
+        "second answer",
+    )));
+
+    let (base_config, mut events) = stream.to_parts().unwrap();
+    let answer = events
+        .iter()
+        .position(|e| e.get("message").and_then(Value::as_str) == Some("second answer"))
+        .unwrap();
+    events.insert(
+        answer,
+        serde_json::json!({
+            "type": "chat_response",
+            "timestamp": "2025-01-01 00:00:00.0",
+            "commentary": "Checking the second question."
+        }),
+    );
+    let mut stream =
+        ConversationStream::from_parts(base_config, events, &PartialAppConfig::empty()).unwrap();
+
+    PendingStreamTrim {
+        replay_turn: true,
+        pop_request: false,
+    }
+    .apply(&mut stream);
+    stream.start_turn("second question, revised");
+
+    let (_, events) = stream.to_parts().unwrap();
+    assert!(
+        !events.iter().any(|e| e.get("commentary").is_some()),
+        "the replayed turn's unreadable chat response survived: {events:#?}"
+    );
+    assert_eq!(turn_start_count(&stream), 2);
+}
+
 #[test]
 fn pending_trim_default_is_noop() {
     let mut stream = ConversationStream::new_test();

crates/jp_conversation/src/event/chat.rs

@@ -107,6 +107,18 @@ pub enum ChatResponse {
 }
 
 impl ChatResponse {
+    /// The content key each variant is stored under, one per variant.
+    ///
+    /// `ChatResponse` is untagged, so the key that is present selects the
+    /// variant.
+    /// A stored `chat_response` holding none of these keys was written by a
+    /// newer build, and the conversation stream deserializer keeps it verbatim
+    /// instead of failing the load.
+    /// Adding a variant means adding its key here too;
+    /// `every_variant_serializes_with_a_listed_content_key` catches a forgotten
+    /// entry.
+    pub(crate) const CONTENT_KEYS: &'static [&'static str] = &["message", "reasoning", "data"];
+
     /// Creates a new message response.
     #[must_use]
     pub fn message(content: impl Into<String>) -> Self {

crates/jp_conversation/src/event/chat_tests.rs

@@ -61,6 +61,35 @@ fn untagged_deserialization_distinguishes_variants() {
     assert!(structured.is_structured());
 }
 
+#[test]
+fn every_variant_serializes_with_a_listed_content_key() {
+    let responses = [
+        ChatResponse::message("hello"),
+        ChatResponse::reasoning("let me think"),
+        ChatResponse::structured(json!({"key": "value"})),
+    ];
+
+    for response in responses {
+        // Exhaustive, so a new variant stops compiling here until it has a
+        // sample above and a key in `CONTENT_KEYS`.
+        let key = match &response {
+            ChatResponse::Message { .. } => "message",
+            ChatResponse::Reasoning { .. } => "reasoning",
+            ChatResponse::Structured { .. } => "data",
+        };
+
+        let json = serde_json::to_value(&response).unwrap();
+        assert!(
+            json.get(key).is_some(),
+            "{response:?} is not stored under {key:?}"
+        );
+        assert!(
+            ChatResponse::CONTENT_KEYS.contains(&key),
+            "{key:?} is missing from CONTENT_KEYS"
+        );
+    }
+}
+
 #[test]
 fn structured_within_event_kind_roundtrip() {
     let kind = EventKind::ChatResponse(ChatResponse::structured(json!([1, 2, 3])));

crates/jp_conversation/src/stream.rs

@@ -223,7 +223,7 @@ impl ConversationStream {
             EventPayload::Event(_)
             | EventPayload::Compaction(_)
             | EventPayload::Overlay(_)
-            | EventPayload::Unknown(_) => None,
+            | EventPayload::Unknown { .. } => None,
         });
 
         for delta in iter {
@@ -746,21 +746,30 @@ impl ConversationStream {
     /// Pops the last [`ConversationEvent`] from the stream, returning it
     /// wrapped in a [`ConversationEventWithConfig`], containing the
     /// [`PartialAppConfig`] at the time the event was added.
+    ///
+    /// Turn entries this build cannot read that follow the popped event (a chat
+    /// response a newer jp wrote in reply to it) are removed with it.
     #[must_use]
     pub fn pop(&mut self) -> Option<ConversationEventWithConfig> {
-        // The last conversation event, ignoring any trailing overlays
+        // The event `last_turn_event` returns, ignoring any trailing overlays
         // (`ConfigDelta`/`Compaction`) that follow it. Those overlays are left
         // in place rather than discarded — a `Compaction` references turn
         // ranges and applies regardless of tail position.
-        let pos = self
-            .events
-            .iter()
-            .rposition(|e| e.scope() == EventScope::Turn)?;
+        let pos = self.events.iter().rposition(|e| e.as_event().is_some())?;
 
         let config = self
             .last()
             .map_or_else(|| self.base_config.to_partial(), |v| v.config);
 
+        // Any turn-scoped entry after `pos` is unreadable. Left behind, it
+        // would read as part of the turn before once the popped event is gone.
+        let mut index = 0;
+        self.events.retain(|e| {
+            let keep = index <= pos || e.scope() == EventScope::Global;
+            index += 1;
+            keep
+        });
+
         let internal = self.events.remove(pos);
         let event_id = internal.event_id.clone();
         internal
@@ -831,6 +840,12 @@ impl ConversationStream {
     /// [`ConfigDelta`]s are always preserved: they apply to the conversation as
     /// a whole, regardless of position.
     ///
+    /// A turn entry this build cannot read (a chat response a newer jp wrote)
+    /// has no event to pass to `f`.
+    /// It is kept or dropped with the readable event before it, which shares
+    /// its turn because a chat response never opens one; with no event before
+    /// it, it is kept.
+    ///
     /// [`Compaction`] overlays are dropped **selectively** when this call
     /// removes turn-scoped events.
     /// An overlay whose turn range lies entirely before the earliest removed
@@ -847,17 +862,26 @@ impl ConversationStream {
     ///
     /// [RFD 097]: https://jp.computer/rfd/097
     pub fn retain(&mut self, mut f: impl FnMut(&ConversationEvent) -> bool) {
+        let mut previous = true;
+        let mut decide = |entry: &InternalEvent| match entry.scope() {
+            EventScope::Global => true,
+            EventScope::Turn => match entry.as_event() {
+                Some(event) => {
+                    previous = f(event);
+                    previous
+                }
+                None => previous,
+            },
+        };
+
         // Fast path: with no overlays present there's nothing to invalidate, so
         // skip the turn-index bookkeeping.
         if !self
             .events
             .iter()
             .any(|e| matches!(&e.payload, EventPayload::Compaction(_)))
         {
-            self.events.retain(|event| match event.scope() {
-                EventScope::Global => true,
-                EventScope::Turn => event.as_event().is_some_and(&mut f),
-            });
+            self.events.retain(|event| decide(event));
             return;
         }
 
@@ -867,10 +891,7 @@ impl ConversationStream {
         let mut first_removed_turn: Option<usize> = None;
         let mut index = 0;
         self.events.retain(|event| {
-            let keep = match event.scope() {
-                EventScope::Global => true,
-                EventScope::Turn => event.as_event().is_some_and(&mut f),
-            };
+            let keep = decide(event);
             if !keep {
                 let turn = turn_indices[index];
                 first_removed_turn = Some(first_removed_turn.map_or(turn, |m| m.min(turn)));
@@ -954,7 +975,7 @@ impl ConversationStream {
                 EventPayload::ConfigDelta(_)
                 | EventPayload::Compaction(_)
                 | EventPayload::Overlay(_)
-                | EventPayload::Unknown(_) => true,
+                | EventPayload::Unknown { .. } => true,
                 EventPayload::Event(e) => e.is_turn_start(),
             }
         });