use super::heal::{heal_legacy_compaction_rewrite_semantics, heal_legacy_revision_strings};
use super::sealed::ValidatedTranscriptHistory;
use super::validate::{
validate_released_0810_transcript_rewrite_record, validate_transcript_history_state,
validate_transcript_rewrite_record,
};
use crate::session::{
TranscriptEditError, TranscriptRewriteReason, TranscriptRewriteSelection,
transcript_messages_digest,
};
use crate::session_store::SessionMessageRowPrefixAccumulator;
use crate::time_compat::SystemTime;
use crate::types::Message;
use serde::{Deserialize, Deserializer, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub struct TranscriptRewriteCommit {
#[serde(default, skip_serializing_if = "rewrite_generation_is_unknown")]
pub rewrite_generation: u64,
pub parent_revision: String,
pub revision: String,
pub selection: TranscriptRewriteSelection,
pub original_span_digest: String,
pub replacement_digest: String,
pub messages_before: usize,
pub messages_after: usize,
pub reason: TranscriptRewriteReason,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub actor: Option<String>,
#[cfg_attr(feature = "schema", schemars(with = "SchemaSystemTime"))]
pub committed_at: SystemTime,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
pub enum TranscriptRewriteParentTransition {
ExactAppend,
ExactSplice,
}
fn rewrite_generation_is_unknown(generation: &u64) -> bool {
*generation == 0
}
#[cfg(test)]
fn normalize_legacy_graph_rewrite_generations(
commits: &mut [TranscriptRewriteCommit],
) -> Result<bool, TranscriptEditError> {
if commits.is_empty() {
return Ok(false);
}
let zero_count = commits
.iter()
.filter(|commit| commit.rewrite_generation == 0)
.count();
if zero_count == commits.len() {
for (index, commit) in commits.iter_mut().enumerate() {
commit.rewrite_generation = u64::try_from(index)
.ok()
.and_then(|index| index.checked_add(1))
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"transcript rewrite generation exceeds u64".to_string(),
)
})?;
}
return Ok(true);
}
if zero_count != 0 {
return Err(TranscriptEditError::HistoryStateMalformed(
"transcript rewrite generations mix 0.8.10 zero markers with current occurrence identities"
.to_string(),
));
}
for (index, commit) in commits.iter().enumerate() {
let expected = u64::try_from(index)
.ok()
.and_then(|index| index.checked_add(1))
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"transcript rewrite generation exceeds u64".to_string(),
)
})?;
if commit.rewrite_generation != expected {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"transcript rewrite generation {} is not the expected contiguous occurrence {expected}",
commit.rewrite_generation
)));
}
}
Ok(false)
}
const TRANSCRIPT_REWRITE_PREFIX_CHAIN_DOMAIN: &[u8] =
b"meerkat.transcript-rewrite-prefix.chain.v1\0";
const TRANSCRIPT_REWRITE_PREFIX_STEP_DOMAIN: &[u8] = b"meerkat.transcript-rewrite-prefix.step.v1\0";
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
pub struct TranscriptRewritePrefixAccumulator {
occurrence_count: u64,
digest: String,
}
impl<'de> Deserialize<'de> for TranscriptRewritePrefixAccumulator {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
struct Wire {
occurrence_count: u64,
digest: String,
}
let wire = Wire::deserialize(deserializer)?;
let accumulator = Self {
occurrence_count: wire.occurrence_count,
digest: wire.digest,
};
if accumulator.raw_digest().is_none() {
return Err(serde::de::Error::custom(
"rewrite-prefix digest must be canonical sha256:<64 lowercase hex>",
));
}
Ok(accumulator)
}
}
impl TranscriptRewritePrefixAccumulator {
#[must_use]
pub fn empty() -> Self {
let mut hasher = Sha256::new();
hasher.update(TRANSCRIPT_REWRITE_PREFIX_CHAIN_DOMAIN);
Self {
occurrence_count: 0,
digest: format!("sha256:{:x}", hasher.finalize()),
}
}
pub fn from_commits(commits: &[TranscriptRewriteCommit]) -> Result<Self, serde_json::Error> {
let mut accumulator = Self::empty();
for commit in commits {
accumulator = accumulator.extend(commit)?;
}
Ok(accumulator)
}
pub fn extend(&self, commit: &TranscriptRewriteCommit) -> Result<Self, serde_json::Error> {
let expected_generation = self.occurrence_count.checked_add(1).ok_or_else(|| {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"rewrite-prefix occurrence count exceeds u64",
))
})?;
if commit.rewrite_generation != expected_generation {
return Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"rewrite-prefix occurrence generation {} is not the expected {expected_generation}",
commit.rewrite_generation
),
)));
}
let bytes = serde_json::to_vec(commit)?;
let previous = self.raw_digest().ok_or_else(|| {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"rewrite-prefix accumulator is not a canonical sha256 digest",
))
})?;
let mut hasher = Sha256::new();
hasher.update(TRANSCRIPT_REWRITE_PREFIX_STEP_DOMAIN);
hasher.update(previous);
hasher.update((bytes.len() as u64).to_be_bytes());
hasher.update(bytes);
Ok(Self {
occurrence_count: expected_generation,
digest: format!("sha256:{:x}", hasher.finalize()),
})
}
#[must_use]
pub const fn occurrence_count(&self) -> u64 {
self.occurrence_count
}
#[must_use]
pub fn digest(&self) -> &str {
&self.digest
}
fn raw_digest(&self) -> Option<[u8; 32]> {
let encoded = self.digest.strip_prefix("sha256:")?;
if encoded.len() != 64 || !encoded.is_ascii() {
return None;
}
let bytes = encoded.as_bytes();
let mut decoded = [0u8; 32];
for (index, output) in decoded.iter_mut().enumerate() {
let high = hex_nibble(bytes[index * 2])?;
let low = hex_nibble(bytes[index * 2 + 1])?;
*output = (high << 4) | low;
}
Some(decoded)
}
}
impl Default for TranscriptRewritePrefixAccumulator {
fn default() -> Self {
Self::empty()
}
}
fn hex_nibble(value: u8) -> Option<u8> {
match value {
b'0'..=b'9' => Some(value - b'0'),
b'a'..=b'f' => Some(value - b'a' + 10),
_ => None,
}
}
pub fn transcript_rewrite_prefix_digest(
commits: &[TranscriptRewriteCommit],
) -> Result<String, serde_json::Error> {
Ok(TranscriptRewritePrefixAccumulator::from_commits(commits)?
.digest()
.to_string())
}
pub fn extend_transcript_rewrite_prefix_accumulator(
mut accumulator: TranscriptRewritePrefixAccumulator,
commits: &[TranscriptRewriteCommit],
) -> Result<TranscriptRewritePrefixAccumulator, serde_json::Error> {
for commit in commits {
accumulator = accumulator.extend(commit)?;
}
Ok(accumulator)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
pub struct TranscriptRevisionBody {
pub revision: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent_revision: Option<String>,
#[cfg_attr(feature = "schema", schemars(with = "Vec<serde_json::Value>"))]
pub messages: Vec<Message>,
#[cfg_attr(feature = "schema", schemars(with = "SchemaSystemTime"))]
pub created_at: SystemTime,
}
#[derive(Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
struct Released0810RevisionEntry {
revision: String,
#[serde(default)]
parent_revision: Option<String>,
created_at: SystemTime,
#[serde(default)]
messages: Option<Vec<Message>>,
#[serde(default)]
rebase: Option<Released0810RevisionRebase>,
}
#[derive(Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
struct Released0810RevisionRebase {
base: String,
at: usize,
removed: usize,
#[serde(default)]
insert: Vec<Message>,
}
#[derive(Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
struct Released0810Commit {
parent_revision: String,
revision: String,
selection: TranscriptRewriteSelection,
original_span_digest: String,
replacement_digest: String,
messages_before: usize,
messages_after: usize,
reason: TranscriptRewriteReason,
#[serde(default)]
actor: Option<String>,
committed_at: SystemTime,
}
#[derive(Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
struct Released0810HistoryWire {
head: String,
#[serde(default)]
commits: Vec<Released0810Commit>,
revisions: Vec<Released0810RevisionEntry>,
digest_format: u32,
#[serde(default, rename = "replay_cursor")]
_replay_cursor: Option<serde::de::IgnoredAny>,
}
impl From<Released0810Commit> for TranscriptRewriteCommit {
fn from(released: Released0810Commit) -> Self {
Self {
rewrite_generation: 0,
parent_revision: released.parent_revision,
revision: released.revision,
selection: released.selection,
original_span_digest: released.original_span_digest,
replacement_digest: released.replacement_digest,
messages_before: released.messages_before,
messages_after: released.messages_after,
reason: released.reason,
actor: released.actor,
committed_at: released.committed_at,
}
}
}
fn decode_released_0810_revision_chain<E>(
entries: Vec<Released0810RevisionEntry>,
) -> Result<Vec<TranscriptRevisionBody>, E>
where
E: serde::de::Error,
{
let mut materialized: std::collections::HashMap<String, usize> =
std::collections::HashMap::with_capacity(entries.len());
let mut bodies: Vec<TranscriptRevisionBody> = Vec::with_capacity(entries.len());
for entry in entries {
let messages = match (entry.messages, entry.rebase) {
(Some(messages), None) => messages,
(None, Some(rebase)) => {
let base = materialized
.get(&rebase.base)
.and_then(|index| bodies.get(*index))
.ok_or_else(|| {
E::custom(format!(
"released 0.8.10 revision {} rebases on missing {}",
entry.revision, rebase.base
))
})?;
let end = rebase
.at
.checked_add(rebase.removed)
.filter(|end| *end <= base.messages.len())
.ok_or_else(|| {
E::custom(format!(
"released 0.8.10 revision {} splice exceeds base {}",
entry.revision, rebase.base
))
})?;
let mut messages =
Vec::with_capacity(base.messages.len() - rebase.removed + rebase.insert.len());
messages.extend_from_slice(&base.messages[..rebase.at]);
messages.extend(rebase.insert);
messages.extend_from_slice(&base.messages[end..]);
messages
}
(Some(_), Some(_)) => {
return Err(E::custom(format!(
"released 0.8.10 revision {} carries messages and rebase",
entry.revision
)));
}
(None, None) => {
return Err(E::custom(format!(
"released 0.8.10 revision {} carries neither messages nor rebase",
entry.revision
)));
}
};
let position = bodies.len();
if materialized
.insert(entry.revision.clone(), position)
.is_some()
{
return Err(E::custom(format!(
"released 0.8.10 revision vector repeats deduplicated id {}",
entry.revision
)));
}
bodies.push(TranscriptRevisionBody {
revision: entry.revision,
parent_revision: entry.parent_revision,
messages,
created_at: entry.created_at,
});
}
Ok(bodies)
}
struct ValidatedReleased0810Wire {
head: String,
commits: Vec<TranscriptRewriteCommit>,
revisions: Vec<TranscriptRevisionBody>,
}
fn validate_released_0810_wire(
wire: Released0810HistoryWire,
) -> Result<ValidatedReleased0810Wire, serde_json::Error> {
if wire.digest_format != TRANSCRIPT_DIGEST_FORMAT_RELEASED_0810 {
return Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"released 0.8.10 transcript graph requires digest format {}, observed {}",
TRANSCRIPT_DIGEST_FORMAT_RELEASED_0810, wire.digest_format
),
)));
}
if wire.commits.is_empty() || wire.revisions.is_empty() {
return Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"released 0.8.10 transcript graph must carry commits and bodies",
)));
}
let commits = wire
.commits
.into_iter()
.map(TranscriptRewriteCommit::from)
.collect::<Vec<_>>();
let revisions = decode_released_0810_revision_chain::<serde_json::Error>(wire.revisions)?;
let bodies_by_revision = revisions
.iter()
.map(|body| (body.revision.as_str(), body))
.collect::<std::collections::HashMap<_, _>>();
if !bodies_by_revision.contains_key(wire.head.as_str()) {
return Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"released 0.8.10 transcript graph omits advertised head {}",
wire.head
),
)));
}
for commit in &commits {
let parent = bodies_by_revision
.get(commit.parent_revision.as_str())
.copied()
.ok_or_else(|| {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"released 0.8.10 graph omits parent body {}",
commit.parent_revision
),
))
})?;
let revision = bodies_by_revision
.get(commit.revision.as_str())
.copied()
.ok_or_else(|| {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"released 0.8.10 graph omits revision body {}",
commit.revision
),
))
})?;
validate_released_0810_transcript_rewrite_record(commit, parent, revision)
.map_err(|error| serde_json::Error::io(std::io::Error::other(error.to_string())))?;
}
Ok(ValidatedReleased0810Wire {
head: wire.head,
commits,
revisions,
})
}
pub(crate) fn validate_current_transcript_history_slice(
bytes: &[u8],
) -> Result<usize, serde_json::Error> {
#[derive(Deserialize)]
struct FormatProbe {
#[serde(default)]
format: Option<String>,
}
match serde_json::from_slice::<FormatProbe>(bytes)?.format {
Some(format) if format == TRANSCRIPT_HISTORY_FORMAT_CURRENT => {
let state: TranscriptHistoryState = serde_json::from_slice(bytes)?;
Ok(state.commit_count())
}
Some(format) => Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("unsupported transcript graph format {format}"),
))),
None => Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"bounded current-graph validation refuses released full-body history",
))),
}
}
#[cfg(feature = "schema")]
#[allow(dead_code)]
#[derive(schemars::JsonSchema)]
#[schemars(rename = "SystemTime")]
struct SchemaSystemTime {
secs_since_epoch: u64,
nanos_since_epoch: u32,
}
pub const TRANSCRIPT_HISTORY_FORMAT_CURRENT: &str = "anchor_occurrence_edges_v1";
const TRANSCRIPT_GRAPH_PREFIX_ANCHOR_DOMAIN: &[u8] = b"meerkat.transcript-history.anchor.v1\0";
const TRANSCRIPT_GRAPH_PREFIX_EDGE_DOMAIN: &[u8] = b"meerkat.transcript-history.edge.v1\0";
static TRANSCRIPT_HISTORY_FULL_BODY_MATERIALIZATIONS: std::sync::atomic::AtomicU64 =
std::sync::atomic::AtomicU64::new(0);
#[must_use]
pub fn transcript_history_full_body_materializations() -> u64 {
TRANSCRIPT_HISTORY_FULL_BODY_MATERIALIZATIONS.load(std::sync::atomic::Ordering::Relaxed)
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub struct TranscriptRevisionAnchor {
revision: String,
#[cfg_attr(feature = "schema", schemars(with = "Vec<serde_json::Value>"))]
messages: Vec<Message>,
row_prefix: SessionMessageRowPrefixAccumulator,
#[cfg_attr(feature = "schema", schemars(with = "SchemaSystemTime"))]
created_at: SystemTime,
}
impl TranscriptRevisionAnchor {
#[must_use]
pub fn revision(&self) -> &str {
&self.revision
}
#[must_use]
pub fn messages(&self) -> &[Message] {
&self.messages
}
#[must_use]
pub fn row_prefix(&self) -> &SessionMessageRowPrefixAccumulator {
&self.row_prefix
}
#[must_use]
pub const fn created_at(&self) -> SystemTime {
self.created_at
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum TranscriptParentAdvance {
ExactAppend {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
#[cfg_attr(feature = "schema", schemars(with = "Vec<serde_json::Value>"))]
appended: Vec<Message>,
},
ExactSplice {
at: usize,
#[cfg_attr(feature = "schema", schemars(with = "Vec<serde_json::Value>"))]
replacement: Vec<Message>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
#[cfg_attr(feature = "schema", schemars(with = "Vec<serde_json::Value>"))]
appended: Vec<Message>,
},
}
impl TranscriptParentAdvance {
#[must_use]
pub fn appended(&self) -> &[Message] {
match self {
Self::ExactAppend { appended } | Self::ExactSplice { appended, .. } => appended,
}
}
#[must_use]
pub fn exact_splice(&self) -> Option<(usize, &[Message])> {
match self {
Self::ExactAppend { .. } => None,
Self::ExactSplice {
at, replacement, ..
} => Some((*at, replacement)),
}
}
#[must_use]
pub fn transition(&self) -> TranscriptRewriteParentTransition {
match self {
Self::ExactAppend { .. } => TranscriptRewriteParentTransition::ExactAppend,
Self::ExactSplice { .. } => TranscriptRewriteParentTransition::ExactSplice,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub struct TranscriptRewritePatch {
at: usize,
#[cfg_attr(feature = "schema", schemars(with = "Vec<serde_json::Value>"))]
replacement: Vec<Message>,
}
impl TranscriptRewritePatch {
#[must_use]
pub const fn at(&self) -> usize {
self.at
}
#[must_use]
pub fn replacement(&self) -> &[Message] {
&self.replacement
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub struct TranscriptEndpointWitness {
message_count: usize,
row_prefix: SessionMessageRowPrefixAccumulator,
}
impl TranscriptEndpointWitness {
pub(in crate::session) fn from_messages(
messages: &[Message],
) -> Result<Self, serde_json::Error> {
let row_prefix = SessionMessageRowPrefixAccumulator::from_messages(messages)
.map_err(session_store_error_as_json)?;
Self::from_messages_with_row_prefix(messages, row_prefix)
}
pub(in crate::session) fn from_messages_with_row_prefix(
messages: &[Message],
row_prefix: SessionMessageRowPrefixAccumulator,
) -> Result<Self, serde_json::Error> {
if row_prefix.row_count() != messages.len() as u64 {
return Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"endpoint row lineage count differs from its materialized messages",
)));
}
Ok(Self {
message_count: messages.len(),
row_prefix,
})
}
#[must_use]
pub const fn message_count(&self) -> usize {
self.message_count
}
#[must_use]
pub fn row_prefix(&self) -> &SessionMessageRowPrefixAccumulator {
&self.row_prefix
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub struct TranscriptRevisionEdge {
commit: TranscriptRewriteCommit,
rewrite_prefix: TranscriptRewritePrefixAccumulator,
base_revision: String,
messages_before_base: usize,
parent_advance: TranscriptParentAdvance,
parent_row_prefix: SessionMessageRowPrefixAccumulator,
rewrite: TranscriptRewritePatch,
result_witness: TranscriptEndpointWitness,
#[cfg_attr(feature = "schema", schemars(with = "SchemaSystemTime"))]
parent_created_at: SystemTime,
#[cfg_attr(feature = "schema", schemars(with = "SchemaSystemTime"))]
revision_created_at: SystemTime,
}
impl TranscriptRevisionEdge {
#[doc(hidden)]
pub fn from_replay_bytes(bytes: &[u8]) -> Result<Self, serde_json::Error> {
serde_json::from_slice(bytes)
}
#[doc(hidden)]
pub fn to_replay_bytes(&self) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(self)
}
#[must_use]
pub fn commit(&self) -> &TranscriptRewriteCommit {
&self.commit
}
#[must_use]
pub fn rewrite_prefix(&self) -> &TranscriptRewritePrefixAccumulator {
&self.rewrite_prefix
}
#[must_use]
pub const fn rewrite_generation(&self) -> u64 {
self.commit.rewrite_generation
}
#[must_use]
pub fn base_revision(&self) -> &str {
&self.base_revision
}
#[must_use]
pub fn parent_revision(&self) -> &str {
&self.commit.parent_revision
}
#[must_use]
pub fn revision(&self) -> &str {
&self.commit.revision
}
#[must_use]
pub const fn messages_before_base(&self) -> usize {
self.messages_before_base
}
#[must_use]
pub const fn messages_before(&self) -> usize {
self.commit.messages_before
}
#[must_use]
pub const fn messages_after(&self) -> usize {
self.commit.messages_after
}
#[must_use]
pub fn parent_advance(&self) -> &TranscriptParentAdvance {
&self.parent_advance
}
#[must_use]
pub fn parent_row_prefix(&self) -> &SessionMessageRowPrefixAccumulator {
&self.parent_row_prefix
}
#[must_use]
pub fn rewrite(&self) -> &TranscriptRewritePatch {
&self.rewrite
}
#[must_use]
pub fn result_witness(&self) -> &TranscriptEndpointWitness {
&self.result_witness
}
#[must_use]
pub const fn parent_created_at(&self) -> SystemTime {
self.parent_created_at
}
#[must_use]
pub const fn revision_created_at(&self) -> SystemTime {
self.revision_created_at
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
pub struct TranscriptGraphPrefixAccumulator {
occurrence_count: u64,
digest: String,
}
impl TranscriptGraphPrefixAccumulator {
fn from_anchor(anchor: &TranscriptRevisionAnchor) -> Result<Self, serde_json::Error> {
let mut hasher = Sha256::new();
hasher.update(TRANSCRIPT_GRAPH_PREFIX_ANCHOR_DOMAIN);
update_graph_prefix_field(&mut hasher, b"revision", &anchor.revision)?;
update_graph_prefix_field(&mut hasher, b"messages", &anchor.messages)?;
update_graph_prefix_field(&mut hasher, b"row_prefix", &anchor.row_prefix)?;
update_graph_prefix_field(&mut hasher, b"created_at", &anchor.created_at)?;
Ok(Self {
occurrence_count: 0,
digest: format!("sha256:{:x}", hasher.finalize()),
})
}
fn extend(&self, edge: &TranscriptRevisionEdge) -> Result<Self, serde_json::Error> {
let expected = self.occurrence_count.checked_add(1).ok_or_else(|| {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"transcript graph occurrence count exceeds u64",
))
})?;
if edge.commit.rewrite_generation != expected {
return Err(serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"transcript graph edge generation {} is not expected {expected}",
edge.commit.rewrite_generation
),
)));
}
let previous = decode_canonical_sha256(&self.digest).ok_or_else(|| {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"transcript graph prefix is not canonical sha256",
))
})?;
let mut hasher = Sha256::new();
hasher.update(TRANSCRIPT_GRAPH_PREFIX_EDGE_DOMAIN);
hasher.update(previous);
update_graph_prefix_field(&mut hasher, b"commit", &edge.commit)?;
update_graph_prefix_field(&mut hasher, b"rewrite_prefix", &edge.rewrite_prefix)?;
update_graph_prefix_field(&mut hasher, b"base_revision", &edge.base_revision)?;
update_graph_prefix_field(
&mut hasher,
b"messages_before_base",
&edge.messages_before_base,
)?;
update_graph_prefix_field(&mut hasher, b"parent_advance", &edge.parent_advance)?;
update_graph_prefix_field(&mut hasher, b"parent_row_prefix", &edge.parent_row_prefix)?;
update_graph_prefix_field(&mut hasher, b"rewrite", &edge.rewrite)?;
update_graph_prefix_field(&mut hasher, b"result_witness", &edge.result_witness)?;
update_graph_prefix_field(&mut hasher, b"parent_created_at", &edge.parent_created_at)?;
update_graph_prefix_field(
&mut hasher,
b"revision_created_at",
&edge.revision_created_at,
)?;
Ok(Self {
occurrence_count: expected,
digest: format!("sha256:{:x}", hasher.finalize()),
})
}
pub(super) fn from_graph<'a>(
anchor: &TranscriptRevisionAnchor,
edges: impl IntoIterator<Item = &'a TranscriptRevisionEdge>,
) -> Result<Self, serde_json::Error> {
let mut prefix = Self::from_anchor(anchor)?;
for edge in edges {
prefix = prefix.extend(edge)?;
}
Ok(prefix)
}
#[must_use]
pub const fn occurrence_count(&self) -> u64 {
self.occurrence_count
}
#[must_use]
pub fn digest(&self) -> &str {
&self.digest
}
}
fn update_graph_prefix_field<T: Serialize + ?Sized>(
hasher: &mut Sha256,
label: &[u8],
value: &T,
) -> Result<(), serde_json::Error> {
let bytes = serde_json::to_vec(value)?;
hasher.update((label.len() as u64).to_be_bytes());
hasher.update(label);
hasher.update((bytes.len() as u64).to_be_bytes());
hasher.update(bytes);
Ok(())
}
fn session_store_error_as_json(error: crate::SessionStoreError) -> serde_json::Error {
serde_json::Error::io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
error.to_string(),
))
}
fn decode_canonical_sha256(value: &str) -> Option<[u8; 32]> {
let encoded = value.strip_prefix("sha256:")?;
if encoded.len() != 64 || !encoded.is_ascii() {
return None;
}
let mut decoded = [0u8; 32];
for (index, output) in decoded.iter_mut().enumerate() {
let bytes = encoded.as_bytes();
*output = (hex_nibble(bytes[index * 2])? << 4) | hex_nibble(bytes[index * 2 + 1])?;
}
Some(decoded)
}
#[derive(Debug, Clone, Serialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
pub struct TranscriptRewriteRecord {
pub commit: TranscriptRewriteCommit,
pub parent_body: TranscriptRevisionBody,
pub revision_body: TranscriptRevisionBody,
#[serde(default, skip_serializing_if = "digest_format_is_unknown")]
pub digest_format: u32,
}
impl<'de> Deserialize<'de> for TranscriptRewriteRecord {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(rename_all = "snake_case")]
struct Wire {
commit: TranscriptRewriteCommit,
parent_body: TranscriptRevisionBody,
revision_body: TranscriptRevisionBody,
#[serde(default)]
digest_format: u32,
}
let wire = Wire::deserialize(deserializer)?;
crate::digest_observability::record_rewrite_record_body_decode();
let mut revisions = vec![wire.parent_body, wire.revision_body];
let mut commits = vec![wire.commit];
match wire.digest_format {
0 => {
heal_legacy_revision_strings(&mut revisions, &mut commits, None)
.map_err(serde::de::Error::custom)?;
heal_legacy_compaction_rewrite_semantics(&mut commits, &revisions);
}
TRANSCRIPT_DIGEST_FORMAT_CURRENT => {}
unsupported => {
return Err(serde::de::Error::custom(format!(
"current transcript rewrite record digest format {unsupported} is not supported"
)));
}
}
let mut revisions = revisions.into_iter();
let parent_body = revisions
.next()
.ok_or_else(|| serde::de::Error::custom("rewrite record lost its parent body"))?;
let revision_body = revisions
.next()
.ok_or_else(|| serde::de::Error::custom("rewrite record lost its revision body"))?;
let commit = commits
.into_iter()
.next()
.ok_or_else(|| serde::de::Error::custom("rewrite record lost its commit"))?;
Ok(Self {
commit,
parent_body,
revision_body,
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
})
}
}
impl TranscriptRewriteRecord {
pub fn validate(&self) -> Result<(), TranscriptEditError> {
if self.commit.rewrite_generation == 0 {
return Err(TranscriptEditError::HistoryStateMalformed(
"current transcript rewrite records require a non-zero occurrence generation"
.to_string(),
));
}
validate_transcript_rewrite_record(&self.commit, &self.parent_body, &self.revision_body)
}
pub fn new(
commit: TranscriptRewriteCommit,
parent_body: TranscriptRevisionBody,
revision_body: TranscriptRevisionBody,
) -> Result<Self, TranscriptEditError> {
let record = Self {
commit,
parent_body,
revision_body,
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
};
record.validate()?;
Ok(record)
}
}
#[derive(Debug)]
struct TranscriptEdgeNode {
previous: Option<Arc<TranscriptEdgeNode>>,
edge: Arc<TranscriptRevisionEdge>,
graph_prefix: TranscriptGraphPrefixAccumulator,
}
#[derive(Debug)]
struct TranscriptEdgeChain {
tail: Option<Arc<TranscriptEdgeNode>>,
len: usize,
ordered: std::sync::OnceLock<Vec<Arc<TranscriptRevisionEdge>>>,
}
#[derive(Debug, Clone)]
struct PersistentTranscriptEdges {
chain: Arc<TranscriptEdgeChain>,
}
impl PersistentTranscriptEdges {
fn empty() -> Self {
Self {
chain: Arc::new(TranscriptEdgeChain {
tail: None,
len: 0,
ordered: std::sync::OnceLock::new(),
}),
}
}
fn from_vec(
anchor: &TranscriptRevisionAnchor,
edges: Vec<TranscriptRevisionEdge>,
) -> Result<Self, serde_json::Error> {
let mut persistent = Self::empty();
let mut graph_prefix = TranscriptGraphPrefixAccumulator::from_anchor(anchor)?;
for edge in edges {
graph_prefix = graph_prefix.extend(&edge)?;
persistent.push(edge, graph_prefix.clone());
}
Ok(persistent)
}
fn push(
&mut self,
edge: TranscriptRevisionEdge,
graph_prefix: TranscriptGraphPrefixAccumulator,
) {
let node = Arc::new(TranscriptEdgeNode {
previous: self.chain.tail.clone(),
edge: Arc::new(edge),
graph_prefix,
});
self.chain = Arc::new(TranscriptEdgeChain {
tail: Some(node),
len: self.chain.len + 1,
ordered: std::sync::OnceLock::new(),
});
}
fn len(&self) -> usize {
self.chain.len
}
fn last(&self) -> Option<&TranscriptRevisionEdge> {
self.chain.tail.as_ref().map(|node| node.edge.as_ref())
}
fn ordered(&self) -> &[Arc<TranscriptRevisionEdge>] {
self.chain.ordered.get_or_init(|| {
let mut ordered = Vec::with_capacity(self.chain.len);
let mut cursor = self.chain.tail.clone();
while let Some(node) = cursor {
ordered.push(Arc::clone(&node.edge));
cursor = node.previous.clone();
}
ordered.reverse();
ordered
})
}
fn get(&self, index: usize) -> Option<&TranscriptRevisionEdge> {
if index >= self.chain.len {
return None;
}
let mut cursor = self.chain.tail.as_deref();
for _ in index + 1..self.chain.len {
cursor = cursor?.previous.as_deref();
}
cursor.map(|node| node.edge.as_ref())
}
fn graph_prefix_at(&self, edge_count: usize) -> Option<&TranscriptGraphPrefixAccumulator> {
if edge_count == 0 || edge_count > self.chain.len {
return None;
}
let mut cursor = self.chain.tail.as_deref();
for _ in edge_count..self.chain.len {
cursor = cursor?.previous.as_deref();
}
cursor.map(|node| &node.graph_prefix)
}
fn suffix(&self, start: usize) -> Option<Vec<Arc<TranscriptRevisionEdge>>> {
if start > self.chain.len {
return None;
}
let suffix_len = self.chain.len - start;
let mut suffix = Vec::with_capacity(suffix_len);
let mut cursor = self.chain.tail.clone();
for _ in 0..suffix_len {
let node = cursor?;
suffix.push(Arc::clone(&node.edge));
cursor = node.previous.clone();
}
suffix.reverse();
Some(suffix)
}
fn prefix(&self, edge_count: usize) -> Option<Self> {
if edge_count == 0 || edge_count > self.chain.len {
return None;
}
let mut tail = self.chain.tail.clone();
for _ in edge_count..self.chain.len {
tail = tail?.previous.clone();
}
Some(Self {
chain: Arc::new(TranscriptEdgeChain {
tail,
len: edge_count,
ordered: std::sync::OnceLock::new(),
}),
})
}
}
#[derive(Debug, Clone)]
pub struct TranscriptHistoryState {
format: &'static str,
anchor: Arc<TranscriptRevisionAnchor>,
edges: PersistentTranscriptEdges,
rewrite_prefix: TranscriptRewritePrefixAccumulator,
graph_prefix: TranscriptGraphPrefixAccumulator,
digest_format: u32,
}
#[cfg(feature = "schema")]
#[allow(dead_code)]
#[derive(schemars::JsonSchema)]
#[schemars(rename = "TranscriptHistoryState")]
struct SchemaTranscriptHistoryState {
format: String,
anchor: TranscriptRevisionAnchor,
edges: Vec<TranscriptRevisionEdge>,
rewrite_prefix: TranscriptRewritePrefixAccumulator,
graph_prefix: TranscriptGraphPrefixAccumulator,
digest_format: u32,
}
#[cfg(feature = "schema")]
impl schemars::JsonSchema for TranscriptHistoryState {
fn schema_name() -> std::borrow::Cow<'static, str> {
"TranscriptHistoryState".into()
}
fn json_schema(generator: &mut schemars::SchemaGenerator) -> schemars::Schema {
<SchemaTranscriptHistoryState as schemars::JsonSchema>::json_schema(generator)
}
}
fn digest_format_is_unknown(format: &u32) -> bool {
*format == 0
}
pub(crate) const TRANSCRIPT_DIGEST_FORMAT_RELEASED_0810: u32 = 2;
pub(crate) const TRANSCRIPT_DIGEST_FORMAT_CURRENT: u32 = 3;
impl Serialize for TranscriptHistoryState {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
use serde::ser::SerializeStruct as _;
let mut wire = serializer.serialize_struct("TranscriptHistoryState", 6)?;
wire.serialize_field("format", self.format)?;
wire.serialize_field("anchor", self.anchor.as_ref())?;
wire.serialize_field("edges", self.edges.ordered())?;
wire.serialize_field("rewrite_prefix", &self.rewrite_prefix)?;
wire.serialize_field("graph_prefix", &self.graph_prefix)?;
wire.serialize_field("digest_format", &self.digest_format)?;
wire.end()
}
}
impl<'de> Deserialize<'de> for TranscriptHistoryState {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(rename_all = "snake_case", deny_unknown_fields)]
struct CurrentWire {
format: String,
anchor: TranscriptRevisionAnchor,
edges: Vec<TranscriptRevisionEdge>,
rewrite_prefix: TranscriptRewritePrefixAccumulator,
graph_prefix: TranscriptGraphPrefixAccumulator,
digest_format: u32,
}
let wire = CurrentWire::deserialize(deserializer)?;
if wire.format != TRANSCRIPT_HISTORY_FORMAT_CURRENT {
return Err(serde::de::Error::custom(format!(
"unsupported current transcript graph format {}",
wire.format
)));
}
if wire.digest_format != TRANSCRIPT_DIGEST_FORMAT_CURRENT {
return Err(serde::de::Error::custom(format!(
"current transcript graph digest format {} is not supported",
wire.digest_format
)));
}
let persistent_edges = PersistentTranscriptEdges::from_vec(&wire.anchor, wire.edges)
.map_err(serde::de::Error::custom)?;
let state = TranscriptHistoryState {
format: TRANSCRIPT_HISTORY_FORMAT_CURRENT,
anchor: Arc::new(wire.anchor),
edges: persistent_edges,
rewrite_prefix: wire.rewrite_prefix,
graph_prefix: wire.graph_prefix,
digest_format: wire.digest_format,
};
validate_transcript_history_state(&state).map_err(serde::de::Error::custom)?;
Ok(state)
}
}
pub(crate) fn import_released_0810_history(
value: serde_json::Value,
) -> Result<Option<TranscriptHistoryState>, serde_json::Error> {
let wire: Released0810HistoryWire = serde_json::from_value(value)?;
let ValidatedReleased0810Wire {
mut head,
mut commits,
mut revisions,
} = validate_released_0810_wire(wire)?;
remap_released_0810_digest_identities(&mut head, &mut commits, &mut revisions)?;
if commits.is_empty() {
return Ok(None);
}
TranscriptHistoryState::from_legacy_full_bodies(head, commits, revisions)
.map(Some)
.map_err(|error| serde_json::Error::io(std::io::Error::other(error.to_string())))
}
#[derive(Debug)]
#[doc(hidden)]
pub enum ProvenReleased0810RewriteRemap {
Retained(TranscriptRewriteRecord),
Collapsed {
current_revision: String,
},
}
#[doc(hidden)]
pub fn remap_proven_released_0810_rewrite_record(
mut commit: TranscriptRewriteCommit,
mut parent_body: TranscriptRevisionBody,
mut revision_body: TranscriptRevisionBody,
expected_generation: u64,
) -> Result<ProvenReleased0810RewriteRemap, TranscriptEditError> {
let malformed = |message: &str| TranscriptEditError::HistoryStateMalformed(message.to_string());
if commit.rewrite_generation != 0 || expected_generation == 0 {
return Err(malformed(
"released 0.8.10 rewrite remap requires generation zero and a non-zero target generation",
));
}
if decode_canonical_sha256(&parent_body.revision).is_none()
|| decode_canonical_sha256(&revision_body.revision).is_none()
|| decode_canonical_sha256(&commit.parent_revision).is_none()
|| decode_canonical_sha256(&commit.revision).is_none()
|| decode_canonical_sha256(&commit.original_span_digest).is_none()
|| decode_canonical_sha256(&commit.replacement_digest).is_none()
{
return Err(malformed(
"released 0.8.10 rewrite remap requires canonical sha256 labels",
));
}
validate_released_0810_transcript_rewrite_record(&commit, &parent_body, &revision_body)?;
if revision_body.parent_revision.as_deref() != Some(commit.parent_revision.as_str()) {
return Err(malformed(
"released 0.8.10 rewrite child body does not name its exact parent revision",
));
}
let current_parent = transcript_messages_digest(&parent_body.messages)
.map_err(|error| malformed(&error.to_string()))?;
let current_revision = transcript_messages_digest(&revision_body.messages)
.map_err(|error| malformed(&error.to_string()))?;
let (start, end) = commit.selection.bounds();
let removed = end
.checked_sub(start)
.ok_or_else(|| malformed("released 0.8.10 rewrite selection is inverted"))?;
let retained = commit
.messages_before
.checked_sub(removed)
.ok_or_else(|| malformed("released 0.8.10 rewrite removes more rows than its parent"))?;
let replacement_len = commit
.messages_after
.checked_sub(retained)
.ok_or_else(|| malformed("released 0.8.10 rewrite replacement length is invalid"))?;
let replacement_end = start
.checked_add(replacement_len)
.ok_or_else(|| malformed("released 0.8.10 rewrite replacement span overflows"))?;
let current_original_span = transcript_messages_digest(&parent_body.messages[start..end])
.map_err(|error| malformed(&error.to_string()))?;
let current_replacement_span =
transcript_messages_digest(&revision_body.messages[start..replacement_end])
.map_err(|error| malformed(&error.to_string()))?;
if current_parent == current_revision {
if current_original_span != current_replacement_span {
return Err(malformed(
"released 0.8.10 collapsed endpoints retain different selected-span identities",
));
}
return Ok(ProvenReleased0810RewriteRemap::Collapsed {
current_revision: current_parent,
});
}
commit.rewrite_generation = expected_generation;
commit.parent_revision = current_parent.clone();
commit.revision = current_revision.clone();
commit.original_span_digest = current_original_span;
commit.replacement_digest = current_replacement_span;
parent_body.revision = current_parent.clone();
parent_body.parent_revision = None;
revision_body.revision = current_revision;
revision_body.parent_revision = Some(current_parent);
TranscriptRewriteRecord::new(commit, parent_body, revision_body)
.map(ProvenReleased0810RewriteRemap::Retained)
}
fn remap_released_0810_digest_identities(
head: &mut String,
commits: &mut Vec<TranscriptRewriteCommit>,
revisions: &mut Vec<TranscriptRevisionBody>,
) -> Result<(), serde_json::Error> {
let mut remap = BTreeMap::new();
for body in revisions.iter() {
if decode_canonical_sha256(&body.revision).is_none() {
return Err(serde_json::Error::io(std::io::Error::other(
"released 0.8.10 transcript body carries a non-canonical revision label",
)));
}
let current = transcript_messages_digest(&body.messages)?;
remap.insert(body.revision.clone(), current);
}
let bodies_by_released_revision = revisions
.iter()
.map(|body| (body.revision.as_str(), body))
.collect::<std::collections::HashMap<_, _>>();
let mut remapped_commits = Vec::with_capacity(commits.len());
let mut collapsed_adjacency: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
for commit in commits.iter() {
let parent = bodies_by_released_revision
.get(commit.parent_revision.as_str())
.copied()
.ok_or_else(|| {
serde_json::Error::io(std::io::Error::other(
"released 0.8.10 commit parent body disappeared before remap",
))
})?;
let revision = bodies_by_released_revision
.get(commit.revision.as_str())
.copied()
.ok_or_else(|| {
serde_json::Error::io(std::io::Error::other(
"released 0.8.10 commit revision body disappeared before remap",
))
})?;
let generation = u64::try_from(remapped_commits.len())
.ok()
.and_then(|index| index.checked_add(1))
.ok_or_else(|| {
serde_json::Error::io(std::io::Error::other(
"released 0.8.10 rewrite generation overflowed",
))
})?;
let outcome = remap_proven_released_0810_rewrite_record(
commit.clone(),
parent.clone(),
revision.clone(),
generation,
)
.map_err(|error| serde_json::Error::io(std::io::Error::other(error.to_string())))?;
match outcome {
ProvenReleased0810RewriteRemap::Retained(record) => {
remapped_commits.push(record.commit);
}
ProvenReleased0810RewriteRemap::Collapsed { current_revision } => {
if remap.get(&commit.parent_revision) != Some(¤t_revision)
|| remap.get(&commit.revision) != Some(¤t_revision)
{
return Err(serde_json::Error::io(std::io::Error::other(
"released 0.8.10 collapsed rewrite disagrees with graph body remap",
)));
}
collapsed_adjacency
.entry(commit.parent_revision.clone())
.or_default()
.insert(commit.revision.clone());
collapsed_adjacency
.entry(commit.revision.clone())
.or_default()
.insert(commit.parent_revision.clone());
}
}
}
let mut current_owner: BTreeMap<String, String> = BTreeMap::new();
for (released, current) in &remap {
if let Some(previous) = current_owner.insert(current.clone(), released.clone())
&& !released_ids_connected_by_collapsed_rewrites(
&collapsed_adjacency,
&previous,
released,
)
{
return Err(serde_json::Error::io(std::io::Error::other(
"released 0.8.10 transcript bodies collapse without a structurally proven rewrite path",
)));
}
}
*head = remap.get(head).cloned().ok_or_else(|| {
serde_json::Error::io(std::io::Error::other(
"released 0.8.10 transcript head has no verified format-2 body",
))
})?;
for body in revisions.iter_mut() {
body.revision = remap.get(&body.revision).cloned().ok_or_else(|| {
serde_json::Error::io(std::io::Error::other(
"released 0.8.10 transcript body lost its format-2 identity",
))
})?;
if let Some(parent) = body.parent_revision.as_mut() {
*parent = remap.get(parent).cloned().ok_or_else(|| {
serde_json::Error::io(std::io::Error::other(
"released 0.8.10 transcript body parent has no verified format-2 identity",
))
})?;
}
}
let mut retained_revision_ids = BTreeSet::new();
revisions.retain(|body| retained_revision_ids.insert(body.revision.clone()));
*commits = remapped_commits;
Ok(())
}
fn released_ids_connected_by_collapsed_rewrites(
adjacency: &BTreeMap<String, BTreeSet<String>>,
start: &str,
target: &str,
) -> bool {
if start == target {
return true;
}
let mut visited = BTreeSet::new();
let mut pending = vec![start.to_string()];
while let Some(current) = pending.pop() {
if !visited.insert(current.clone()) {
continue;
}
let Some(neighbors) = adjacency.get(¤t) else {
continue;
};
if neighbors.contains(target) {
return true;
}
pending.extend(neighbors.iter().cloned());
}
false
}
impl TranscriptHistoryState {
pub(in crate::session) fn from_store_replayed_compact_graph(
anchor_revision: String,
anchor_messages: Vec<Message>,
anchor_row_prefix: SessionMessageRowPrefixAccumulator,
edges: Vec<TranscriptRevisionEdge>,
expected_rewrite_prefix: &TranscriptRewritePrefixAccumulator,
expected_graph_prefix: &TranscriptGraphPrefixAccumulator,
) -> Result<Self, TranscriptEditError> {
let first = edges.first().ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"store-replayed compact graph carries no rewrite edge".to_string(),
)
})?;
if first.base_revision() != anchor_revision
|| first.messages_before_base() != anchor_messages.len()
|| anchor_row_prefix.row_count() != anchor_messages.len() as u64
{
return Err(TranscriptEditError::HistoryStateMalformed(
"store-replayed compact graph anchor does not match its first edge".to_string(),
));
}
let actual_anchor_revision = transcript_messages_digest(&anchor_messages)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
if actual_anchor_revision != anchor_revision {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"store-replayed compact graph anchor revision {anchor_revision} has digest {actual_anchor_revision}"
)));
}
let anchor = TranscriptRevisionAnchor {
revision: anchor_revision,
messages: anchor_messages,
row_prefix: anchor_row_prefix,
created_at: first.parent_created_at(),
};
let rewrite_prefix = edges
.last()
.map(TranscriptRevisionEdge::rewrite_prefix)
.cloned()
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"store-replayed compact graph lost its final rewrite prefix".to_string(),
)
})?;
if &rewrite_prefix != expected_rewrite_prefix {
return Err(TranscriptEditError::HistoryStateMalformed(
"store-replayed compact graph rewrite prefix differs from the physical head"
.to_string(),
));
}
let graph_prefix = TranscriptGraphPrefixAccumulator::from_graph(&anchor, &edges)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
if &graph_prefix != expected_graph_prefix {
return Err(TranscriptEditError::HistoryStateMalformed(
"store-replayed compact graph prefix differs from the physical head".to_string(),
));
}
let persistent_edges = PersistentTranscriptEdges::from_vec(&anchor, edges)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let state = Self {
format: TRANSCRIPT_HISTORY_FORMAT_CURRENT,
anchor: Arc::new(anchor),
edges: persistent_edges,
rewrite_prefix,
graph_prefix,
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
};
validate_transcript_history_state(&state)?;
Ok(state)
}
fn from_legacy_full_bodies(
legacy_head: String,
commits: Vec<TranscriptRewriteCommit>,
revisions: Vec<TranscriptRevisionBody>,
) -> Result<Self, TranscriptEditError> {
let first = commits.first().ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"released 0.8.10 transcript graph carries no rewrite commit".to_string(),
)
})?;
for body in &revisions {
let digest = transcript_messages_digest(&body.messages)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
if digest != body.revision {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"released 0.8.10 transcript body {} has digest {digest}",
body.revision
)));
}
}
let bodies_by_revision = revisions
.iter()
.map(|body| (body.revision.as_str(), body))
.collect::<std::collections::HashMap<_, _>>();
if bodies_by_revision.len() != revisions.len() {
return Err(TranscriptEditError::HistoryStateMalformed(
"released 0.8.10 graph repeats a revision body id its writer deduplicated"
.to_string(),
));
}
let anchor_body = bodies_by_revision
.get(first.parent_revision.as_str())
.copied()
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(format!(
"released 0.8.10 graph omits first parent body {}",
first.parent_revision
))
})?;
let anchor = TranscriptRevisionAnchor {
revision: anchor_body.revision.clone(),
messages: anchor_body.messages.clone(),
row_prefix: SessionMessageRowPrefixAccumulator::from_messages(&anchor_body.messages)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?,
created_at: anchor_body.created_at,
};
let mut edges = Vec::with_capacity(commits.len());
let mut rewrite_prefix = TranscriptRewritePrefixAccumulator::empty();
let mut previous = anchor_body;
let mut previous_witness = TranscriptEndpointWitness::from_messages(&anchor_body.messages)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
for commit in commits {
let parent = bodies_by_revision
.get(commit.parent_revision.as_str())
.copied()
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(format!(
"released 0.8.10 graph omits parent body {}",
commit.parent_revision
))
})?;
let revision = bodies_by_revision
.get(commit.revision.as_str())
.copied()
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(format!(
"released 0.8.10 graph omits revision body {}",
commit.revision
))
})?;
validate_transcript_rewrite_record(&commit, parent, revision)?;
rewrite_prefix = rewrite_prefix
.extend(&commit)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let edge = edge_from_materialized_bodies(
previous,
&previous_witness,
parent,
revision,
commit,
rewrite_prefix.clone(),
MaterializedParentAdvanceSource::Released0810Import,
)?;
previous_witness = edge.result_witness().clone();
edges.push(edge);
previous = revision;
}
if legacy_head != previous.revision {
let live_head = bodies_by_revision
.get(legacy_head.as_str())
.copied()
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(format!(
"released 0.8.10 graph omits advertised head body {legacy_head}"
))
})?;
parent_advance_from_materialized(
previous,
live_head,
u64::MAX,
MaterializedParentAdvanceSource::Released0810Import,
)?;
}
let graph_prefix = TranscriptGraphPrefixAccumulator::from_graph(&anchor, &edges)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let persistent_edges = PersistentTranscriptEdges::from_vec(&anchor, edges)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let state = Self {
format: TRANSCRIPT_HISTORY_FORMAT_CURRENT,
anchor: Arc::new(anchor),
edges: persistent_edges,
rewrite_prefix,
graph_prefix,
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
};
validate_transcript_history_state(&state)?;
Ok(state)
}
#[allow(clippy::too_many_arguments)]
pub(in crate::session) fn from_authorized_first_rewrite(
parent: TranscriptRevisionBody,
parent_row_prefix: SessionMessageRowPrefixAccumulator,
revision: &str,
revision_messages: &[Message],
revision_created_at: SystemTime,
result_row_prefix: SessionMessageRowPrefixAccumulator,
replacement: Vec<Message>,
commit: TranscriptRewriteCommit,
) -> Result<Self, TranscriptEditError> {
if parent_row_prefix.row_count() != parent.messages.len() as u64
|| result_row_prefix.row_count() != revision_messages.len() as u64
|| commit.parent_revision != parent.revision
|| commit.revision != revision
|| commit.messages_before != parent.messages.len()
|| commit.messages_after != revision_messages.len()
{
return Err(TranscriptEditError::HistoryStateMalformed(
"authorized first rewrite carries inconsistent endpoint authority".to_string(),
));
}
let anchor = TranscriptRevisionAnchor {
revision: parent.revision.clone(),
messages: parent.messages,
row_prefix: parent_row_prefix.clone(),
created_at: parent.created_at,
};
let rewrite_prefix = TranscriptRewritePrefixAccumulator::empty()
.extend(&commit)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let base_witness = TranscriptEndpointWitness::from_messages_with_row_prefix(
&anchor.messages,
parent_row_prefix.clone(),
)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let result_witness = TranscriptEndpointWitness::from_messages_with_row_prefix(
revision_messages,
result_row_prefix,
)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let (at, _) = commit.selection.bounds();
let edge = TranscriptRevisionEdge {
commit,
rewrite_prefix: rewrite_prefix.clone(),
base_revision: anchor.revision.clone(),
messages_before_base: anchor.messages.len(),
parent_advance: TranscriptParentAdvance::ExactAppend {
appended: Vec::new(),
},
parent_row_prefix,
rewrite: TranscriptRewritePatch { at, replacement },
result_witness,
parent_created_at: anchor.created_at,
revision_created_at,
};
super::validate::validate_transcript_revision_edge(
anchor.revision(),
&base_witness,
&edge,
)?;
let graph_prefix = TranscriptGraphPrefixAccumulator::from_anchor(&anchor)
.and_then(|prefix| prefix.extend(&edge))
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let persistent_edges = PersistentTranscriptEdges::from_vec(&anchor, vec![edge])
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
Ok(Self {
format: TRANSCRIPT_HISTORY_FORMAT_CURRENT,
anchor: Arc::new(anchor),
edges: persistent_edges,
rewrite_prefix,
graph_prefix,
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
})
}
#[allow(clippy::too_many_arguments)]
pub(in crate::session) fn append_authorized_rewrite(
&mut self,
commit: TranscriptRewriteCommit,
messages_before_base: usize,
parent_advance: TranscriptParentAdvance,
parent_row_prefix: SessionMessageRowPrefixAccumulator,
replacement: Vec<Message>,
result_witness: TranscriptEndpointWitness,
parent_created_at: SystemTime,
revision_created_at: SystemTime,
) -> Result<(), TranscriptEditError> {
let base_revision = self.head().to_string();
let (at, _) = commit.selection.bounds();
let rewrite_prefix = self
.rewrite_prefix()
.extend(&commit)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let edge = TranscriptRevisionEdge {
commit,
rewrite_prefix,
base_revision,
messages_before_base,
parent_advance,
parent_row_prefix,
rewrite: TranscriptRewritePatch { at, replacement },
result_witness,
parent_created_at,
revision_created_at,
};
let base_witness = self.final_endpoint_witness().ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"current transcript graph has no audited endpoint".to_string(),
)
})?;
super::validate::validate_transcript_revision_edge(
self.edges
.last()
.map_or(self.anchor.revision(), |edge| edge.revision()),
base_witness,
&edge,
)?;
self.graph_prefix = self
.graph_prefix
.extend(&edge)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
self.rewrite_prefix = edge.rewrite_prefix.clone();
self.edges.push(edge, self.graph_prefix.clone());
Ok(())
}
#[must_use]
pub fn format(&self) -> &str {
self.format
}
#[must_use]
pub const fn digest_format(&self) -> u32 {
self.digest_format
}
#[must_use]
pub fn head(&self) -> &str {
self.edges
.last()
.map(TranscriptRevisionEdge::revision)
.unwrap_or_else(|| self.anchor.revision())
}
pub fn commits(&self) -> impl ExactSizeIterator<Item = &TranscriptRewriteCommit> {
self.edges.ordered().iter().map(|edge| edge.commit())
}
#[must_use]
pub fn commit_count(&self) -> usize {
self.edges.len()
}
#[must_use]
pub fn commit(&self, index: usize) -> Option<&TranscriptRewriteCommit> {
self.edges.get(index).map(TranscriptRevisionEdge::commit)
}
pub(crate) fn edge(&self, index: usize) -> Option<&TranscriptRevisionEdge> {
self.edges.get(index)
}
#[must_use]
pub fn last_commit(&self) -> Option<&TranscriptRewriteCommit> {
self.edges.last().map(TranscriptRevisionEdge::commit)
}
#[must_use]
pub fn rewrite_prefix(&self) -> &TranscriptRewritePrefixAccumulator {
&self.rewrite_prefix
}
#[must_use]
pub fn anchor(&self) -> &TranscriptRevisionAnchor {
self.anchor.as_ref()
}
#[must_use]
pub fn edges(&self) -> &[Arc<TranscriptRevisionEdge>] {
self.edges.ordered()
}
pub(super) fn edge_suffix(&self, start: usize) -> Option<Vec<Arc<TranscriptRevisionEdge>>> {
self.edges.suffix(start)
}
#[must_use]
pub fn graph_prefix(&self) -> &TranscriptGraphPrefixAccumulator {
&self.graph_prefix
}
#[must_use]
pub(crate) fn extends_exact_graph(&self, prefix: &Self) -> bool {
let prefix_count = prefix.commit_count();
if prefix_count > self.commit_count() {
return false;
}
if prefix_count == 0 {
let expected_graph_prefix =
TranscriptGraphPrefixAccumulator::from_anchor(prefix.anchor()).ok();
return self.anchor() == prefix.anchor()
&& prefix.rewrite_prefix() == &TranscriptRewritePrefixAccumulator::empty()
&& expected_graph_prefix.as_ref() == Some(prefix.graph_prefix());
}
self.graph_prefix_at(prefix_count) == Some(prefix.graph_prefix())
}
#[must_use]
pub(crate) fn unique_revision_position(&self, revision: &str) -> Option<usize> {
fn observe(
candidate: &str,
candidate_position: usize,
revision: &str,
found: &mut Option<usize>,
ambiguous: &mut bool,
) {
if candidate == revision && found.replace(candidate_position).is_some() {
*ambiguous = true;
}
}
let mut position = 0usize;
let mut found = None;
let mut ambiguous = false;
let mut preceding_child = self.anchor.revision();
observe(
self.anchor.revision(),
position,
revision,
&mut found,
&mut ambiguous,
);
for edge in self.edges.ordered() {
if edge.parent_revision() != preceding_child {
position += 1;
observe(
edge.parent_revision(),
position,
revision,
&mut found,
&mut ambiguous,
);
}
position += 1;
observe(
edge.revision(),
position,
revision,
&mut found,
&mut ambiguous,
);
preceding_child = edge.revision();
}
if ambiguous { None } else { found }
}
#[must_use]
pub(crate) fn revision_extends(&self, descendant: &str, ancestor: &str) -> bool {
match (
self.unique_revision_position(descendant),
self.unique_revision_position(ancestor),
) {
(Some(descendant), Some(ancestor)) => descendant >= ancestor,
_ => false,
}
}
#[must_use]
pub(crate) fn contains_exact_commit(&self, commit: &TranscriptRewriteCommit) -> bool {
commit
.rewrite_generation
.checked_sub(1)
.and_then(|index| usize::try_from(index).ok())
.and_then(|index| self.commit(index))
== Some(commit)
}
pub(crate) fn graph_prefix_at(
&self,
occurrence_count: usize,
) -> Option<&TranscriptGraphPrefixAccumulator> {
self.edges.graph_prefix_at(occurrence_count)
}
pub fn final_endpoint_witness(&self) -> Option<&TranscriptEndpointWitness> {
self.edges
.last()
.map(TranscriptRevisionEdge::result_witness)
}
pub(in crate::session) fn derive_live_row_lineage_after_final_semantic_replay(
&self,
live: &[Message],
) -> Result<Option<SessionMessageRowPrefixAccumulator>, TranscriptEditError> {
let endpoint_witness = self.final_endpoint_witness().ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"compact transcript graph has no final endpoint witness".to_string(),
)
})?;
let endpoint = self.materialize_revision(self.head())?;
if live.len() < endpoint.messages.len() {
return Ok(None);
}
let advance = if live[..endpoint.messages.len()] == endpoint.messages {
TranscriptParentAdvance::ExactAppend {
appended: live[endpoint.messages.len()..].to_vec(),
}
} else {
return Ok(None);
};
row_prefix_after_parent_advance(endpoint_witness.row_prefix(), &advance).map(Some)
}
#[must_use]
pub fn parent_transition(&self, index: usize) -> Option<TranscriptRewriteParentTransition> {
self.edges
.get(index)
.map(|edge| edge.parent_advance.transition())
}
#[must_use]
pub fn parent_transitions(&self) -> Vec<TranscriptRewriteParentTransition> {
self.edges
.ordered()
.iter()
.map(|edge| edge.parent_advance.transition())
.collect()
}
#[must_use]
pub fn contains_revision(&self, revision: &str) -> bool {
self.anchor.revision == revision
|| self
.edges
.ordered()
.iter()
.any(|edge| edge.parent_revision() == revision || edge.revision() == revision)
}
#[must_use]
pub fn retained_revision_count(&self) -> usize {
1usize.saturating_add(self.edges.len().saturating_mul(2))
}
pub(crate) fn compact_mechanical_revision_bodies(&mut self) -> Result<(), TranscriptEditError> {
validate_transcript_history_state(self)?;
Ok(())
}
pub(crate) fn canonicalize_to_latest_audited_head(&mut self) {
}
pub(crate) fn prune_mechanical_revision_bodies(&mut self) {}
pub(super) fn proved_prefix(&self, edge_count: usize) -> Result<Self, TranscriptEditError> {
if edge_count == 0 || edge_count > self.edges.len() {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"compact graph prefix {edge_count} is outside 1..={}",
self.edges.len()
)));
}
let edges = self.edges.prefix(edge_count).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"compact graph prefix could not address its persistent tail".to_string(),
)
})?;
let rewrite_prefix = edges
.last()
.map(TranscriptRevisionEdge::rewrite_prefix)
.cloned()
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"compact graph prefix lost its final edge".to_string(),
)
})?;
let graph_prefix = TranscriptGraphPrefixAccumulator::from_graph(
self.anchor.as_ref(),
edges.ordered().iter().map(AsRef::as_ref),
)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
Ok(Self {
format: TRANSCRIPT_HISTORY_FORMAT_CURRENT,
anchor: Arc::clone(&self.anchor),
edges,
rewrite_prefix,
graph_prefix,
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
})
}
pub fn materialize_revision(
&self,
revision: &str,
) -> Result<TranscriptRevisionBody, TranscriptEditError> {
if let Some(index) = self
.edges
.ordered()
.iter()
.rposition(|edge| edge.revision() == revision)
{
return self.materialize_occurrence(index, false);
}
if let Some(index) = self
.edges
.ordered()
.iter()
.rposition(|edge| edge.parent_revision() == revision)
{
return self.materialize_occurrence(index, true);
}
if self.anchor.revision == revision {
TRANSCRIPT_HISTORY_FULL_BODY_MATERIALIZATIONS
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
return Ok(TranscriptRevisionBody {
revision: self.anchor.revision.clone(),
parent_revision: None,
messages: self.anchor.messages.clone(),
created_at: self.anchor.created_at,
});
}
Err(TranscriptEditError::HistoryStateMalformed(format!(
"missing transcript revision {revision}"
)))
}
pub(crate) fn materialize_occurrence_parent(
&self,
edge_index: usize,
) -> Result<TranscriptRevisionBody, TranscriptEditError> {
self.materialize_occurrence(edge_index, true)
}
pub(crate) fn materialize_occurrence_child(
&self,
edge_index: usize,
) -> Result<TranscriptRevisionBody, TranscriptEditError> {
self.materialize_occurrence(edge_index, false)
}
fn materialize_occurrence(
&self,
edge_index: usize,
parent: bool,
) -> Result<TranscriptRevisionBody, TranscriptEditError> {
if edge_index >= self.edges.len() {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite occurrence index {edge_index} is outside the compact graph"
)));
}
TRANSCRIPT_HISTORY_FULL_BODY_MATERIALIZATIONS
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let mut messages = self.anchor.messages.clone();
for (index, edge) in self.edges.ordered().iter().enumerate().take(edge_index + 1) {
apply_parent_advance(&mut messages, &edge.parent_advance)?;
if index == edge_index && parent {
let digest = transcript_messages_digest(&messages).map_err(|error| {
TranscriptEditError::HistoryStateMalformed(error.to_string())
})?;
if digest != edge.parent_revision() {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite occurrence {} materialized parent has digest {digest}, expected {}",
edge.rewrite_generation(),
edge.parent_revision()
)));
}
return Ok(TranscriptRevisionBody {
revision: edge.parent_revision().to_string(),
parent_revision: Some(edge.base_revision.clone()),
messages,
created_at: edge.parent_created_at,
});
}
let (_, end) = edge.commit.selection.bounds();
if end > messages.len() {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite occurrence {} patch exceeds materialized parent",
edge.rewrite_generation()
)));
}
let removed_digest = transcript_messages_digest(&messages[edge.rewrite.at..end])
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
if removed_digest != edge.commit.original_span_digest {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite occurrence {} materialized removed span has wrong digest",
edge.rewrite_generation()
)));
}
messages.splice(edge.rewrite.at..end, edge.rewrite.replacement.clone());
if index == edge_index {
let digest = transcript_messages_digest(&messages).map_err(|error| {
TranscriptEditError::HistoryStateMalformed(error.to_string())
})?;
if digest != edge.revision() {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite occurrence {} materialized child has digest {digest}, expected {}",
edge.rewrite_generation(),
edge.revision()
)));
}
return Ok(TranscriptRevisionBody {
revision: edge.revision().to_string(),
parent_revision: Some(edge.parent_revision().to_string()),
messages,
created_at: edge.revision_created_at,
});
}
}
Err(TranscriptEditError::HistoryStateMalformed(
"rewrite occurrence materialization did not reach its target".to_string(),
))
}
pub fn materialize_revision_bodies(
&self,
) -> Result<Vec<TranscriptRevisionBody>, TranscriptEditError> {
let mut seen = BTreeSet::new();
let mut bodies = Vec::new();
for revision in std::iter::once(self.anchor.revision.as_str()).chain(
self.edges
.ordered()
.iter()
.flat_map(|edge| [edge.parent_revision(), edge.revision()]),
) {
if seen.insert(revision.to_string()) {
bodies.push(self.materialize_revision(revision)?);
}
}
Ok(bodies)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MaterializedParentAdvanceSource {
CurrentAudit,
Released0810Import,
}
fn parent_advance_from_materialized(
base: &TranscriptRevisionBody,
parent: &TranscriptRevisionBody,
rewrite_generation: u64,
source: MaterializedParentAdvanceSource,
) -> Result<TranscriptParentAdvance, TranscriptEditError> {
if parent.messages.len() < base.messages.len() {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite occurrence {rewrite_generation} parent is shorter than its base"
)));
}
if parent.messages[..base.messages.len()] == base.messages {
return Ok(TranscriptParentAdvance::ExactAppend {
appended: parent.messages[base.messages.len()..].to_vec(),
});
}
if source == MaterializedParentAdvanceSource::CurrentAudit {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"current rewrite occurrence {rewrite_generation} parent is not an exact append"
)));
}
let retained_parent = &parent.messages[..base.messages.len()];
let at = base
.messages
.iter()
.zip(retained_parent)
.take_while(|(base, parent)| base == parent)
.count();
let common_suffix = base.messages[at..]
.iter()
.rev()
.zip(retained_parent[at..].iter().rev())
.take_while(|(base, parent)| base == parent)
.count();
let end = base.messages.len() - common_suffix;
if at >= end {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite occurrence {rewrite_generation} importer could not derive an exact parent splice"
)));
}
Ok(TranscriptParentAdvance::ExactSplice {
at,
replacement: retained_parent[at..end].to_vec(),
appended: parent.messages[base.messages.len()..].to_vec(),
})
}
fn row_prefix_after_parent_advance(
base: &SessionMessageRowPrefixAccumulator,
advance: &TranscriptParentAdvance,
) -> Result<SessionMessageRowPrefixAccumulator, TranscriptEditError> {
let advanced = match advance {
TranscriptParentAdvance::ExactAppend { .. } => base.clone(),
TranscriptParentAdvance::ExactSplice {
at, replacement, ..
} => {
let replacement_rows = replacement
.iter()
.map(serde_json::to_vec)
.collect::<Result<Vec<_>, _>>()
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let end = at.checked_add(replacement.len()).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"imported parent splice end overflowed".to_string(),
)
})?;
let at = u64::try_from(*at).map_err(|_| {
TranscriptEditError::HistoryStateMalformed(
"imported parent splice start exceeds durable row coordinates".to_string(),
)
})?;
let end = u64::try_from(end).map_err(|_| {
TranscriptEditError::HistoryStateMalformed(
"imported parent splice end exceeds durable row coordinates".to_string(),
)
})?;
base.replace_serialized_range(at, end, &replacement_rows)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?
}
};
let appended = advance
.appended()
.iter()
.map(serde_json::to_vec)
.collect::<Result<Vec<_>, _>>()
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
advanced
.extend_serialized_rows(&appended)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))
}
fn edge_from_materialized_bodies(
base: &TranscriptRevisionBody,
base_witness: &TranscriptEndpointWitness,
parent: &TranscriptRevisionBody,
revision: &TranscriptRevisionBody,
commit: TranscriptRewriteCommit,
rewrite_prefix: TranscriptRewritePrefixAccumulator,
source: MaterializedParentAdvanceSource,
) -> Result<TranscriptRevisionEdge, TranscriptEditError> {
let parent_advance =
parent_advance_from_materialized(base, parent, commit.rewrite_generation, source)?;
let parent_row_prefix =
row_prefix_after_parent_advance(base_witness.row_prefix(), &parent_advance)?;
let (at, end) = commit.selection.bounds();
let removed_len = end.checked_sub(at).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed("rewrite selection is inverted".to_string())
})?;
let retained = commit
.messages_before
.checked_sub(removed_len)
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite removes more messages than parent carries".to_string(),
)
})?;
let replacement_len = commit.messages_after.checked_sub(retained).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite message counts cannot describe replacement".to_string(),
)
})?;
let replacement_end = at.checked_add(replacement_len).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed("rewrite replacement end overflow".to_string())
})?;
let replacement = revision.messages[at..replacement_end].to_vec();
let replacement_rows = replacement
.iter()
.map(serde_json::to_vec)
.collect::<Result<Vec<_>, _>>()
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let result_row_prefix = parent_row_prefix
.replace_serialized_range(at as u64, end as u64, &replacement_rows)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?;
let edge = TranscriptRevisionEdge {
rewrite_prefix,
base_revision: base.revision.clone(),
messages_before_base: base.messages.len(),
parent_advance,
parent_row_prefix,
rewrite: TranscriptRewritePatch { at, replacement },
result_witness: TranscriptEndpointWitness::from_messages_with_row_prefix(
&revision.messages,
result_row_prefix,
)
.map_err(|error| TranscriptEditError::HistoryStateMalformed(error.to_string()))?,
parent_created_at: parent.created_at,
revision_created_at: revision.created_at,
commit,
};
super::validate::validate_transcript_revision_edge(&base.revision, base_witness, &edge)?;
Ok(edge)
}
fn apply_parent_advance(
messages: &mut Vec<Message>,
advance: &TranscriptParentAdvance,
) -> Result<(), TranscriptEditError> {
match advance {
TranscriptParentAdvance::ExactAppend { appended } => messages.extend_from_slice(appended),
TranscriptParentAdvance::ExactSplice {
at,
replacement,
appended,
} => {
let end = at.checked_add(replacement.len()).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"imported parent splice end overflowed".to_string(),
)
})?;
if replacement.is_empty() || end > messages.len() {
return Err(TranscriptEditError::HistoryStateMalformed(
"imported parent splice is empty or outside its base".to_string(),
));
}
messages.splice(*at..end, replacement.iter().cloned());
messages.extend_from_slice(appended);
}
}
Ok(())
}
pub(super) fn record_is_proved_by(
_proved: Option<&ValidatedTranscriptHistory>,
_record: &TranscriptRewriteRecord,
) -> bool {
false
}
impl TranscriptHistoryState {
pub fn from_rewrite_records<I>(records: I) -> Result<Option<Self>, TranscriptEditError>
where
I: IntoIterator<Item = TranscriptRewriteRecord>,
{
Self::from_rewrite_records_with_proved(records, None)
}
pub fn from_rewrite_records_with_proved<I>(
records: I,
proved: Option<&ValidatedTranscriptHistory>,
) -> Result<Option<Self>, TranscriptEditError>
where
I: IntoIterator<Item = TranscriptRewriteRecord>,
{
let mut records = records.into_iter().collect::<Vec<_>>();
if records.is_empty() {
return Ok(None);
}
if records
.iter()
.any(|record| record.commit.rewrite_generation == 0)
{
return Err(TranscriptEditError::HistoryStateMalformed(
"generation-less 0.8.10 rewrite records require checkpoint-bound EventStore reconciliation"
.to_string(),
));
}
records.sort_by_key(|record| record.commit.rewrite_generation);
let proved_count = proved.map_or(0, |history| history.state().commit_count());
let first_generation = records[0].commit.rewrite_generation;
if first_generation > 1 && first_generation != proved_count as u64 + 1 {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite record set starts at occurrence {first_generation} without the exact proved prefix"
)));
}
let mut new_records = Vec::new();
for record in records {
let generation = record.commit.rewrite_generation;
if generation <= proved_count as u64 {
validate_transcript_rewrite_record(
&record.commit,
&record.parent_body,
&record.revision_body,
)?;
let matches_prefix = proved.and_then(|history| {
usize::try_from(generation - 1)
.ok()
.and_then(|index| history.state().commit(index))
}) == Some(&record.commit);
if !matches_prefix {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite record occurrence {generation} conflicts with proved graph prefix"
)));
}
continue;
}
let expected = proved_count as u64 + new_records.len() as u64 + 1;
if generation != expected {
return Err(TranscriptEditError::HistoryStateMalformed(format!(
"rewrite record occurrence {generation} is not expected contiguous generation {expected}"
)));
}
new_records.push(record);
}
if new_records.is_empty() {
return Ok(proved.map(|history| history.state().clone()));
}
let mut state = proved.map(|history| history.state().clone());
let mut previous = if let Some(history) = proved {
Some(
history
.state()
.materialize_revision(history.state().head())?,
)
} else {
None
};
let mut previous_witness =
proved.and_then(|history| history.state().final_endpoint_witness().cloned());
for record in new_records {
validate_transcript_rewrite_record(
&record.commit,
&record.parent_body,
&record.revision_body,
)?;
if state.is_none() {
let (start, end) = record.commit.selection.bounds();
let removed = end.checked_sub(start).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite record selection is inverted".to_string(),
)
})?;
let retained = record
.commit
.messages_before
.checked_sub(removed)
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite record selection exceeds its parent".to_string(),
)
})?;
let replacement_len = record
.commit
.messages_after
.checked_sub(retained)
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite record successor is shorter than retained spans".to_string(),
)
})?;
let replacement_end = start.checked_add(replacement_len).ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite record replacement range overflow".to_string(),
)
})?;
let replacement = record
.revision_body
.messages
.get(start..replacement_end)
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite record replacement range exceeds successor".to_string(),
)
})?
.to_vec();
let parent_prefix =
SessionMessageRowPrefixAccumulator::from_messages(&record.parent_body.messages)
.map_err(|error| {
TranscriptEditError::HistoryStateMalformed(error.to_string())
})?;
let replacement_rows = replacement
.iter()
.map(serde_json::to_vec)
.collect::<Result<Vec<_>, _>>()
.map_err(|error| {
TranscriptEditError::HistoryStateMalformed(error.to_string())
})?;
let result_prefix = parent_prefix
.replace_serialized_range(start as u64, end as u64, &replacement_rows)
.map_err(|error| {
TranscriptEditError::HistoryStateMalformed(error.to_string())
})?;
state = Some(Self::from_authorized_first_rewrite(
record.parent_body.clone(),
parent_prefix,
&record.revision_body.revision,
&record.revision_body.messages,
record.revision_body.created_at,
result_prefix,
replacement,
record.commit.clone(),
)?);
} else {
let base = previous.as_ref().ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite tail lost its preceding endpoint".to_string(),
)
})?;
let next_rewrite_prefix = state
.as_ref()
.ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite state initialization failed".to_string(),
)
})?
.rewrite_prefix
.extend(&record.commit)
.map_err(|error| {
TranscriptEditError::HistoryStateMalformed(error.to_string())
})?;
let edge = edge_from_materialized_bodies(
base,
previous_witness.as_ref().ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite tail lost its preceding row-lineage witness".to_string(),
)
})?,
&record.parent_body,
&record.revision_body,
record.commit.clone(),
next_rewrite_prefix.clone(),
MaterializedParentAdvanceSource::CurrentAudit,
)?;
let state = state.as_mut().ok_or_else(|| {
TranscriptEditError::HistoryStateMalformed(
"rewrite state initialization failed".to_string(),
)
})?;
state.rewrite_prefix = next_rewrite_prefix;
state.graph_prefix = state.graph_prefix.extend(&edge).map_err(|error| {
TranscriptEditError::HistoryStateMalformed(error.to_string())
})?;
state.edges.push(edge, state.graph_prefix.clone());
}
previous_witness = state
.as_ref()
.and_then(TranscriptHistoryState::final_endpoint_witness)
.cloned();
previous = Some(record.revision_body);
}
Ok(state)
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::redundant_clone)]
mod tests {
use super::*;
#[test]
fn rewrite_prefix_digest_binds_order_and_exact_commit_facts() {
let records = rewrite_chain(2);
let commits = records
.iter()
.map(|record| record.commit.clone())
.collect::<Vec<_>>();
let baseline =
transcript_rewrite_prefix_digest(&commits).expect("canonical prefix serializes");
let first = TranscriptRewritePrefixAccumulator::from_commits(&commits[..1])
.expect("first prefix serializes");
let extended = extend_transcript_rewrite_prefix_accumulator(first, &commits[1..])
.expect("tail extension serializes");
assert_eq!(extended.digest(), baseline);
assert_eq!(
extended.occurrence_count(),
u64::try_from(commits.len()).expect("test length fits u64")
);
let mut actor_changed = commits.clone();
actor_changed[0].actor = Some("different-actor".to_string());
assert_ne!(
transcript_rewrite_prefix_digest(&actor_changed).expect("changed prefix serializes"),
baseline,
"the receipt must bind full commit facts, not only revisions"
);
let mut reordered = commits.clone();
reordered.reverse();
for (index, commit) in reordered.iter_mut().enumerate() {
commit.rewrite_generation = u64::try_from(index)
.expect("test index fits u64")
.saturating_add(1);
}
assert_ne!(
transcript_rewrite_prefix_digest(&reordered).expect("reordered prefix serializes"),
baseline,
"the receipt must bind lineage order"
);
}
#[test]
fn rewrite_generation_distinguishes_byte_equal_occurrences_and_legacy_order_normalizes() {
let mut commits = rewrite_chain(1)
.into_iter()
.map(|record| record.commit)
.collect::<Vec<_>>();
let mut recurrence = commits[0].clone();
recurrence.rewrite_generation = 2;
commits.push(recurrence);
let one =
transcript_rewrite_prefix_digest(&commits[..1]).expect("one occurrence serializes");
let two = transcript_rewrite_prefix_digest(&commits).expect("two occurrences serialize");
assert_ne!(
one, two,
"byte-equal rewrite facts at distinct generations are distinct occurrences"
);
for commit in &mut commits {
commit.rewrite_generation = 0;
}
assert!(
normalize_legacy_graph_rewrite_generations(&mut commits)
.expect("proved 0.8.10 vector order normalizes")
);
assert_eq!(
commits
.iter()
.map(|commit| commit.rewrite_generation)
.collect::<Vec<_>>(),
vec![1, 2]
);
}
#[test]
fn generic_record_rebuild_cannot_authorize_generation_zero_from_input_order() {
let mut records = rewrite_chain(2);
for record in &mut records {
record.commit.rewrite_generation = 0;
}
let error = TranscriptHistoryState::from_rewrite_records(records)
.expect_err("audit row order is not legacy occurrence authority");
assert!(
matches!(error, TranscriptEditError::HistoryStateMalformed(ref message)
if message.contains("checkpoint-bound EventStore reconciliation")),
"unexpected error: {error}"
);
}
#[test]
fn rewrite_prefix_wire_and_count_mismatch_refuse_fail_closed() {
let malformed = serde_json::json!({
"occurrence_count": 1,
"digest": "sha256:ABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCDEFABCD"
});
assert!(
serde_json::from_value::<TranscriptRewritePrefixAccumulator>(malformed).is_err(),
"uppercase or otherwise non-canonical digest text must not enter authority"
);
let records = rewrite_chain(2);
let mut state = rebuild(&records);
state.rewrite_prefix.occurrence_count = 1;
assert!(
validate_transcript_history_state(&state).is_err(),
"the carried count/digest pair must match the exact graph occurrences"
);
}
use crate::digest_observability::session_content_digest_bytes;
use crate::session::{TranscriptRewriteReason, TranscriptRewriteSelection};
use crate::types::UserMessage;
fn message(text: &str) -> Message {
Message::User(UserMessage::text(text.to_string()))
}
fn body(messages: Vec<Message>, parent: Option<&str>) -> TranscriptRevisionBody {
let revision = transcript_messages_digest(&messages).expect("digest revision body");
TranscriptRevisionBody {
revision,
parent_revision: parent.map(str::to_string),
messages,
created_at: SystemTime::UNIX_EPOCH,
}
}
#[test]
fn released_importer_preserves_arbitrary_exact_parent_splice() {
let retained_prefix = message("retained prefix");
let retained_turn = message("retained turn");
let released_replacement = message("released replacement");
let released_appended_turn = message("released appended turn");
let base = body(
vec![
retained_prefix.clone(),
message("released base"),
retained_turn.clone(),
],
None,
);
let parent = body(
vec![
retained_prefix,
released_replacement.clone(),
retained_turn,
released_appended_turn.clone(),
],
Some(&base.revision),
);
let advance = parent_advance_from_materialized(
&base,
&parent,
2,
MaterializedParentAdvanceSource::Released0810Import,
)
.expect("released historical relationship remains materializable");
assert!(matches!(
&advance,
TranscriptParentAdvance::ExactSplice {
at: 1,
replacement,
appended,
} if replacement == &[released_replacement]
&& appended == &[released_appended_turn]
));
let mut materialized = base.messages.clone();
apply_parent_advance(&mut materialized, &advance)
.expect("historical parent advance applies");
assert_eq!(materialized, parent.messages);
}
#[test]
fn current_record_rebuild_refuses_non_append_parent_divergence() {
let first_parent = body(vec![message("retained prefix"), message("question")], None);
let first_revision = body(
vec![message("retained prefix"), message("first edit")],
Some(&first_parent.revision),
);
let first_commit = TranscriptRewriteCommit {
rewrite_generation: 1,
parent_revision: first_parent.revision.clone(),
revision: first_revision.revision.clone(),
selection: TranscriptRewriteSelection::MessageRange { start: 1, end: 2 },
original_span_digest: transcript_messages_digest(&first_parent.messages[1..2])
.expect("digest first original span"),
replacement_digest: transcript_messages_digest(&first_revision.messages[1..2])
.expect("digest first replacement span"),
messages_before: first_parent.messages.len(),
messages_after: first_revision.messages.len(),
reason: TranscriptRewriteReason::new("unit-test"),
actor: None,
committed_at: SystemTime::UNIX_EPOCH,
};
let first_record =
TranscriptRewriteRecord::new(first_commit, first_parent, first_revision.clone())
.expect("first rewrite record is valid");
let second_parent = body(
vec![message("divergent prefix"), message("first edit")],
None,
);
let divergent_prefix = second_parent.messages[0].clone();
let second_revision = body(
vec![divergent_prefix, message("second edit")],
Some(&second_parent.revision),
);
let second_commit = TranscriptRewriteCommit {
rewrite_generation: 2,
parent_revision: second_parent.revision.clone(),
revision: second_revision.revision.clone(),
selection: TranscriptRewriteSelection::MessageRange { start: 1, end: 2 },
original_span_digest: transcript_messages_digest(&second_parent.messages[1..2])
.expect("digest second original span"),
replacement_digest: transcript_messages_digest(&second_revision.messages[1..2])
.expect("digest second replacement span"),
messages_before: second_parent.messages.len(),
messages_after: second_revision.messages.len(),
reason: TranscriptRewriteReason::new("unit-test"),
actor: None,
committed_at: SystemTime::UNIX_EPOCH,
};
let second_record =
TranscriptRewriteRecord::new(second_commit, second_parent, second_revision)
.expect("second rewrite record is internally valid");
let error = TranscriptHistoryState::from_rewrite_records([first_record, second_record])
.expect_err("current audit reconstruction must reject a non-append parent bridge");
assert!(
matches!(error, TranscriptEditError::HistoryStateMalformed(ref message)
if message.contains("current rewrite occurrence 2 parent is not an exact append")),
"unexpected error: {error}"
);
}
#[test]
fn current_record_cannot_request_legacy_compaction_semantic_healing() {
let parent_body = body(vec![message("old turn"), message("newer turn")], None);
let revision_body = body(
vec![Message::User(UserMessage::compaction_summary("summary"))],
Some(&parent_body.revision),
);
let commit = TranscriptRewriteCommit {
rewrite_generation: 1,
parent_revision: parent_body.revision.clone(),
revision: revision_body.revision.clone(),
selection: TranscriptRewriteSelection::MessageRange { start: 0, end: 2 },
original_span_digest: transcript_messages_digest(&parent_body.messages)
.expect("digest original transcript"),
replacement_digest: transcript_messages_digest(&revision_body.messages)
.expect("digest replacement transcript"),
messages_before: parent_body.messages.len(),
messages_after: revision_body.messages.len(),
reason: TranscriptRewriteReason::new("current-untyped-selection"),
actor: None,
committed_at: SystemTime::UNIX_EPOCH,
};
let current_wire = serde_json::to_value(TranscriptRewriteRecord {
commit,
parent_body,
revision_body,
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
})
.expect("serialize current record");
let current: TranscriptRewriteRecord =
serde_json::from_value(current_wire.clone()).expect("decode current record");
assert!(
current.commit.selection.is_legacy_untyped(),
"a current-format record must not enter importer-only semantic healing"
);
let mut explicit_released_wire = current_wire.clone();
explicit_released_wire["digest_format"] =
serde_json::json!(TRANSCRIPT_DIGEST_FORMAT_RELEASED_0810);
let error = serde_json::from_value::<TranscriptRewriteRecord>(explicit_released_wire)
.expect_err("explicit released records require the source-authorized importer");
assert!(
error
.to_string()
.contains("current transcript rewrite record digest format 2 is not supported"),
"unexpected error: {error}"
);
let mut released_wire = current_wire;
released_wire
.as_object_mut()
.expect("record is an object")
.remove("digest_format");
let released: TranscriptRewriteRecord =
serde_json::from_value(released_wire).expect("decode released record");
assert!(
!released.commit.selection.is_legacy_untyped(),
"an unstamped released record retains its one-time typed compaction migration"
);
}
#[test]
fn ordinary_current_graph_ingress_refuses_released_digest_format() {
let current = rebuild(&rewrite_chain(1));
let mut wire = serde_json::to_value(current).expect("current graph serializes");
wire["digest_format"] = serde_json::json!(TRANSCRIPT_DIGEST_FORMAT_RELEASED_0810);
let error = serde_json::from_value::<TranscriptHistoryState>(wire)
.expect_err("released format must cross only the explicit 0.8.10 importer");
assert!(
error
.to_string()
.contains("current transcript graph digest format 2 is not supported"),
"unexpected error: {error}"
);
}
fn rewrite_chain(count: usize) -> Vec<TranscriptRewriteRecord> {
let mut messages = (0..6)
.map(|index| message(&format!("turn {index}")))
.collect::<Vec<_>>();
let mut records = Vec::with_capacity(count);
for generation in 0..count {
let parent_body = body(messages.clone(), None);
messages[0] = message(&format!("system prompt generation {generation}"));
let revision_body = body(messages.clone(), Some(&parent_body.revision));
let commit = TranscriptRewriteCommit {
rewrite_generation: u64::try_from(generation).expect("test generation fits u64")
+ 1,
parent_revision: parent_body.revision.clone(),
revision: revision_body.revision.clone(),
selection: TranscriptRewriteSelection::MessageRange { start: 0, end: 1 },
original_span_digest: transcript_messages_digest(&parent_body.messages[..1])
.expect("digest original span"),
replacement_digest: transcript_messages_digest(&revision_body.messages[..1])
.expect("digest replacement span"),
messages_before: parent_body.messages.len(),
messages_after: revision_body.messages.len(),
reason: TranscriptRewriteReason::new("adversarial-rewrite-chain"),
actor: None,
committed_at: SystemTime::UNIX_EPOCH,
};
records.push(
TranscriptRewriteRecord::new(commit, parent_body, revision_body)
.expect("chain record is valid"),
);
}
records
}
fn recurrence_chain() -> Vec<TranscriptRewriteRecord> {
let a = body(vec![message("A")], None);
let b = body(vec![message("B")], Some(&a.revision));
let first = TranscriptRewriteRecord::new(
TranscriptRewriteCommit {
rewrite_generation: 1,
parent_revision: a.revision.clone(),
revision: b.revision.clone(),
selection: TranscriptRewriteSelection::MessageRange { start: 0, end: 1 },
original_span_digest: transcript_messages_digest(&a.messages).expect("digest A"),
replacement_digest: transcript_messages_digest(&b.messages).expect("digest B"),
messages_before: 1,
messages_after: 1,
reason: TranscriptRewriteReason::new("recurrence-test"),
actor: None,
committed_at: SystemTime::UNIX_EPOCH,
},
a.clone(),
b.clone(),
)
.expect("A to B record");
let returned_a = body(a.messages.clone(), Some(&b.revision));
let second = TranscriptRewriteRecord::new(
TranscriptRewriteCommit {
rewrite_generation: 2,
parent_revision: b.revision.clone(),
revision: returned_a.revision.clone(),
selection: TranscriptRewriteSelection::MessageRange { start: 0, end: 1 },
original_span_digest: transcript_messages_digest(&b.messages).expect("digest B"),
replacement_digest: transcript_messages_digest(&returned_a.messages)
.expect("digest returned A"),
messages_before: 1,
messages_after: 1,
reason: TranscriptRewriteReason::new("recurrence-test"),
actor: None,
committed_at: SystemTime::UNIX_EPOCH,
},
b,
returned_a,
)
.expect("B to A record");
vec![first, second]
}
fn rebuild(records: &[TranscriptRewriteRecord]) -> TranscriptHistoryState {
TranscriptHistoryState::from_rewrite_records(records.to_vec())
.expect("rebuild from records")
.expect("chain is non-empty")
}
fn sealed(records: &[TranscriptRewriteRecord]) -> ValidatedTranscriptHistory {
ValidatedTranscriptHistory::seal_owned(rebuild(records)).expect("rebuilt chain seals")
}
fn hashed_bytes<T>(operation: impl FnOnce() -> T) -> (T, u64) {
let before = session_content_digest_bytes();
let value = operation();
(value, session_content_digest_bytes() - before)
}
fn assert_same_graph(left: &TranscriptHistoryState, right: &TranscriptHistoryState) {
assert_eq!(
serde_json::to_value(left).expect("left graph serializes"),
serde_json::to_value(right).expect("right graph serializes"),
"compact graph wires differ"
);
}
fn zero_edge_prefix(anchor_body: &TranscriptRevisionBody) -> TranscriptHistoryState {
let anchor = TranscriptRevisionAnchor {
revision: anchor_body.revision.clone(),
messages: anchor_body.messages.clone(),
row_prefix: SessionMessageRowPrefixAccumulator::from_messages(&anchor_body.messages)
.expect("anchor row prefix"),
created_at: anchor_body.created_at,
};
TranscriptHistoryState {
format: TRANSCRIPT_HISTORY_FORMAT_CURRENT,
graph_prefix: TranscriptGraphPrefixAccumulator::from_anchor(&anchor)
.expect("anchor graph prefix"),
anchor: Arc::new(anchor),
edges: PersistentTranscriptEdges::empty(),
rewrite_prefix: TranscriptRewritePrefixAccumulator::empty(),
digest_format: TRANSCRIPT_DIGEST_FORMAT_CURRENT,
}
}
#[test]
fn exact_graph_extension_accepts_only_the_same_zero_edge_anchor_authority() {
let records = rewrite_chain(2);
let full = rebuild(&records);
let prefix = zero_edge_prefix(&records[0].parent_body);
assert!(
full.extends_exact_graph(&prefix),
"the exact anchor is the zero-occurrence prefix of its rewrite graph"
);
let foreign_anchor = body(vec![message("foreign anchor")], None);
assert!(
!full.extends_exact_graph(&zero_edge_prefix(&foreign_anchor)),
"a different anchor cannot authorize a zero-occurrence prefix"
);
let mut malformed = prefix;
malformed.rewrite_prefix =
TranscriptRewritePrefixAccumulator::from_commits(&[records[0].commit.clone()])
.expect("one-commit prefix");
assert!(
!full.extends_exact_graph(&malformed),
"zero edges cannot carry a non-empty rewrite prefix"
);
}
#[test]
fn digest_only_ancestry_refuses_nonconsecutive_revision_recurrence() {
let records = recurrence_chain();
let state = rebuild(&records);
let a = records[0].commit.parent_revision.as_str();
let b = records[0].commit.revision.as_str();
assert_eq!(state.unique_revision_position(a), None);
assert!(state.unique_revision_position(b).is_some());
assert!(
!state.revision_extends(a, b),
"the later A cannot make digest-only A authorize the exact B occurrence"
);
assert!(
!state.revision_extends(b, a),
"the earlier A cannot make digest-only A authorize B in the other direction"
);
assert!(
!state.revision_extends(a, a),
"digest equality cannot bypass ambiguous A occurrence identity"
);
assert!(state.revision_extends(b, b));
}
#[test]
fn full_body_reconciliation_revalidates_bytes_even_when_commits_are_proved() {
let records = rewrite_chain(6);
let proved = sealed(&records);
let (replayed, hashed) = hashed_bytes(|| {
TranscriptHistoryState::from_rewrite_records_with_proved(records.clone(), Some(&proved))
});
let replayed = replayed.expect("replay succeeds").expect("non-empty");
assert!(
hashed > 0,
"a compact edge cannot authorize arbitrary bytes in a legacy full-body record"
);
assert_same_graph(&replayed, &proved);
}
#[test]
fn full_body_reconciliation_cost_tracks_all_legacy_evidence() {
let hash_one_new_record = |chain_len: usize| {
let records = rewrite_chain(chain_len);
let proved = sealed(&records[..chain_len - 1]);
let (replayed, hashed) = hashed_bytes(|| {
TranscriptHistoryState::from_rewrite_records_with_proved(
records.clone(),
Some(&proved),
)
});
let replayed = replayed.expect("replay succeeds").expect("non-empty");
assert_same_graph(&replayed, &rebuild(&records));
assert!(
hashed > 0,
"the trailing record is not carried by the proved graph and must \
be proved in full"
);
hashed
};
assert!(
hash_one_new_record(8) > hash_one_new_record(2),
"the one-time 0.8.10 reconciliation must validate every legacy full-body row"
);
}
#[test]
fn a_corrupted_body_is_rejected_when_its_commit_is_proved() {
let records = rewrite_chain(3);
let proved = sealed(&records);
let mut corrupted = records.clone();
corrupted[1].revision_body.messages[3] = message("corrupted tail");
let error =
TranscriptHistoryState::from_rewrite_records_with_proved(corrupted, Some(&proved))
.expect_err("a body that does not digest to its commit must be refused");
assert!(
matches!(error, TranscriptEditError::HistoryStateMalformed(_)),
"unexpected error: {error}"
);
}
#[test]
fn a_corrupted_new_record_is_rejected() {
let records = rewrite_chain(3);
let mut corrupted = records.clone();
corrupted[2].parent_body.messages[3] = message("corrupted tail");
let error = TranscriptHistoryState::from_rewrite_records(corrupted)
.expect_err("a body that does not digest to its commit must be refused");
assert!(
matches!(error, TranscriptEditError::HistoryStateMalformed(_)),
"unexpected error: {error}"
);
}
#[test]
fn a_proved_prefix_cannot_launder_a_corrupted_tail_record() {
let records = rewrite_chain(2);
let proved = sealed(&records[..1]);
let mut corrupted = records.clone();
corrupted[1].revision_body.messages[3] = message("corrupted tail");
let error =
TranscriptHistoryState::from_rewrite_records_with_proved(corrupted, Some(&proved))
.expect_err("a record whose endpoint the proved graph dropped is not proved");
assert!(
matches!(error, TranscriptEditError::HistoryStateMalformed(_)),
"unexpected error: {error}"
);
}
#[test]
fn a_body_mislabelled_against_its_commit_is_rejected_under_a_proved_graph() {
let records = rewrite_chain(3);
let proved = sealed(&records);
let mut mislabelled = records.clone();
mislabelled[1].parent_body.revision = "sha256:not-the-parent".to_string();
let error =
TranscriptHistoryState::from_rewrite_records_with_proved(mislabelled, Some(&proved))
.expect_err("a body labelled with a revision it does not carry must be refused");
assert!(
matches!(error, TranscriptEditError::HistoryStateMalformed(_)),
"unexpected error: {error}"
);
}
#[test]
fn a_proved_replay_builds_the_same_graph_as_an_unproved_one() {
let records = rewrite_chain(5);
let proved = sealed(&records);
let with_proof = TranscriptHistoryState::from_rewrite_records_with_proved(
records.clone(),
Some(&proved),
)
.expect("proved replay succeeds")
.expect("non-empty");
assert_same_graph(&with_proof, &rebuild(&records));
validate_transcript_history_state(&with_proof)
.expect("the proved replay's output is itself a valid graph");
}
#[test]
fn a_graph_does_not_serialize_the_retired_replay_cursor() {
let state = rebuild(&rewrite_chain(3));
let wire = serde_json::to_value(&state).expect("graph serializes");
assert!(
wire.get("replay_cursor").is_none(),
"physical store position is not session-document authority: {wire}"
);
let decoded: TranscriptHistoryState = serde_json::from_value(wire).expect("graph decodes");
assert_same_graph(&decoded, &state);
}
#[test]
fn a_replay_cursor_is_refused_by_the_current_graph_decoder() {
let state = rebuild(&rewrite_chain(2));
let mut wire = serde_json::to_value(&state).expect("graph serializes");
wire["replay_cursor"] = serde_json::json!({
"seq": 41,
"commits": 2,
"last_commit_revision": &state.commit(1).expect("second commit").revision,
});
assert!(
serde_json::from_value::<TranscriptHistoryState>(wire).is_err(),
"current graph ingress must reject the retired physical cursor"
);
}
#[test]
fn current_graph_refuses_missing_parent_advance_evidence() {
let state = rebuild(&rewrite_chain(3));
let mut wire = serde_json::to_value(&state).expect("graph serializes");
wire["edges"][0]
.as_object_mut()
.expect("edge wire is an object")
.remove("parent_advance");
assert!(
serde_json::from_value::<TranscriptHistoryState>(wire).is_err(),
"current ingress cannot synthesize missing parent-advance evidence"
);
}
#[test]
fn null_parent_advance_evidence_is_malformed() {
let state = rebuild(&rewrite_chain(2));
let mut wire = serde_json::to_value(&state).expect("graph serializes");
wire["edges"][0]["parent_advance"] = serde_json::Value::Null;
assert!(
serde_json::from_value::<TranscriptHistoryState>(wire).is_err(),
"null parent-advance evidence is malformed current state"
);
}
}