car_sync/

conversation.rs

//! Transcript resume: the conversation surface as an ordered, role-threaded
//! projection of the oplog (slice **B2** of
//! `docs/proposals/multi-device-sync.md`), hardened by kernel review.
//!
//! # What this closes
//!
//! `docs/solutions/conversation-persistence-removed-in-0.25.md` records that
//! the disk-backed `ConversationStore` island was **removed in 0.25** — dead
//! code with a latent *compaction-vs-store incoherence* bug (compaction
//! summarized the in-memory graph but never wrote the summaries back to the
//! JSONL store, so a resume would have reloaded stale raw turns). That doc's
//! forward path: "an append-only oplog as the source of truth with files/caches
//! as projections." B2 is that path, oplog-native — a conversation is a
//! projection of ordered turn ops, not a resurrected second store.
//!
//! # The model: a transcript is a projection, not a store
//!
//! A conversation turn is an [`crate::oplog::OpRecord`] on the
//! [`crate::oplog::Surface::Conversation`] surface. The transcript is a *pure
//! fold* of those ops: filter by `conversation_id`, order by the canonical
//! `(hlc, op_id)` total order the crate already agrees on
//! ([`crate::fold::SyncState::log_entries`]), project each payload into a typed
//! [`Turn`]. One source of truth; no second store to drift.
//!
//! # Turn identity: an EVENT STREAM (op_id-keyed), not a content entity
//!
//! **Kernel-review correction (reversed from the first B2 cut).** A conversation
//! turn folds as an **event-stream multiset keyed by `op_id`**
//! ([`crate::oplog::Surface::is_event_stream`] returns `true` for
//! `Conversation`), NOT by content. The content-keyed first cut had a
//! reproduced **silent data-loss** bug: `stable_key` hashes the caller's
//! *payload* — `{conversation_id, role, content, timestamp}` — so two genuine
//! user "yes" turns stamped at the same payload timestamp (second-granularity
//! stamps, a cached `now()`, a rapid double-confirm) collapsed to ONE entry.
//! The earlier justification ("a repeated utterance differs in timestamp / the
//! HLC advances") was wrong: the fold keyed on the payload timestamp, not the
//! HLC.
//!
//! The right identity: **a turn has exactly one author and propagates by op
//! replication, so op identity IS turn identity.** Keyed by `op_id`, a *resent*
//! op dedups (retransmission), while two *distinct* authorings never collapse —
//! even byte-identical ones. This reuses B1's multiset machinery (the same the
//! routing observations use). Conversation differs from routing only in that
//! its entries are **independent** (no path-dependent EMA replay), so it
//! tolerates `LastN` retention where routing forbids any trim — see
//! [`crate::oplog::Surface::is_replay_stream`].
//!
//! # Ordering across devices
//!
//! HLC gives causal order; two devices talking to the same agent concurrently
//! interleave deterministically by `Hlc`'s derived `Ord`
//! (`(wall_ms, counter, device_id)`), tie-broken on `op_id`. A turn that
//! causally follows another (its writer `observe`d it) always sorts after it;
//! genuinely concurrent turns fall back to the stable `device_id` tiebreak. So
//! every device reconstructs a byte-identical transcript from any delivery
//! order.
//!
//! **But determinism ≠ provider-validity** (the second kernel-review defect).
//! Causal order says nothing about *concurrent* turns: two devices each replying
//! to the same user turn yield `[user, assistant, assistant]` — which Anthropic
//! 400s. So [`crate::fold::SyncState::resume_messages`] does not emit the raw
//! transcript; it runs a **repair** that guarantees a provider-valid `Message`
//! sequence (the "runtime validates" thesis applied to the projection): adjacent
//! same-role turns are coalesced, an orphan `tool_result` (one not answering a
//! preceding assistant `tool_call` — e.g. a `LastN` window that cut inside a
//! tool exchange) is dropped, and a dangling assistant `tool_call` with no
//! following `tool_result` has its calls stripped. See [`repair`].
//!
//! # Compaction coherence (why the 0.25 bug cannot return)
//!
//! Conversation retention is B4's `RetentionRule::LastN` over the folded
//! snapshot; the checkpoint keeps the last N turns by `timestamp` and older raw
//! turns drop from the read model. The 0.25 incoherence was structural — *two*
//! stores on two write paths. B2 has **one** source of truth (the oplog) and the
//! transcript is a projection of the *same* folded state B4's checkpoint
//! serializes. `apply_retention` over the compacted device's state equals
//! `apply_retention` over a fresh full fold (byte- and hash-identical). The
//! **semantic** summarization of aged-out turns lives in memgine
//! (`ConversationSummary` nodes, a B6 concern); B2 supplies the ordered raw
//! turns it summarizes and the last-N window, nothing lossy.
//!
//! # The resume bridge
//!
//! [`Turn::to_message`] builds each turn as the **real**
//! [`car_inference_types::Message`] (`user` / `assistant {content, tool_calls}`
//! / `tool_result {tool_use_id, content}`), and `resume_messages` returns the
//! repaired `Vec<Message>` car-inference's multi-turn path replays. Because
//! car-sync depends on the shared `car-inference-types` crate — not a hand-copied
//! mirror — a change to `Message`'s shape is a **compile error** here, not a
//! runtime `from_value::<Message>` break in the daemon. The daemon/memgine
//! adoption (feeding these into the engine's multi-turn path) is B6.

