From dc1d295920e520cd4c88289a1750b4efd53d1ae2 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Wed, 14 Oct 2026 16:32:53 +0200 Subject: [PATCH 1/4] fix(conversation): Keep chat responses a newer jp wrote A conversation holding a chat response variant this build does not know can no longer fail to load. `ChatResponse` is untagged, so a new variant is stored under the known `chat_response` type with a new content key, and the unknown-event preservation from #778 never saw it: the entry failed to deserialize and the workspace skipped the whole conversation. A `chat_response` holding none of the known content keys (`message`, `reasoning`, `data`) is now kept verbatim and hidden, like an unknown event type, and written back unchanged on save. A known key with a malformed value still fails the load. This ships ahead of the assistant commentary work, which adds a `commentary` chat response, so builds have time to pick it up before anything writes one. Signed-off-by: Jean Mertz --- crates/jp_conversation/src/event/chat.rs | 12 ++++ .../jp_conversation/src/event/chat_tests.rs | 29 +++++++++ crates/jp_conversation/src/stream/entry.rs | 23 ++++++- crates/jp_conversation/src/stream_tests.rs | 62 +++++++++++++++++++ 4 files changed, 125 insertions(+), 1 deletion(-) diff --git a/crates/jp_conversation/src/event/chat.rs b/crates/jp_conversation/src/event/chat.rs index 32d0de352..fc0a7c7a8 100644 --- a/crates/jp_conversation/src/event/chat.rs +++ b/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) -> Self { diff --git a/crates/jp_conversation/src/event/chat_tests.rs b/crates/jp_conversation/src/event/chat_tests.rs index 6037a4cda..50bc915ac 100644 --- a/crates/jp_conversation/src/event/chat_tests.rs +++ b/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]))); diff --git a/crates/jp_conversation/src/stream/entry.rs b/crates/jp_conversation/src/stream/entry.rs index e8a695bab..c67025477 100644 --- a/crates/jp_conversation/src/stream/entry.rs +++ b/crates/jp_conversation/src/stream/entry.rs @@ -26,7 +26,7 @@ use tracing::warn; use super::config_delta::{self, ConfigDelta}; use crate::{ Compaction, EventId, EventOverlay, - event::{ConversationEvent, EventKind}, + event::{ChatResponse, ConversationEvent, EventKind}, storage::{decode_event_value, encode_event}, }; @@ -193,6 +193,27 @@ impl<'de> Deserialize<'de> for EventPayload { return Ok(Self::Unknown(value)); } + // `ChatResponse` is untagged: its content key selects the variant, so a + // variant added by a newer jp arrives under a known `type` tag. One + // with no known content key is preserved the same way. + if tag == "chat_response" + && !ChatResponse::CONTENT_KEYS + .iter() + .any(|key| value.get(key).is_some()) + { + #[cfg(debug_assertions)] + { + let mut probe = value.clone(); + decode_event_value(&mut probe); + debug_assert!( + serde_json::from_value::(probe).is_err(), + "a ChatResponse content key is missing from ChatResponse::CONTENT_KEYS", + ); + } + warn!("Unknown chat response variant; preserving raw event."); + return Ok(Self::Unknown(value)); + } + // Decode base64-encoded storage fields before deserializing. decode_event_value(&mut value); diff --git a/crates/jp_conversation/src/stream_tests.rs b/crates/jp_conversation/src/stream_tests.rs index d26968b89..7e387a604 100644 --- a/crates/jp_conversation/src/stream_tests.rs +++ b/crates/jp_conversation/src/stream_tests.rs @@ -3121,6 +3121,68 @@ fn test_from_parts_tolerates_unknown_event_kind() { ); } +#[test] +fn test_internal_event_preserves_unknown_chat_response_verbatim() { + // A chat response written by a newer jp, selected by a content key this + // build doesn't know. + let raw = serde_json::json!({ + "event_id": "future1", + "type": "chat_response", + "timestamp": "2025-01-01 00:00:00.0", + "commentary": "Found it. Editing the parser next.", + "metadata": { "foo": "YmFy" } + }); + + let internal = serde_json::from_value::(raw.clone()) + .unwrap() + .into_entry(); + assert!(matches!(&internal.payload, EventPayload::Unknown(_))); + + let serialized = serde_json::to_value(&internal).unwrap(); + assert_eq!(serialized, raw); +} + +#[test] +fn test_from_parts_tolerates_unknown_chat_response() { + let mut stream = ConversationStream::new_test(); + stream.start_turn(ChatRequest::from("hello")); + + let (base_config, mut events) = stream.to_parts().unwrap(); + + let unknown = serde_json::json!({ + "event_id": "future1", + "type": "chat_response", + "timestamp": "2025-01-01 00:01:00.0", + "commentary": "Found it. Editing the parser next." + }); + events.push(unknown.clone()); + + let result = + ConversationStream::from_parts(base_config, events, &PartialAppConfig::empty()).unwrap(); + + // TurnStart + ChatRequest; the unknown chat response is not an event here. + assert_eq!(result.len(), 2); + + let (_, round_tripped) = result.to_parts().unwrap(); + assert!( + round_tripped.contains(&unknown), + "unknown chat response must round-trip through to_parts" + ); +} + +#[test] +fn test_malformed_known_chat_response_fails_deserialization() { + // A known content key with a value of the wrong type is corruption, not a + // newer format, and keeps failing loudly. + let raw = serde_json::json!({ + "type": "chat_response", + "timestamp": "2025-01-01 00:00:00.0", + "message": 5 + }); + + assert!(serde_json::from_value::(raw).is_err()); +} + /// Characterization test: extending an empty stream from a source stream /// reproduces the source's observed iter sequence and serialized shape. /// From f8f23b83aca5c2b0a77f2b35053bbbabd26883f5 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 27 Oct 2026 10:05:20 +0100 Subject: [PATCH 2/4] rustqual Signed-off-by: Jean Mertz --- .config/rustqual/baseline-dry.json | 18 +++++++++--------- .config/rustqual/baseline.json | 26 +++++++++++++------------- 2 files changed, 22 insertions(+), 22 deletions(-) diff --git a/.config/rustqual/baseline-dry.json b/.config/rustqual/baseline-dry.json index 40f69566d..ed3e0d5b8 100644 --- a/.config/rustqual/baseline-dry.json +++ b/.config/rustqual/baseline-dry.json @@ -1,7 +1,7 @@ { "version": 2, - "quality_score": 0.6522887459652152, - "iosp_score": 0.8123249299719888, + "quality_score": 0.6522264572566293, + "iosp_score": 0.8122844827586206, "violations": 871, "total": 6512, "complexity_warnings": 0, @@ -48,32 +48,32 @@ { "name": "ConversationStream::pop_while", "file": "crates/jp_conversation/src/stream.rs", - "line": 816 + "line": 818 }, { "name": "ConversationStream::remove_orphaned_inquiry_responses", "file": "crates/jp_conversation/src/stream.rs", - "line": 995 + "line": 997 }, { "name": "ConversationStream::remove_orphaned_inquiry_requests", "file": "crates/jp_conversation/src/stream.rs", - "line": 1037 + "line": 1039 }, { "name": "ConversationStream::normalize_turn_starts", "file": "crates/jp_conversation/src/stream.rs", - "line": 1084 + "line": 1086 }, { "name": "ConversationStream::sanitize_orphaned_tool_calls", "file": "crates/jp_conversation/src/stream.rs", - "line": 1171 + "line": 1173 }, { "name": "ConversationStream::from_legacy_events", "file": "crates/jp_conversation/src/stream.rs", - "line": 1580 + "line": 1582 }, { "name": "strip_schema", @@ -158,7 +158,7 @@ { "name": "StoredEvent::deserialize", "file": "crates/jp_conversation/src/stream/entry.rs", - "line": 305 + "line": 326 }, { "name": "EventId::random", diff --git a/.config/rustqual/baseline.json b/.config/rustqual/baseline.json index 67757232b..74bb44824 100644 --- a/.config/rustqual/baseline.json +++ b/.config/rustqual/baseline.json @@ -1,13 +1,13 @@ { "version": 2, - "quality_score": 0.7058772822297027, - "iosp_score": 0.9146377525874815, + "quality_score": 0.7058212992056335, + "iosp_score": 0.9146545777076969, "violations": 866, - "total": 13997, + "total": 14002, "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": 818 }, { "name": "ConversationStream::remove_orphaned_inquiry_responses", "file": "crates/jp_conversation/src/stream.rs", - "line": 995 + "line": 997 }, { "name": "ConversationStream::remove_orphaned_inquiry_requests", "file": "crates/jp_conversation/src/stream.rs", - "line": 1037 + "line": 1039 }, { "name": "ConversationStream::normalize_turn_starts", "file": "crates/jp_conversation/src/stream.rs", - "line": 1084 + "line": 1086 }, { "name": "ConversationStream::sanitize_orphaned_tool_calls", "file": "crates/jp_conversation/src/stream.rs", - "line": 1171 + "line": 1173 }, { "name": "ConversationStream::from_legacy_events", "file": "crates/jp_conversation/src/stream.rs", - "line": 1580 + "line": 1582 }, { "name": "strip_schema", @@ -158,7 +158,7 @@ { "name": "StoredEvent::deserialize", "file": "crates/jp_conversation/src/stream/entry.rs", - "line": 305 + "line": 326 }, { "name": "EventId::random", From 601ed922a7f9f222ab6a8878c6b297daae126ac4 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 27 Oct 2026 10:56:50 +0100 Subject: [PATCH 3/4] fix(conversation): Prune unreadable chat responses with their turn A chat response this build cannot read now stays in the turn it came from. It was preserved as a global entry, so `jp query --replay` removed the rest of the replayed turn and left the response behind, attached to the turn before it, and `jp c fork` with a turn selection carried it out of turns it dropped. A build that understands the response would later read it as part of the wrong turn. Such an entry is now turn-scoped. Replay drops it with its turn, turn filtering keeps or drops it with the readable event before it, and popping the last request also removes an unreadable response that followed it. Entries of an unknown `type` stay global, since nothing says which turn they belong to. Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd/query_tests.rs | 46 +++++ crates/jp_conversation/src/stream.rs | 51 ++++-- crates/jp_conversation/src/stream/entry.rs | 52 ++++-- .../src/stream/event_id_tests.rs | 2 +- crates/jp_conversation/src/stream/iter.rs | 8 +- .../jp_conversation/src/stream/projection.rs | 4 +- crates/jp_conversation/src/stream_tests.rs | 161 +++++++++++++++++- 7 files changed, 284 insertions(+), 40 deletions(-) diff --git a/crates/jp_cli/src/cmd/query_tests.rs b/crates/jp_cli/src/cmd/query_tests.rs index eee303241..758d49891 100644 --- a/crates/jp_cli/src/cmd/query_tests.rs +++ b/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(); diff --git a/crates/jp_conversation/src/stream.rs b/crates/jp_conversation/src/stream.rs index 37fa35e7a..877dcce46 100644 --- a/crates/jp_conversation/src/stream.rs +++ b/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 { - // 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,6 +862,18 @@ 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 @@ -854,10 +881,7 @@ impl ConversationStream { .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 = 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(), } }); diff --git a/crates/jp_conversation/src/stream/entry.rs b/crates/jp_conversation/src/stream/entry.rs index c67025477..ee3240593 100644 --- a/crates/jp_conversation/src/stream/entry.rs +++ b/crates/jp_conversation/src/stream/entry.rs @@ -74,15 +74,24 @@ pub(super) enum EventPayload { /// building the LLM request. /// Does not modify or delete any existing events. Overlay(EventOverlay), - /// An event whose `type` tag this build does not recognize. + /// An entry this build cannot read: an unrecognized `type` tag, or a chat + /// response with no content key this build knows. /// /// Conversations are an append-only log that a newer `jp` may have written. - /// Rather than fail the entire stream load on an unknown event kind, the - /// raw JSON is retained verbatim so it round-trips losslessly on the next - /// save. - /// Unknown events are invisible to event iteration, config resolution, and + /// Rather than fail the entire stream load, the raw JSON is retained + /// verbatim so it round-trips losslessly on the next save. + /// Unknown entries are invisible to event iteration, config resolution, and /// providers. - Unknown(Value), + Unknown { + /// The stored JSON, without the `event_id` the wrapper holds. + value: Value, + + /// Whether turn pruning removes the entry with its turn. + /// + /// A known `type` tag says which turn the entry belongs to; an unknown + /// one does not, so such an entry is global. + scope: EventScope, + }, } impl Serialize for EventPayload { @@ -141,7 +150,7 @@ impl Serialize for EventPayload { } .serialize(serializer) } - Self::Unknown(value) => value.serialize(serializer), + Self::Unknown { value, .. } => value.serialize(serializer), } } } @@ -190,12 +199,17 @@ impl<'de> Deserialize<'de> for EventPayload { ); } warn!(%tag, "Unknown conversation event kind; preserving raw event."); - return Ok(Self::Unknown(value)); + return Ok(Self::Unknown { + value, + scope: EventScope::Global, + }); } // `ChatResponse` is untagged: its content key selects the variant, so a // variant added by a newer jp arrives under a known `type` tag. One - // with no known content key is preserved the same way. + // with no known content key is preserved the same way, but stays + // turn-scoped: it answers the request before it, and pruning that turn + // must take it along. if tag == "chat_response" && !ChatResponse::CONTENT_KEYS .iter() @@ -211,7 +225,10 @@ impl<'de> Deserialize<'de> for EventPayload { ); } warn!("Unknown chat response variant; preserving raw event."); - return Ok(Self::Unknown(value)); + return Ok(Self::Unknown { + value, + scope: EventScope::Turn, + }); } // Decode base64-encoded storage fields before deserializing. @@ -231,10 +248,13 @@ impl<'de> Deserialize<'de> for EventPayload { /// Adding a new `EventPayload` variant forces a classification here: /// [`InternalEvent::scope`] is an exhaustive match, so no pruning caller can /// silently mistreat it. +/// +/// A turn-scoped entry is not always a readable event: a chat response a newer +/// jp wrote is [`EventPayload::Unknown`] with [`EventScope::Turn`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(super) enum EventScope { - /// Survives turn pruning: config deltas and compaction overlays apply to - /// the conversation regardless of position. + /// Survives turn pruning: config deltas, compaction and patch overlays, and + /// entries of an unknown kind apply regardless of position. Global, /// Belongs to a turn and is removed when that turn is pruned. Turn, @@ -249,7 +269,7 @@ impl InternalEvent { EventPayload::ConfigDelta(_) | EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => None, + | EventPayload::Unknown { .. } => None, } } @@ -261,7 +281,7 @@ impl InternalEvent { EventPayload::ConfigDelta(_) | EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => None, + | EventPayload::Unknown { .. } => None, } } @@ -272,8 +292,8 @@ impl InternalEvent { match &self.payload { EventPayload::ConfigDelta(_) | EventPayload::Compaction(_) - | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => EventScope::Global, + | EventPayload::Overlay(_) => EventScope::Global, + EventPayload::Unknown { scope, .. } => *scope, EventPayload::Event(_) => EventScope::Turn, } } diff --git a/crates/jp_conversation/src/stream/event_id_tests.rs b/crates/jp_conversation/src/stream/event_id_tests.rs index a79af716c..cfeb4e303 100644 --- a/crates/jp_conversation/src/stream/event_id_tests.rs +++ b/crates/jp_conversation/src/stream/event_id_tests.rs @@ -118,7 +118,7 @@ fn an_unknown_entry_keeps_its_id_on_the_wrapper_only() { let raw = r#"{"event_id":"Future ID!","type":"future_event","nested":{"event_id":"not-the-wrapper"}}"#; let entry = from_str::(raw).unwrap().into_entry(); - let EventPayload::Unknown(payload) = &entry.payload else { + let EventPayload::Unknown { value: payload, .. } = &entry.payload else { panic!("expected an unknown payload"); }; assert_eq!(entry.event_id, EventId::fixed("Future ID!")); diff --git a/crates/jp_conversation/src/stream/iter.rs b/crates/jp_conversation/src/stream/iter.rs index 508f4d181..95162f7d2 100644 --- a/crates/jp_conversation/src/stream/iter.rs +++ b/crates/jp_conversation/src/stream/iter.rs @@ -125,7 +125,7 @@ impl Iterator for IntoIter { } EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => {} + | EventPayload::Unknown { .. } => {} } } } @@ -140,7 +140,7 @@ impl DoubleEndedIterator for IntoIter { EventPayload::ConfigDelta(_) | EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => { + | EventPayload::Unknown { .. } => { // A delta/compaction at the very end of the list affects // nothing that follows it, and it doesn't affect previous // items. We simply discard it. @@ -211,7 +211,7 @@ impl<'a> Iterator for Iter<'a> { } EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => {} + | EventPayload::Unknown { .. } => {} } } @@ -280,7 +280,7 @@ impl<'a> Iterator for IterMut<'a> { } EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => {} + | EventPayload::Unknown { .. } => {} } } diff --git a/crates/jp_conversation/src/stream/projection.rs b/crates/jp_conversation/src/stream/projection.rs index 24b1df2cf..833394d35 100644 --- a/crates/jp_conversation/src/stream/projection.rs +++ b/crates/jp_conversation/src/stream/projection.rs @@ -211,7 +211,7 @@ pub(super) fn apply(events: &mut Vec, event_ids: &mut EventIds) - // Config deltas carry global state; unknown (forward-compat) events // are opaque. Both pass through projection verbatim — the iterators // skip unknown events, so they stay invisible to providers. - EventPayload::ConfigDelta(_) | EventPayload::Unknown(_) => { + EventPayload::ConfigDelta(_) | EventPayload::Unknown { .. } => { projected.push(event); event_origins.push(TurnOrigin::Kept(turn)); } @@ -493,7 +493,7 @@ pub(super) fn assign_turn_indices(events: &[InternalEvent]) -> Vec { EventPayload::ConfigDelta(_) | EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => { + | EventPayload::Unknown { .. } => { indices.push(turn); } } diff --git a/crates/jp_conversation/src/stream_tests.rs b/crates/jp_conversation/src/stream_tests.rs index 7e387a604..08976016f 100644 --- a/crates/jp_conversation/src/stream_tests.rs +++ b/crates/jp_conversation/src/stream_tests.rs @@ -3083,7 +3083,10 @@ fn test_internal_event_preserves_unknown_kind_verbatim() { let internal = serde_json::from_value::(raw.clone()) .unwrap() .into_entry(); - assert!(matches!(&internal.payload, EventPayload::Unknown(_))); + assert!(matches!(&internal.payload, EventPayload::Unknown { + scope: EventScope::Global, + .. + })); // The payload round-trips byte-for-byte: no decode/re-encode is applied, // so the newer jp reads back exactly what it wrote. @@ -3136,7 +3139,10 @@ fn test_internal_event_preserves_unknown_chat_response_verbatim() { let internal = serde_json::from_value::(raw.clone()) .unwrap() .into_entry(); - assert!(matches!(&internal.payload, EventPayload::Unknown(_))); + assert!(matches!(&internal.payload, EventPayload::Unknown { + scope: EventScope::Turn, + .. + })); let serialized = serde_json::to_value(&internal).unwrap(); assert_eq!(serialized, raw); @@ -3183,6 +3189,157 @@ fn test_malformed_known_chat_response_fails_deserialization() { assert!(serde_json::from_value::(raw).is_err()); } +/// A `commentary` chat response as a newer jp stores it; this build cannot read +/// it. +fn future_commentary(text: &str) -> Value { + serde_json::json!({ + "type": "chat_response", + "timestamp": "2025-01-01 00:00:00.0", + "commentary": text + }) +} + +/// Reload `stream` from storage with a future commentary entry inserted +/// directly before each message named in `before_messages`. +fn reload_with_commentary( + stream: &ConversationStream, + before_messages: &[&str], +) -> ConversationStream { + let (base_config, mut events) = stream.to_parts().unwrap(); + for message in before_messages { + let pos = events + .iter() + .position(|e| e.get("message").and_then(Value::as_str) == Some(message)) + .unwrap(); + events.insert(pos, future_commentary(&format!("before {message}"))); + } + ConversationStream::from_parts(base_config, events, &PartialAppConfig::empty()).unwrap() +} + +/// Two turns, each holding an unreadable commentary entry before its answer. +fn two_turns_with_future_commentary() -> ConversationStream { + let mut stream = ConversationStream::new_test(); + stream.start_turn("q0"); + stream.push(ConversationEvent::now(ChatResponse::message("a0"))); + stream.start_turn("q1"); + stream.push(ConversationEvent::now(ChatResponse::message("a1"))); + reload_with_commentary(&stream, &["a0", "a1"]) +} + +/// The stream's entries in order, one short label each. +fn entry_labels(stream: &ConversationStream) -> Vec { + stream + .events + .iter() + .map(|entry| match &entry.payload { + EventPayload::Event(event) => match &event.kind { + EventKind::TurnStart(_) => "turn".to_owned(), + EventKind::ChatRequest(request) => format!("request {}", request.content), + EventKind::ChatResponse(ChatResponse::Message { message }) => { + format!("message {message}") + } + other => other.as_str().to_owned(), + }, + EventPayload::Unknown { value, .. } => format!( + "unknown {}", + value + .get("commentary") + .and_then(Value::as_str) + .unwrap_or("?") + ), + EventPayload::ConfigDelta(_) => "config delta".to_owned(), + EventPayload::Compaction(_) => "compaction".to_owned(), + EventPayload::Overlay(_) => "overlay".to_owned(), + }) + .collect() +} + +#[test] +fn replaying_a_turn_removes_its_unreadable_chat_response() { + let mut stream = two_turns_with_future_commentary(); + + let request = stream.trim_chat_request().unwrap(); + stream.trim_trailing_empty_turn(); + stream.start_turn(request); + + assert_eq!(entry_labels(&stream), [ + "turn", + "request q0", + "unknown before a0", + "message a0", + "turn", + "request q1", + ]); +} + +#[test] +fn retain_turns_drops_an_unreadable_chat_response_with_its_turn() { + let mut stream = two_turns_with_future_commentary(); + + stream.retain_turns(|turn| turn == 0); + + assert_eq!(entry_labels(&stream), [ + "turn", + "request q0", + "unknown before a0", + "message a0", + ]); +} + +#[test] +fn retain_turns_keeps_an_unreadable_chat_response_in_a_kept_turn() { + let mut stream = two_turns_with_future_commentary(); + + stream.retain_turns(|turn| turn == 1); + + assert_eq!(entry_labels(&stream), [ + "turn", + "request q1", + "unknown before a1", + "message a1", + ]); +} + +#[test] +fn retain_last_turns_drops_an_unreadable_chat_response_from_earlier_turns() { + let mut stream = two_turns_with_future_commentary(); + + stream.retain_last_turns(1); + + assert_eq!(entry_labels(&stream), [ + "turn", + "request q1", + "unknown before a1", + "message a1", + ]); +} + +#[test] +fn popping_a_request_takes_the_unreadable_chat_response_after_it() { + // An interrupted turn: the request, and commentary a newer jp wrote in + // response, with no answer yet. Replaying the last request pops it. + let mut stream = ConversationStream::new_test(); + stream.start_turn("q0"); + stream.push(ConversationEvent::now(ChatResponse::message("a0"))); + stream.start_turn("q1"); + let (base_config, mut events) = stream.to_parts().unwrap(); + events.push(future_commentary("working on q1")); + let mut stream = + ConversationStream::from_parts(base_config, events, &PartialAppConfig::empty()).unwrap(); + + let popped = stream.pop_if(ConversationEvent::is_chat_request).unwrap(); + stream.trim_trailing_empty_turn(); + stream.start_turn(popped.event.into_chat_request().unwrap()); + + assert_eq!(entry_labels(&stream), [ + "turn", + "request q0", + "message a0", + "turn", + "request q1", + ]); +} + /// Characterization test: extending an empty stream from a source stream /// reproduces the source's observed iter sequence and serialized shape. /// From 0792f242f6d837eab595ca10cd69ec93f7251505 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 27 Oct 2026 11:00:10 +0100 Subject: [PATCH 4/4] rustqual Signed-off-by: Jean Mertz --- .config/rustqual/baseline-dry.json | 20 ++++++++++---------- .config/rustqual/baseline.json | 22 +++++++++++----------- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/.config/rustqual/baseline-dry.json b/.config/rustqual/baseline-dry.json index ed3e0d5b8..d89cb56bb 100644 --- a/.config/rustqual/baseline-dry.json +++ b/.config/rustqual/baseline-dry.json @@ -1,7 +1,7 @@ { "version": 2, - "quality_score": 0.6522264572566293, - "iosp_score": 0.8122844827586206, + "quality_score": 0.6522887459652152, + "iosp_score": 0.8123249299719888, "violations": 871, "total": 6512, "complexity_warnings": 0, @@ -48,32 +48,32 @@ { "name": "ConversationStream::pop_while", "file": "crates/jp_conversation/src/stream.rs", - "line": 818 + "line": 827 }, { "name": "ConversationStream::remove_orphaned_inquiry_responses", "file": "crates/jp_conversation/src/stream.rs", - "line": 997 + "line": 1018 }, { "name": "ConversationStream::remove_orphaned_inquiry_requests", "file": "crates/jp_conversation/src/stream.rs", - "line": 1039 + "line": 1060 }, { "name": "ConversationStream::normalize_turn_starts", "file": "crates/jp_conversation/src/stream.rs", - "line": 1086 + "line": 1107 }, { "name": "ConversationStream::sanitize_orphaned_tool_calls", "file": "crates/jp_conversation/src/stream.rs", - "line": 1173 + "line": 1194 }, { "name": "ConversationStream::from_legacy_events", "file": "crates/jp_conversation/src/stream.rs", - "line": 1582 + "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": 326 + "line": 346 }, { "name": "EventId::random", diff --git a/.config/rustqual/baseline.json b/.config/rustqual/baseline.json index 74bb44824..817211a76 100644 --- a/.config/rustqual/baseline.json +++ b/.config/rustqual/baseline.json @@ -1,9 +1,9 @@ { "version": 2, - "quality_score": 0.7058212992056335, - "iosp_score": 0.9146545777076969, + "quality_score": 0.7060539191242928, + "iosp_score": 0.9147302087435999, "violations": 866, - "total": 14002, + "total": 14012, "complexity_warnings": 163, "magic_number_warnings": 238, "nesting_depth_warnings": 27, @@ -48,32 +48,32 @@ { "name": "ConversationStream::pop_while", "file": "crates/jp_conversation/src/stream.rs", - "line": 818 + "line": 827 }, { "name": "ConversationStream::remove_orphaned_inquiry_responses", "file": "crates/jp_conversation/src/stream.rs", - "line": 997 + "line": 1018 }, { "name": "ConversationStream::remove_orphaned_inquiry_requests", "file": "crates/jp_conversation/src/stream.rs", - "line": 1039 + "line": 1060 }, { "name": "ConversationStream::normalize_turn_starts", "file": "crates/jp_conversation/src/stream.rs", - "line": 1086 + "line": 1107 }, { "name": "ConversationStream::sanitize_orphaned_tool_calls", "file": "crates/jp_conversation/src/stream.rs", - "line": 1173 + "line": 1194 }, { "name": "ConversationStream::from_legacy_events", "file": "crates/jp_conversation/src/stream.rs", - "line": 1582 + "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": 326 + "line": 346 }, { "name": "EventId::random",