feat(conversation): Introduce turn markers and stream sanitization

This commit introduces a structured turn model and robust stream
sanitization to `jp_conversation`. A new `TurnStart` event kind is
used to group sequences of events within a single query invocation,
enabling better navigation and filtering of conversation history.

The `EventBuilder` is added to handle the accumulation of partial LLM
stream chunks into complete `ConversationEvent` objects, supporting
interleaved reasoning, message, and structured output parts.

To improve privacy and prevent sensitive data from leaking into
workspace search results, a storage-level encoding layer has been
implemented. Tool arguments, results, and metadata are now base64-encoded
when persisted to disk, while remaining transparent to the rest of the
application.

Additionally, the `ConversationStream` now includes a `sanitize` process
to repair structural invariants that may be violated after filtering or
interrupted tool executions. This includes injecting synthetic error
responses for orphaned tool calls and re-indexing turn markers.

The `Thread` model has also been updated to replace the concept of
"instructions" with "sections", providing a more flexible way to
render context before sending it to providers.

Signed-off-by: Jean Mertz <git@jeanmertz.com>

Author: JeanMertz

Cargo.lock

@@ -16,17 +16,17 @@ version.workspace = true
 jp_attachment = { workspace = true }
 jp_config = { workspace = true }
 jp_id = { workspace = true }
-jp_serde = { workspace = true }
-
+base64 = { workspace = true, features = ["std"] }
+chrono = { workspace = true }
 indexmap = { workspace = true }
 quick-xml = { workspace = true, features = ["serialize"] }
 serde = { workspace = true }
 serde_json = { workspace = true, features = ["preserve_order"] }
 thiserror = { workspace = true }
-chrono = { workspace = true }
 tracing = { workspace = true }
 
 [dev-dependencies]
+assert_matches = { workspace = true }
 insta = { workspace = true, features = ["json"] }
 test-log = { workspace = true }
 

crates/jp_conversation/Cargo.toml

@@ -5,9 +5,8 @@ use std::{fmt, str::FromStr};
 use chrono::{DateTime, Utc};
 use jp_id::{
     Id, NANOSECONDS_PER_DECISECOND,
-    parts::{GlobalId, TargetId, Variant},
+    parts::{TargetId, Variant},
 };
-use jp_serde::skip_if;
 use serde::{Deserialize, Serialize};
 
 use crate::error::{Error, Result};
@@ -28,7 +27,7 @@ pub struct Conversation {
 
     /// Whether the conversation is stored in the user or workspace storage.
     // TODO: rename to `user_local`
-    #[serde(default, rename = "local", skip_serializing_if = "skip_if::is_false")]
+    #[serde(default, rename = "local", skip_serializing_if = "std::ops::Not::not")]
     pub user: bool,
 
     /// When the conversation expires.
@@ -110,7 +109,7 @@ pub struct ConversationId(#[serde(with = "jp_id::serde")] DateTime<Utc>);
 impl fmt::Debug for ConversationId {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_tuple("ConversationId")
-            .field(&self.to_string())
+            .field(&self.as_deciseconds())
             .finish()
     }
 }
@@ -236,10 +235,6 @@ impl Id for ConversationId {
     fn target_id(&self) -> TargetId {
         self.as_deciseconds().to_string().into()
     }
-
-    fn global_id(&self) -> GlobalId {
-        jp_id::global::get().into()
-    }
 }
 
 impl fmt::Display for ConversationId {

crates/jp_conversation/src/conversation.rs

@@ -3,6 +3,7 @@
 mod chat;
 mod inquiry;
 mod tool_call;
+mod turn;
 
 use std::fmt;
 
@@ -13,8 +14,12 @@ use serde_json::{Map, Value};
 
 pub use self::{
     chat::{ChatRequest, ChatResponse},
-    inquiry::{InquiryAnswerType, InquiryQuestion, InquiryRequest, InquiryResponse, InquirySource},
+    inquiry::{
+        InquiryAnswerType, InquiryId, InquiryQuestion, InquiryRequest, InquiryResponse,
+        InquirySource, SelectOption,
+    },
     tool_call::{ToolCallRequest, ToolCallResponse},
+    turn::TurnStart,
 };
 
 /// A single event in a conversation.
@@ -32,11 +37,7 @@ pub struct ConversationEvent {
     pub kind: EventKind,
 
     /// Additional opaque metadata associated with the event.
-    #[serde(
-        default,
-        skip_serializing_if = "Map::is_empty",
-        with = "jp_serde::repr::base64_json_map"
-    )]
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
     pub metadata: Map<String, Value>,
 }
 
@@ -305,12 +306,43 @@ impl ConversationEvent {
             _ => None,
         }
     }
+
+    /// Returns `true` if the event is a [`TurnStart`].
+    #[must_use]
+    pub const fn is_turn_start(&self) -> bool {
+        matches!(self.kind, EventKind::TurnStart(_))
+    }
+
+    /// Returns a reference to the [`TurnStart`], if applicable.
+    #[must_use]
+    pub const fn as_turn_start(&self) -> Option<&TurnStart> {
+        match &self.kind {
+            EventKind::TurnStart(turn_start) => Some(turn_start),
+            _ => None,
+        }
+    }
+
+    /// Consumes the event and returns the [`TurnStart`], if applicable.
+    #[must_use]
+    pub fn into_turn_start(self) -> Option<TurnStart> {
+        match self.kind {
+            EventKind::TurnStart(turn_start) => Some(turn_start),
+            _ => None,
+        }
+    }
 }
 
 /// A type of event in a conversation.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum EventKind {
+    /// A turn start event.
+    ///
+    /// This event marks the beginning of a new turn in the conversation. A turn
+    /// groups together a user's chat request through the assistant's final
+    /// response, including any intermediate tool calls.
+    TurnStart(TurnStart),
+
     /// A chat request event.
     ///
     /// This event is usually triggered by the user, but can also be
@@ -353,6 +385,22 @@ pub enum EventKind {
     InquiryResponse(InquiryResponse),
 }
 
+impl EventKind {
+    /// Returns the name of the event kind.
+    #[must_use]
+    pub const fn as_str(&self) -> &str {
+        match self {
+            Self::TurnStart(_) => "TurnStart",
+            Self::ChatRequest(_) => "ChatRequest",
+            Self::ChatResponse(_) => "ChatResponse",
+            Self::ToolCallRequest(_) => "ToolCallRequest",
+            Self::ToolCallResponse(_) => "ToolCallResponse",
+            Self::InquiryRequest(_) => "InquiryRequest",
+            Self::InquiryResponse(_) => "InquiryResponse",
+        }
+    }
+}
+
 impl From<ChatRequest> for EventKind {
     fn from(request: ChatRequest) -> Self {
         Self::ChatRequest(request)
@@ -389,6 +437,12 @@ impl From<InquiryResponse> for EventKind {
     }
 }
 
+impl From<TurnStart> for EventKind {
+    fn from(turn_start: TurnStart) -> Self {
+        Self::TurnStart(turn_start)
+    }
+}
+
 impl From<ChatRequest> for ConversationEvent {
     fn from(request: ChatRequest) -> Self {
         Self::now(request)
@@ -424,3 +478,9 @@ impl From<InquiryResponse> for ConversationEvent {
         Self::now(response)
     }
 }
+
+impl From<TurnStart> for ConversationEvent {
+    fn from(turn_start: TurnStart) -> Self {
+        Self::now(turn_start)
+    }
+}