use crate::fold::{FoldedRecord, SyncState};
use crate::oplog::Hlc;
use car_inference_types::{Message, Provenance, ToolCall};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

/// The conversation a turn belongs to when its payload carries no explicit
/// `conversation_id` (e.g. the legacy `{speaker, text, timestamp}` turns the
/// B4 tests emit). Such turns fold into one unnamed default transcript.
pub const DEFAULT_CONVERSATION: &str = "";

/// A conversation turn's role. Serializes snake_case (`user`/`assistant`/
/// `tool`); [`Turn::to_message`] maps `Tool` onto the `Message::ToolResult`
/// role.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Role {
    /// A user turn (`Message::User`).
    User,
    /// An assistant turn, possibly with `tool_calls` (`Message::Assistant`).
    Assistant,
    /// A tool-result turn (`Message::ToolResult`).
    Tool,
}

impl Role {
    /// Parse the `role`/`speaker` string a Conversation payload carries.
    /// Tolerant of the legacy `speaker` values and of `tool_result` (the
    /// Message role name) as well as `tool`.
    fn parse(s: &str) -> Role {
        match s {
            "assistant" => Role::Assistant,
            "tool" | "tool_result" => Role::Tool,
            _ => Role::User,
        }
    }
}

/// One folded, ordered transcript turn — the typed projection of a
/// [`crate::oplog::Surface::Conversation`] op.
///
/// `hlc`/`op_id` are provenance/ordering only (not part of the replayed
/// `Message`): they record where the turn sits in the causal order and which
/// op produced it (op identity IS turn identity — see the module docs).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Turn {
    /// Which conversation this turn threads into ([`DEFAULT_CONVERSATION`] when
    /// the payload names none).
    pub conversation_id: String,
    /// user / assistant / tool.
    pub role: Role,
    /// The turn text (`content`, falling back to the legacy `text` field).
    pub content: String,
    /// Assistant tool calls — the REAL [`car_inference_types::ToolCall`] type,
    /// parsed from the payload. Empty for non-assistant turns or an assistant
    /// turn that called no tool.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tool_calls: Vec<ToolCall>,
    /// For a `Tool` turn: the id of the tool call this result answers
    /// (`Message::ToolResult.tool_use_id`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tool_use_id: Option<String>,
    /// For a `Tool` turn: whether the result came from outside the trust
    /// boundary (car#723).
    ///
    /// Carried through the oplog on purpose. A marking that survives the live
    /// turn but not a resume is worse than none — the same fetched bytes would
    /// come back as trusted after a restart, and the one path where nobody is
    /// watching is exactly where that matters. `#[serde(default)]` reads turns
    /// recorded before this field as `Internal`, which is how they were treated.
    #[serde(default, skip_serializing_if = "Provenance::is_internal")]
    pub provenance: Provenance,
    /// Turn timestamp (ms) — the recency key B4's `LastN` conversation
    /// retention orders on.
    pub timestamp: u64,
    /// The causal-order stamp this turn folded with (ordering/provenance).
    pub hlc: Hlc,
    /// The op that produced this turn (provenance; the fold identity).
    pub op_id: String,
}

impl Turn {
    /// Build the Conversation op **payload** for a user turn — append it via
    /// `DeviceLog::append(scope, Surface::Conversation, payload)`.
    pub fn user_payload(conversation_id: &str, content: &str, timestamp: u64) -> Value {
        json!({
            "conversation_id": conversation_id,
            "role": "user",
            "content": content,
            "timestamp": timestamp,
        })
    }

    /// Build the Conversation op payload for an assistant turn. `tool_calls`
    /// are raw `ToolCall` JSON values (`{id?, name, arguments}` — how a model
    /// or the daemon already holds them); [`Turn::from_record`] parses them
    /// into the typed [`ToolCall`]. Empty for a plain text reply.
    pub fn assistant_payload(
        conversation_id: &str,
        content: &str,
        tool_calls: Vec<Value>,
        timestamp: u64,
    ) -> Value {
        json!({
            "conversation_id": conversation_id,
            "role": "assistant",
            "content": content,
            "tool_calls": tool_calls,
            "timestamp": timestamp,
        })
    }

    /// Build the Conversation op payload for a tool-result turn.
    pub fn tool_payload(
        conversation_id: &str,
        tool_use_id: &str,
        content: &str,
        timestamp: u64,
    ) -> Value {
        Self::tool_payload_with_provenance(
            conversation_id,
            tool_use_id,
            content,
            timestamp,
            Provenance::Internal,
        )
    }

