diff --git a/.config/rustqual/baseline-dry.json b/.config/rustqual/baseline-dry.json index 40f69566d..d89cb56bb 100644 --- a/.config/rustqual/baseline-dry.json +++ b/.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", diff --git a/.config/rustqual/baseline.json b/.config/rustqual/baseline.json index 67757232b..817211a76 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.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", 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/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.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 e8a695bab..ee3240593 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}, }; @@ -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,7 +199,36 @@ 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, 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() + .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, + scope: EventScope::Turn, + }); } // Decode base64-encoded storage fields before deserializing. @@ -210,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, @@ -228,7 +269,7 @@ impl InternalEvent { EventPayload::ConfigDelta(_) | EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => None, + | EventPayload::Unknown { .. } => None, } } @@ -240,7 +281,7 @@ impl InternalEvent { EventPayload::ConfigDelta(_) | EventPayload::Compaction(_) | EventPayload::Overlay(_) - | EventPayload::Unknown(_) => None, + | EventPayload::Unknown { .. } => None, } } @@ -251,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 d26968b89..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. @@ -3121,6 +3124,222 @@ 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 { + scope: EventScope::Turn, + .. + })); + + 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()); +} + +/// 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. ///