    /// [`Turn::tool_payload`] for a result whose bytes came from outside the
    /// trust boundary. Separate constructor rather than a defaulted argument so
    /// the caller has to say which it is; `tool_payload` keeps the common case
    /// short and every existing call site correct.
    pub fn tool_payload_with_provenance(
        conversation_id: &str,
        tool_use_id: &str,
        content: &str,
        timestamp: u64,
        provenance: Provenance,
    ) -> Value {
        let mut v = json!({
            "conversation_id": conversation_id,
            "role": "tool",
            "tool_use_id": tool_use_id,
            "content": content,
            "timestamp": timestamp,
        });
        if provenance.is_external() {
            v["provenance"] = json!("external");
        }
        v
    }

    /// Project a folded conversation record into a typed [`Turn`]. Returns
    /// `None` for a tombstone stub (a retention-dropped entity's referential
    /// placeholder — never a real turn).
    ///
    /// Tolerant of both the rich B2 payload (`role`/`content`/`tool_calls`) and
    /// the legacy `(speaker, text, timestamp)` shape the B4 tests use, so one
    /// transcript view serves every conversation op ever written.
    pub fn from_record(record: &FoldedRecord) -> Option<Turn> {
        if crate::compact::is_tombstone(record) {
            return None;
        }
        let p = &record.payload;
        let role = p
            .get("role")
            .or_else(|| p.get("speaker"))
            .and_then(Value::as_str)
            .map(Role::parse)
            .unwrap_or(Role::User);
        let content = p
            .get("content")
            .or_else(|| p.get("text"))
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string();
        let conversation_id = p
            .get("conversation_id")
            .and_then(Value::as_str)
            .unwrap_or(DEFAULT_CONVERSATION)
            .to_string();
        // Parse the tool_calls array into the REAL ToolCall type (best-effort:
        // a malformed entry is dropped rather than poisoning the whole turn).
        let tool_calls = p
            .get("tool_calls")
            .and_then(Value::as_array)
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| serde_json::from_value::<ToolCall>(v.clone()).ok())
                    .collect()
            })
            .unwrap_or_default();
        let tool_use_id = p
            .get("tool_use_id")
            .and_then(Value::as_str)
            .map(str::to_string);
        // Unknown/absent → Internal. An unrecognised value is treated as
        // Internal rather than rejected: a fold that errors on an unfamiliar
        // string would make a newer peer's op poison an older peer's replay,
        // and the CRDT fold has to stay total.
        let provenance = match p.get("provenance").and_then(|v| v.as_str()) {
            Some("external") => Provenance::External,
            _ => Provenance::Internal,
        };
        let timestamp = p
            .get("timestamp")
            .and_then(|v| v.as_u64().or_else(|| v.as_f64().map(|f| f as u64)))
            .unwrap_or(0);
        Some(Turn {
            conversation_id,
            role,
            content,
            tool_calls,
            tool_use_id,
            provenance,
            timestamp,
            hlc: record.hlc.clone(),
            op_id: record.op_id.clone(),
        })
    }

    /// Build the REAL [`car_inference_types::Message`] for this turn (typed —
    /// a shape change to `Message` is a compile error here). The mapping:
    ///
    /// - [`Role::User`] → `Message::User`
    /// - [`Role::Assistant`] → `Message::Assistant { content, tool_calls }`
    /// - [`Role::Tool`] → `Message::ToolResult { tool_use_id, content, provenance }`
    pub fn to_message(&self) -> Message {
        match self.role {
            Role::User => Message::User {
                content: self.content.clone(),
            },
            Role::Assistant => Message::Assistant {
                content: self.content.clone(),
                tool_calls: self.tool_calls.clone(),
                // Cross-session thinking-block resume is a follow-up: the oplog
                // Turn does not (yet) record thinking, so resumed turns replay
                // without it. Empty keeps the wire valid (Anthropic drops absent
                // thinking silently on a fresh session's first turn).
                thinking: Vec::new(),
            },
            Role::Tool => Message::ToolResult {
                tool_use_id: self.tool_use_id.clone().unwrap_or_default(),
                content: self.content.clone(),
                provenance: self.provenance,
            },
        }
    }
}

/// Join two turn bodies for a coalesced turn (skip an empty side).
fn join_content(a: &str, b: &str) -> String {
    match (a.is_empty(), b.is_empty()) {
        (true, _) => b.to_string(),
        (_, true) => a.to_string(),
        _ => format!("{a}\n\n{b}"),
    }
}

/// Repair an ordered transcript into a **provider-valid** `Message` sequence.
///
/// The raw transcript is causally ordered but not provider-valid: concurrent
/// turns can produce invalid adjacencies (`[user, assistant, assistant]` →
/// Anthropic 400), and a `LastN` window can cut inside a tool exchange leaving
/// an orphan `tool_result` at the head (also a 400). This is the runtime
/// validating its own projection. Guarantees on the output:
///
/// 1. **No adjacent same-role messages** — consecutive `User`s (or `Assistant`s)
///    are coalesced into one, their bodies joined and (for assistants) their
///    tool_calls concatenated. Two concurrent replies to one user turn merge;
///    an assistant "threaded after" an unrelated user turn it never observed is
///    merged rather than mis-attributed (best-effort — the deterministic order
///    is preserved, the invalid adjacency is not emitted).
/// 2. **No orphan `tool_result`** — a `Tool` turn is kept only if it answers a
///    preceding assistant that carried `tool_calls` (or a run of such results);
///    otherwise it is dropped (the `LastN`-cut-mid-exchange case).
/// 3. **No dangling assistant `tool_call`** — an assistant whose `tool_calls`
///    are not answered by a following `tool_result` has its calls stripped
///    (kept as a plain text turn), so the sequence never presents an
///    unanswered tool_use.
///
/// Leading-role normalization (ensuring the sequence *starts* with a user turn,
/// a per-provider requirement) is the caller/protocol-handler's concern — it
/// already folds the system prompt and prepends the next user message; `repair`
/// only guarantees internal adjacency + tool-pairing validity.
pub fn repair(turns: Vec<Turn>) -> Vec<Message> {
    let mut out: Vec<Turn> = Vec::new();
    // Does the last emitted turn leave a tool exchange open (an assistant with
    // tool_calls, or a tool_result continuing one)? Only then is a tool_result
    // valid.
    let mut tool_open = false;
    for t in turns {
        match t.role {
            Role::Tool => {
                if tool_open {
                    out.push(t); // a valid result; the exchange stays open
                }
                // else: orphan tool_result → dropped
            }
            Role::User => {
                if let Some(last) = out.last_mut() {
                    if last.role == Role::User {
                        last.content = join_content(&last.content, &t.content);
                        continue;
                    }
                }
                tool_open = false;
                out.push(t);
            }
            Role::Assistant => {
                if let Some(last) = out.last_mut() {
                    if last.role == Role::Assistant {
                        last.content = join_content(&last.content, &t.content);
                        last.tool_calls.extend(t.tool_calls);
                        tool_open = !last.tool_calls.is_empty();
                        continue;
                    }
                }
                tool_open = !t.tool_calls.is_empty();
                out.push(t);
            }
        }
    }
    // Post-pass: strip a dangling assistant tool_call (no following tool_result),
    // so no unanswered tool_use is ever presented to a provider.
    for i in 0..out.len() {
        if out[i].role == Role::Assistant && !out[i].tool_calls.is_empty() {
            let answered = out.get(i + 1).is_some_and(|n| n.role == Role::Tool);
            if !answered {
                out[i].tool_calls.clear();
            }
        }
    }
    out.iter().map(Turn::to_message).collect()
}

impl SyncState {
    /// The ordered, role-threaded transcript for one conversation — the
    /// causally `(hlc, op_id)`-ordered `Vec<Turn>` (B2). This is the raw
    /// projection (every folded turn, in order); [`SyncState::resume_messages`]
    /// is the provider-valid `Message` view. Turns whose payload names no
    /// `conversation_id` belong to [`DEFAULT_CONVERSATION`]; tombstone stubs are
    /// skipped. Order-independent of delivery, byte-identical on every device
    /// that folded the same op-set.
    pub fn transcript(&self, conversation_id: &str) -> Vec<Turn> {
        self.log_entries(&crate::oplog::Surface::Conversation.tag())
            .into_iter()
            .filter_map(Turn::from_record)
            .filter(|t| t.conversation_id == conversation_id)
            .collect()
    }

    /// Every conversation id present in the folded state, in stable sorted
    /// order (includes [`DEFAULT_CONVERSATION`] when unnamed turns exist).
    pub fn conversation_ids(&self) -> Vec<String> {
        let mut ids: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
        for rec in self.log_entries(&crate::oplog::Surface::Conversation.tag()) {
            if let Some(turn) = Turn::from_record(rec) {
                ids.insert(turn.conversation_id);
            }
        }
        ids.into_iter().collect()
    }

    /// Reconstruct the runtime's multi-turn conversation state: the ordered,
    /// **provider-valid** [`car_inference_types::Message`] sequence
    /// car-inference's multi-turn path replays to continue the conversation
    /// (B2's resume bridge). The raw transcript is [`repair`]ed first, so the
    /// result never contains an invalid role adjacency or an orphan/dangling
    /// tool exchange. The daemon/memgine adoption is B6.
    pub fn resume_messages(&self, conversation_id: &str) -> Vec<Message> {
        repair(self.transcript(conversation_id))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::compact::{
        apply_retention, plan_compaction, AckTable, RetentionPolicy, RetentionRule,
    };
    use crate::fold::{fold, fold_onto, state_hash};
    use crate::oplog::{DeviceLog, Scope, Surface};

    /// A tool result fetched from the open internet must still be marked as
    /// such after a restart. If provenance survived the live turn but not the
    /// oplog, the same bytes would come back trusted on resume — and resume is
    /// exactly the path where nobody is watching.
    #[test]
    fn external_provenance_survives_the_oplog_round_trip() {
        let mut d = DeviceLog::new("dev-a");
        let ops = vec![
            d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::user_payload("c1", "what does that page say?", 10),
            ),
            d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::assistant_payload(
                    "c1",
                    "",
                    vec![serde_json::json!({"id": "call_0", "name": "web_search",
                                            "arguments": {"q": "x"}})],
                    11,
                ),
            ),
            d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::tool_payload_with_provenance(
                    "c1",
                    "call_0",
                    "fetched page text",
                    12,
                    Provenance::External,
                ),
            ),
        ];
        let messages: Vec<Message> = fold(&ops).resume_messages("c1");
        let tool = messages
            .iter()
            .find(|m| matches!(m, Message::ToolResult { .. }))
            .expect("the tool result must survive resume");
        match tool {
            Message::ToolResult {
                content,
                provenance,
                ..
            } => {
                assert_eq!(content, "fetched page text");
                assert_eq!(
                    *provenance,
                    Provenance::External,
                    "resume downgraded external content to trusted"
                );
            }
            other => panic!("expected ToolResult, got {other:?}"),
        }
    }

    #[test]
    fn plain_tool_payload_is_internal_and_omits_the_field() {
        let payload = Turn::tool_payload("c1", "call_0", "exit 0", 10);
        assert!(
            payload.get("provenance").is_none(),
            "internal must add no bytes: {payload}"
        );
        let external =
            Turn::tool_payload_with_provenance("c1", "call_0", "x", 10, Provenance::External);
        assert_eq!(external["provenance"], "external");
    }

    #[test]
    fn unknown_provenance_value_folds_as_internal_rather_than_failing() {
        // The CRDT fold has to stay total: a newer peer writing a value this
        // build does not know must not poison an older peer's replay.
        let mut d = DeviceLog::new("dev-a");
        let mut payload = Turn::tool_payload("c1", "call_0", "x", 10);
        payload["provenance"] = serde_json::json!("from-the-future");
        let ops = vec![d.append(Scope::Personal, Surface::Conversation, payload)];
        let turns = fold(&ops).transcript("c1");
        assert_eq!(turns.len(), 1);
        assert_eq!(turns[0].provenance, Provenance::Internal);
    }

    /// Assert a repaired sequence is provider-valid: no adjacent same-role
    /// (assistant/assistant, user/user), no orphan/leading tool_result, no
    /// dangling assistant tool_call.
    fn assert_provider_valid(ms: &[Message]) {
        for (i, w) in ms.windows(2).enumerate() {
            let dup = matches!(
                (&w[0], &w[1]),
                (Message::Assistant { .. }, Message::Assistant { .. })
                    | (Message::User { .. }, Message::User { .. })
            );
            assert!(!dup, "invalid adjacency at {i}: {:?} then {:?}", w[0], w[1]);
        }
        for (i, m) in ms.iter().enumerate() {
            if matches!(m, Message::ToolResult { .. }) {
                let opens_here = i > 0
                    && (matches!(&ms[i - 1], Message::Assistant { tool_calls, .. } if !tool_calls.is_empty())
                        || matches!(&ms[i - 1], Message::ToolResult { .. }));
                assert!(opens_here, "orphan tool_result at index {i}");
            }
            if let Message::Assistant { tool_calls, .. } = m {
                if !tool_calls.is_empty() {
                    let answered = ms
                        .get(i + 1)
                        .is_some_and(|n| matches!(n, Message::ToolResult { .. }));
                    assert!(answered, "dangling assistant tool_call at index {i}");
                }
            }
        }
    }

    /// Append the three turn kinds to one conversation on a single device.
    fn one_device_conversation(conv: &str) -> Vec<crate::oplog::OpRecord> {
        let mut d = DeviceLog::new("dev-a");
        vec![
            d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::user_payload(conv, "what's the weather?", 10),
            ),
            d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::assistant_payload(
                    conv,
                    "",
                    vec![
                        json!({"id": "call_0", "name": "get_weather", "arguments": {"city": "SF"}}),
                    ],
                    11,
                ),
            ),
            d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::tool_payload(conv, "call_0", "sunny, 72F", 12),
            ),
            d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::assistant_payload(conv, "It's sunny and 72F in SF.", vec![], 13),
            ),
        ]
    }

    #[test]
    fn transcript_folds_in_causal_order_regardless_of_delivery() {
        let ops = one_device_conversation("c1");
        let expected = [
            "what's the weather?",
            "",
            "sunny, 72F",
            "It's sunny and 72F in SF.",
        ];
        for delivery in [ops.clone(), ops.iter().rev().cloned().collect::<Vec<_>>()] {
            let turns = fold(&delivery).transcript("c1");
            let texts: Vec<&str> = turns.iter().map(|t| t.content.as_str()).collect();
            assert_eq!(texts, expected);
            assert_eq!(
                turns.iter().map(|t| t.role).collect::<Vec<_>>(),
                vec![Role::User, Role::Assistant, Role::Tool, Role::Assistant]
            );
        }
    }

    #[test]
    fn resume_produces_a_valid_message_sequence_of_the_real_type() {
        // The resume bridge builds the REAL Message type (compile-checked), and
        // the sequence is provider-valid.
        let ops = one_device_conversation("c1");
        let messages: Vec<Message> = fold(&ops).resume_messages("c1");
        assert_eq!(messages.len(), 4);
        assert_provider_valid(&messages);

        match &messages[0] {
            Message::User { content } => assert_eq!(content, "what's the weather?"),
            other => panic!("turn 0 should be a user message, got {other:?}"),
        }
        match &messages[1] {
            Message::Assistant { tool_calls, .. } => {
                assert_eq!(
                    tool_calls.len(),
                    1,
                    "assistant tool_calls round-trip as the real ToolCall"
                );
                assert_eq!(tool_calls[0].name, "get_weather");
            }
            other => panic!("turn 1 should be an assistant tool call, got {other:?}"),
        }
        match &messages[2] {
            Message::ToolResult {
                tool_use_id,
                content,
                provenance,
            } => {
                assert_eq!(tool_use_id, "call_0");
                assert_eq!(content, "sunny, 72F");
                assert_eq!(*provenance, Provenance::Internal);
            }
            other => panic!("turn 2 should be a tool_result, got {other:?}"),
        }
        match &messages[3] {
            Message::Assistant {
                content,
                tool_calls,
                ..
            } => {
                assert_eq!(content, "It's sunny and 72F in SF.");
                assert!(tool_calls.is_empty());
            }
            other => panic!("turn 3 should be a plain assistant reply, got {other:?}"),
        }
    }

    // ------------------------------------------------------------------
    // CRIT-2 (reproduced): content-keyed identity DROPPED distinct turns. Now a
    // conversation turn is an event stream — op identity is turn identity — so
    // two distinct authorings never collapse; only a resent op dedups.
    // ------------------------------------------------------------------
    #[test]
    fn crit2_two_genuine_same_payload_turns_do_not_collapse() {
        // Two genuine user "yes" turns stamped at the SAME payload timestamp
        // (second-granularity clock / rapid double-confirm). Under the old
        // content-key fold these were ONE entry (silent data loss). As an event
        // stream they are two distinct ops → two transcript entries.
        let mut d = DeviceLog::new("dev-a");
        let yes = || Turn::user_payload("c1", "yes", 5); // identical payload incl. timestamp
        let o1 = d.append(Scope::Personal, Surface::Conversation, yes());
        let o2 = d.append(Scope::Personal, Surface::Conversation, yes());
        assert_ne!(
            o1.op_id, o2.op_id,
            "distinct ops (different seq/hlc) → distinct op_id"
        );
        assert_eq!(
            fold(&[o1.clone(), o2]).transcript("c1").len(),
            2,
            "both genuine turns survive (CRIT-2 fixed)"
        );
        // A RESENT op (same op_id, retransmission) still dedups to one.
        assert_eq!(
            fold(&[o1.clone(), o1]).transcript("c1").len(),
            1,
            "resent op dedups on op_id"
        );
    }

    #[test]
    fn same_payload_turns_on_two_devices_are_distinct_events() {
        // Two devices independently authoring a byte-identical turn are TWO
        // events (distinct op_id via distinct device_id), not one.
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let payload = Turn::user_payload("c1", "hello", 5);
        let oa = a.append(Scope::Personal, Surface::Conversation, payload.clone());
        let ob = b.append(Scope::Personal, Surface::Conversation, payload);
        assert_eq!(fold(&[oa, ob]).transcript("c1").len(), 2);
    }

    // ------------------------------------------------------------------
    // CRIT-1 (reproduced): concurrent-device replies produced an invalid
    // [user, assistant, assistant] adjacency. resume_messages now repairs it.
    // ------------------------------------------------------------------
    #[test]
    fn crit1_concurrent_assistant_replies_repair_to_a_valid_sequence() {
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let u = a.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::user_payload("c1", "hi", 1),
        );
        b.observe(&u.hlc);
        // Two concurrent replies to u (neither observed the other).
        let ra = a.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::assistant_payload("c1", "reply A", vec![], 2),
        );
        let rb = b.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::assistant_payload("c1", "reply B", vec![], 2),
        );

        let ops = vec![u, ra, rb];
        // Raw transcript has the invalid [user, assistant, assistant]…
        let raw = fold(&ops).transcript("c1");
        assert_eq!(raw.len(), 3);
        assert_eq!(
            (raw[1].role, raw[2].role),
            (Role::Assistant, Role::Assistant)
        );
        // …but resume repairs it, order-independently over every delivery.
        let baseline = fold(&ops).resume_messages("c1");
        assert_eq!(baseline.len(), 2, "the two concurrent replies coalesce");
        assert_provider_valid(&baseline);
        assert!(matches!(&baseline[0], Message::User { .. }));
        assert!(
            matches!(&baseline[1], Message::Assistant { content, .. } if content.contains("reply A") && content.contains("reply B"))
        );
        for perm in permutations(&ops) {
            assert_eq!(
                fold(&perm).resume_messages("c1"),
                baseline,
                "repair is delivery-order-independent"
            );
        }
    }

    #[test]
    fn adjacent_user_turns_coalesce_before_a_reply() {
        // Two user turns land back-to-back (userA, then userB observing it),
        // then a reply that observed both. Raw = [user, user, assistant] — the
        // two users are an invalid adjacency. Repair coalesces them so the
        // assistant follows a single merged user context, never a mis-thread.
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let ua = a.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::user_payload("c1", "userA", 1),
        );
        b.observe(&ua.hlc);
        let ub = b.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::user_payload("c1", "userB", 2),
        );
        a.observe(&ub.hlc);
        let ra = a.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::assistant_payload("c1", "reply", vec![], 3),
        );

        let ops = vec![ua, ub, ra];
        assert_eq!(
            fold(&ops)
                .transcript("c1")
                .iter()
                .map(|t| t.role)
                .collect::<Vec<_>>(),
            vec![Role::User, Role::User, Role::Assistant],
            "raw has the invalid user/user adjacency"
        );
        let ms = fold(&ops).resume_messages("c1");
        assert_eq!(ms.len(), 2, "the two users coalesce");
        assert_provider_valid(&ms);
        assert!(
            matches!(&ms[0], Message::User { content } if content.contains("userA") && content.contains("userB"))
        );
        assert!(matches!(&ms[1], Message::Assistant { .. }));
    }

    // ------------------------------------------------------------------
    // CRIT-3 (reproduced): a LastN window cut inside a tool exchange orphaned a
    // leading tool_result. resume_messages drops it.
    // ------------------------------------------------------------------
    #[test]
    fn crit3_lastn_orphan_tool_result_is_dropped_on_resume() {
        // [user, assistant(tool_call), tool_result, assistant] under LastN{2}
        // retains [tool_result, assistant] — a LEADING orphan tool_result
        // (provider 400). (This ALSO exercises that LastN is ALLOWED on the
        // event-stream conversation surface — the retention-guard reconciliation.)
        let ops = one_device_conversation("c1"); // ts 10..13
        let mut policy = RetentionPolicy::keep_all();
        policy
            .rules
            .insert("conversation".to_string(), RetentionRule::LastN { n: 2 });
        let (retained, _) = apply_retention(&fold(&ops), &policy, 1_000).unwrap();

        let raw = retained.transcript("c1");
        assert_eq!(
            raw.iter().map(|t| t.role).collect::<Vec<_>>(),
            vec![Role::Tool, Role::Assistant],
            "the retained window is the orphan [tool_result, assistant]"
        );

        let ms = retained.resume_messages("c1");
        assert!(
            !matches!(ms.first(), Some(Message::ToolResult { .. })),
            "leading orphan tool_result dropped"
        );
        assert_provider_valid(&ms);
        assert_eq!(ms.len(), 1);
        assert!(matches!(&ms[0], Message::Assistant { .. }));
    }

    #[test]
    fn repair_drops_a_leading_orphan_tool_result_directly() {
        // Unit-level: repair guarantees no leading/orphan tool_result even from
        // a hand-built transcript that starts mid-exchange.
        let hlc = Hlc {
            wall_ms: 0,
            counter: 0,
            device_id: "d".into(),
        };
        let orphan = Turn {
            conversation_id: "c".into(),
            role: Role::Tool,
            content: "res".into(),
            tool_calls: vec![],
            tool_use_id: Some("call_0".into()),
            provenance: Provenance::Internal,
            timestamp: 1,
            hlc: hlc.clone(),
            op_id: "op-x".into(),
        };
        let asst = Turn {
            conversation_id: "c".into(),
            role: Role::Assistant,
            content: "done".into(),
            tool_calls: vec![],
            tool_use_id: None,
            provenance: Provenance::Internal,
            timestamp: 2,
            hlc,
            op_id: "op-y".into(),
        };
        let ms = repair(vec![orphan, asst]);
        assert!(!matches!(ms.first(), Some(Message::ToolResult { .. })));
        assert_provider_valid(&ms);
    }

    #[test]
    fn concurrent_device_turns_interleave_deterministically_and_order_independently() {
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let a1 = a.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::user_payload("c1", "from A", 1),
        );
        b.observe(&a1.hlc);
        let b1 = b.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::assistant_payload("c1", "B replies to A", vec![], 2),
        );
        let a2 = a.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::user_payload("c1", "A concurrent", 3),
        );
        let b2 = b.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::user_payload("c1", "B concurrent", 3),
        );

        let ops = vec![a1, b1, a2, b2];
        let baseline = fold(&ops).transcript("c1");
        assert_eq!(baseline.len(), 4);
        let texts: Vec<&str> = baseline.iter().map(|t| t.content.as_str()).collect();
        let pos = |s: &str| texts.iter().position(|t| *t == s).unwrap();
        assert_eq!(pos("from A"), 0);
        assert!(pos("from A") < pos("B replies to A"), "causality survives");

        let baseline_hash = state_hash(&fold(&ops));
        for perm in permutations(&ops) {
            let folded = fold(&perm);
            assert_eq!(
                folded.transcript("c1"),
                baseline,
                "transcript is delivery-order-independent"
            );
            assert_eq!(state_hash(&folded), baseline_hash);
        }
    }

    #[test]
    fn transcripts_are_partitioned_by_conversation_id() {
        let ops = {
            let mut v = one_device_conversation("work");
            let mut d = DeviceLog::new("dev-b");
            v.push(d.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::user_payload("home", "dinner?", 20),
            ));
            v
        };
        let state = fold(&ops);
        assert_eq!(
            state.conversation_ids(),
            vec!["home".to_string(), "work".to_string()]
        );
        assert_eq!(state.transcript("work").len(), 4);
        assert_eq!(state.transcript("home").len(), 1);
        assert!(state.transcript("nonexistent").is_empty());
    }

    #[test]
    fn legacy_speaker_text_turns_project_and_resume() {
        let mut d = DeviceLog::new("dev-a");
        let ops = vec![
            d.append(
                Scope::Personal,
                Surface::Conversation,
                json!({"speaker": "user", "text": "hi", "timestamp": 1}),
            ),
            d.append(
                Scope::Personal,
                Surface::Conversation,
                json!({"speaker": "assistant", "text": "hello", "timestamp": 2}),
            ),
        ];
        let state = fold(&ops);
        let turns = state.transcript(DEFAULT_CONVERSATION);
        assert_eq!(turns.len(), 2);
        assert_eq!(
            (turns[0].role, turns[1].role),
            (Role::User, Role::Assistant)
        );
        let ms = state.resume_messages(DEFAULT_CONVERSATION);
        assert_provider_valid(&ms);
        assert!(matches!(&ms[0], Message::User { content } if content == "hi"));
        assert!(matches!(&ms[1], Message::Assistant { content, .. } if content == "hello"));
    }

    #[test]
    fn lastn_compaction_keeps_the_last_n_in_order_and_round_trips() {
        // Conversation is an event stream (op_id-keyed) but INDEPENDENT, so
        // LastN is allowed and works. After LastN{2} the resumable transcript is
        // the retained snapshot window + live tail, in order — the same folded
        // state compaction produced (0.25 incoherence cannot recur).
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let mut ops = vec![
            a.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::user_payload("c1", "t0", 100),
            ),
            a.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::assistant_payload("c1", "t1", vec![], 101),
            ),
            a.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::user_payload("c1", "t2", 102),
            ),
            a.append(
                Scope::Personal,
                Surface::Conversation,
                Turn::assistant_payload("c1", "t3", vec![], 103),
            ),
        ];
        let split = ops.len();
        for op in &ops {
            b.observe(&op.hlc);
        }
        ops.push(b.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::user_payload("c1", "t4", 104),
        ));
        ops.push(b.append(
            Scope::Personal,
            Surface::Conversation,
            Turn::assistant_payload("c1", "t5", vec![], 105),
        ));

        let frontier = ops[..split].iter().map(|o| o.hlc.clone()).max().unwrap();
        let mut acks = AckTable::new();
        for op in &ops {
            acks.ack(op.device_id.clone(), frontier.clone());
        }
        let mut policy = RetentionPolicy::keep_all();
        policy
            .rules
            .insert("conversation".to_string(), RetentionRule::LastN { n: 2 });
        let plan = plan_compaction(&ops, &acks, &policy, Some(1_000)).unwrap();

        let ckpt = plan.checkpoint.state.transcript("c1");
        assert_eq!(
            ckpt.iter().map(|t| t.content.as_str()).collect::<Vec<_>>(),
            vec!["t2", "t3"]
        );

        // Checkpoint round-trip: serialize+deserialize; transcript unchanged.
        let ckpt_json = serde_json::to_string(&plan.checkpoint.state).unwrap();
        let ckpt_back: SyncState = serde_json::from_str(&ckpt_json).unwrap();
        assert_eq!(
            ckpt_back.transcript("c1"),
            plan.checkpoint.state.transcript("c1")
        );

        let reconstructed = fold_onto(&plan.checkpoint.state, &plan.retained_ops);
        let resumed: Vec<String> = reconstructed
            .transcript("c1")
            .iter()
            .map(|t| t.content.clone())
            .collect();
        assert_eq!(
            resumed,
            vec!["t2", "t3", "t4", "t5"],
            "resume = retained window + live tail, in order"
        );
        assert_provider_valid(&reconstructed.resume_messages("c1"));

        let (global, _) = apply_retention(&fold(&ops), &policy, 1_000).unwrap();
        let (local, _) = apply_retention(&reconstructed, &policy, 1_000).unwrap();
        assert_eq!(local.transcript("c1"), global.transcript("c1"));
        assert_eq!(
            global
                .transcript("c1")
                .iter()
                .map(|t| t.content.clone())
                .collect::<Vec<_>>(),
            vec!["t4", "t5"],
            "the last-N display window is the same on every device"
        );
        assert_eq!(state_hash(&local), state_hash(&global));
    }

    /// Heap's algorithm — every permutation, no rand dependency.
    fn permutations<T: Clone>(items: &[T]) -> Vec<Vec<T>> {
        fn heap<T: Clone>(k: usize, arr: &mut Vec<T>, out: &mut Vec<Vec<T>>) {
            if k == 1 {
                out.push(arr.clone());
                return;
            }
            for i in 0..k {
                heap(k - 1, arr, out);
                if k.is_multiple_of(2) {
                    arr.swap(i, k - 1);
                } else {
                    arr.swap(0, k - 1);
                }
            }
        }
        let mut arr = items.to_vec();
        let mut out = Vec::new();
        heap(arr.len(), &mut arr, &mut out);
        out
    }
}