heddle_object_model/object/
thread_replication.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Portable Thread identity and immutable replication operations. Source and
3//! discussion causality have separate graphs so selective sharing is closed.
4pub mod capture_visibility;
5pub mod hosted_import;
6pub mod integration;
7pub mod local_integration;
8pub mod metadata;
9pub mod ownership_claim;
10pub mod ownership_resolution;
11pub mod source_author;
12use std::collections::BTreeSet;
13
14pub use capture_visibility::CaptureVisibility;
15use serde::{Deserialize, Serialize};
16pub use source_author::{AuthoredCapture, SOURCE_AUTHORIZATION_METHOD, SourceAuthor};
17
18use crate::{
19    error::{HeddleError, Result},
20    object::{CollaborationOperationEnvelope, ContentHash, State, StateId},
21};
22
23pub const GENESIS_FORMAT: &str = "heddle-thread-genesis-v1";
24pub const OPERATION_FORMAT: &str = "heddle-thread-operation-v1";
25pub const MAX_OPERATION_BYTES: usize = 256 * 1024;
26
27/// Ownership fixed by the original signed genesis. Hosting a key-owned Thread
28/// requires a separately verified, explicit ownership claim.
29#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
30#[serde(rename_all = "snake_case")]
31pub enum GenesisOwner {
32    LocalKey([u8; 32]),
33    Account(uuid::Uuid),
34}
35
36impl GenesisOwner {
37    fn is_valid(&self) -> bool {
38        match self {
39            Self::LocalKey(key) => *key != [0; 32],
40            Self::Account(account) => !account.is_nil(),
41        }
42    }
43}
44
45#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(deny_unknown_fields)]
47pub struct ThreadGenesis {
48    pub version: u16,
49    /// Canonical non-nil UUID shared by local and hosted replicas. Mutable
50    /// namespace/name addresses and native spool-link locators are not identity.
51    pub spool: String,
52    pub parent: Option<ContentHash>,
53    pub base: StateId,
54    pub name: String,
55    pub intent: String,
56    /// Immutable original ownership; uploading does not transfer it.
57    pub owner: GenesisOwner,
58    pub creator: [u8; 32],
59    pub nonce: Vec<u8>,
60}
61
62impl ThreadGenesis {
63    pub fn encode(&self) -> Result<Vec<u8>> {
64        if self.version != 1
65            || !self.owner.is_valid()
66            || matches!(&self.owner, GenesisOwner::LocalKey(key) if *key != self.creator)
67            || !uuid::Uuid::parse_str(&self.spool)
68                .is_ok_and(|id| !id.is_nil() && id.to_string() == self.spool)
69            || self.name.is_empty()
70            || self.nonce.len() > 64
71        {
72            return Err(invalid("invalid Thread genesis"));
73        }
74        let bytes = rmp_serde::to_vec_named(self)?;
75        bounded(&bytes)?;
76        Ok(bytes)
77    }
78
79    pub fn id(&self) -> Result<ContentHash> {
80        Ok(ContentHash::compute_typed(GENESIS_FORMAT, &self.encode()?))
81    }
82
83    pub fn decode(bytes: &[u8]) -> Result<Self> {
84        bounded(bytes)?;
85        let genesis: Self = rmp_serde::from_slice(bytes)?;
86        if genesis.encode()? != bytes {
87            return Err(invalid("non-canonical Thread genesis"));
88        }
89        Ok(genesis)
90    }
91}
92
93#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
94#[serde(rename_all = "snake_case")]
95pub enum ThreadFacet {
96    Source,
97    Discussion,
98    Metadata,
99}
100impl ThreadFacet {
101    pub const ALL: [Self; 3] = [Self::Source, Self::Discussion, Self::Metadata];
102}
103
104/// Durable causal admission. Receiving bytes alone does not accept an operation.
105#[derive(Clone, Debug, PartialEq, Eq)]
106pub enum Admission {
107    Accepted,
108    Pending,
109    Rejected(String),
110}
111
112#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
113#[serde(tag = "kind", content = "canonical", rename_all = "snake_case")]
114#[allow(clippy::large_enum_variant)] // wire codec; boxing would change MessagePack layout
115pub enum ThreadOperationBody {
116    Capture(AuthoredCapture),
117    Integration(Vec<u8>),
118    HostedImport(Vec<u8>),
119    LocalIntegration(Vec<u8>),
120    Discussion(Vec<u8>),
121    Context(Vec<u8>),
122    Metadata(Vec<u8>),
123}
124
125/// Source result shared by authored captures and executor-derived integrations.
126/// Per-Thread reference metadata never changes State identity.
127#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
128#[serde(deny_unknown_fields)]
129pub struct Capture {
130    pub state: Vec<u8>,
131    pub source_targets: Option<ContentHash>,
132    /// Original author-signed privacy declarations; independent of the courier.
133    pub visibility: Option<CaptureVisibility>,
134}
135impl From<Vec<u8>> for Capture {
136    fn from(state: Vec<u8>) -> Self {
137        Self {
138            state,
139            source_targets: None,
140            visibility: None,
141        }
142    }
143}
144
145#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
146#[serde(deny_unknown_fields)]
147pub struct ThreadOperation {
148    pub version: u16,
149    pub thread: ContentHash,
150    pub parents: BTreeSet<ContentHash>,
151    /// Signing key of the publisher of this immutable operation. Attribution
152    /// inside a source/discussion object is retained, not promoted to authority.
153    pub publisher: [u8; 32],
154    pub body: ThreadOperationBody,
155}
156
157impl ThreadOperation {
158    pub fn reference_proof(
159        &self,
160        genesis: &ThreadGenesis,
161    ) -> Result<Option<crate::object::source_target::capture::ReferenceProof>> {
162        let Some(capture) = self.source_result()? else {
163            return Ok(None);
164        };
165        let Some(descriptor) = capture.source_targets else {
166            return Ok(None);
167        };
168        if self.thread != genesis.id()? {
169            return Err(invalid("capture reference Thread mismatch"));
170        }
171        Ok(Some(
172            crate::object::source_target::capture::ReferenceProof {
173                descriptor,
174                scope: crate::object::CollaborationScope {
175                    spool: genesis.spool.parse().map_err(invalid)?,
176                    thread: Some(self.thread),
177                },
178                state: State::decode_current_msgpack(&capture.state)?.id(),
179            },
180        ))
181    }
182    pub fn facet(&self) -> ThreadFacet {
183        match self.body {
184            ThreadOperationBody::Capture(_)
185            | ThreadOperationBody::Integration(_)
186            | ThreadOperationBody::HostedImport(_)
187            | ThreadOperationBody::LocalIntegration(_) => ThreadFacet::Source,
188            ThreadOperationBody::Metadata(_) => ThreadFacet::Metadata,
189            ThreadOperationBody::Discussion(_) | ThreadOperationBody::Context(_) => {
190                ThreadFacet::Discussion
191            }
192        }
193    }
194
195    /// Uniform source result for capture and both integration authorities.
196    /// Original authored source identity. Hosted execution uses its executor proof.
197    pub fn source_author(&self) -> Result<Option<SourceAuthor>> {
198        if let ThreadOperationBody::Capture(capture) = &self.body {
199            return Ok(Some(capture.author.clone()));
200        }
201        Ok(self
202            .local_integration()?
203            .map(|integration| integration.author))
204    }
205
206    pub fn source_result(&self) -> Result<Option<Capture>> {
207        match &self.body {
208            ThreadOperationBody::Capture(capture) => Ok(Some(capture.result.clone())),
209            ThreadOperationBody::Integration(bytes) => {
210                Ok(Some(integration::HostedIntegration::decode(bytes)?.result))
211            }
212            ThreadOperationBody::HostedImport(bytes) => {
213                Ok(Some(hosted_import::HostedImport::decode(bytes)?.result))
214            }
215            ThreadOperationBody::LocalIntegration(bytes) => Ok(Some(
216                local_integration::LocalIntegration::decode(bytes)?.result,
217            )),
218            _ => Ok(None),
219        }
220    }
221
222    /// The exact source revision represented by a capture or hosted integration.
223    pub fn source_state(&self) -> Result<Option<State>> {
224        match &self.body {
225            ThreadOperationBody::Capture(bytes) => bytes.result.validated_state().map(Some),
226            ThreadOperationBody::Integration(bytes) => {
227                integration::HostedIntegration::decode(bytes)?
228                    .resulting_state()
229                    .map(Some)
230            }
231            ThreadOperationBody::LocalIntegration(bytes) => {
232                local_integration::LocalIntegration::decode(bytes)?
233                    .resulting_state()
234                    .map(Some)
235            }
236            ThreadOperationBody::HostedImport(bytes) => hosted_import::HostedImport::decode(bytes)?
237                .resulting_state()
238                .map(Some),
239            _ => Ok(None),
240        }
241    }
242    pub fn local_integration(&self) -> Result<Option<local_integration::LocalIntegration>> {
243        match &self.body {
244            ThreadOperationBody::LocalIntegration(bytes) => {
245                local_integration::LocalIntegration::decode(bytes).map(Some)
246            }
247            _ => Ok(None),
248        }
249    }
250    pub fn integration(&self) -> Result<Option<integration::HostedIntegration>> {
251        match &self.body {
252            ThreadOperationBody::Integration(bytes) => {
253                integration::HostedIntegration::decode(bytes).map(Some)
254            }
255            _ => Ok(None),
256        }
257    }
258
259    pub fn hosted_import(&self) -> Result<Option<hosted_import::HostedImport>> {
260        match &self.body {
261            ThreadOperationBody::HostedImport(bytes) => {
262                hosted_import::HostedImport::decode(bytes).map(Some)
263            }
264            _ => Ok(None),
265        }
266    }
267
268    /// A context root may be extracted atomically by a discussion resolution.
269    /// Later context revisions retain that original signed resolution as parent.
270    pub fn context_revision(&self) -> Result<Option<crate::object::ContextRevision>> {
271        use crate::object::{CollaborationOperationBodyV1 as Body, CollaborationResolution};
272        match &self.body {
273            ThreadOperationBody::Context(bytes) => crate::object::ContextRevision::decode(bytes)
274                .map(Some)
275                .map_err(invalid),
276            ThreadOperationBody::Discussion(bytes) => {
277                let record = CollaborationOperationEnvelope::decode(bytes)
278                    .map_err(invalid)?
279                    .operation;
280                match record.body {
281                    Body::Resolve {
282                        resolution: CollaborationResolution::IntoContext { context },
283                    } => Ok(Some(context)),
284                    _ => Ok(None),
285                }
286            }
287            _ => Ok(None),
288        }
289    }
290
291    pub fn encode(&self) -> Result<Vec<u8>> {
292        if self.version != 1 {
293            return Err(invalid("unsupported Thread operation version"));
294        }
295        match &self.body {
296            ThreadOperationBody::Capture(bytes) => {
297                bytes.author.validate()?;
298                bytes.result.validated_state()?;
299            }
300            ThreadOperationBody::Integration(bytes) => {
301                integration::HostedIntegration::decode(bytes)?.validate_operation(self)?;
302            }
303            ThreadOperationBody::HostedImport(bytes) => {
304                hosted_import::HostedImport::decode(bytes)?.validate_operation(self)?
305            }
306            ThreadOperationBody::LocalIntegration(bytes) => {
307                local_integration::LocalIntegration::decode(bytes)?.validate_operation(self)?;
308            }
309            ThreadOperationBody::Metadata(bytes) => {
310                metadata::ThreadControl::decode(bytes)?.validate_operation(self)?;
311            }
312            ThreadOperationBody::Context(bytes) => {
313                let context = crate::object::ContextRevision::decode(bytes).map_err(invalid)?;
314                if context.metadata.scope.thread != Some(self.thread)
315                    || context.parents.iter().copied().collect::<BTreeSet<_>>() != self.parents
316                {
317                    return Err(invalid(
318                        "context scope or causal parents differ from Thread operation",
319                    ));
320                }
321            }
322            ThreadOperationBody::Discussion(bytes) => {
323                let decoded = CollaborationOperationEnvelope::decode(bytes).map_err(invalid)?;
324                if decoded
325                    .operation
326                    .metadata
327                    .as_ref()
328                    .is_some_and(|m| m.scope.thread != Some(self.thread))
329                {
330                    return Err(invalid("collaboration metadata belongs to another Thread"));
331                }
332                if let Some(context) = self.context_revision()?
333                    && (Some(&context.metadata) != decoded.operation.metadata.as_ref()
334                        || context.extracted_from != Some(decoded.operation.discussion_id)
335                        || context.parents.iter().copied().collect::<BTreeSet<_>>() != self.parents)
336                {
337                    return Err(invalid(
338                        "extracted context differs from signed discussion actor, scope or parents",
339                    ));
340                }
341                if decoded.operation.encode().map_err(invalid)? != *bytes {
342                    return Err(invalid("non-canonical discussion operation"));
343                }
344            }
345        }
346        let bytes = rmp_serde::to_vec_named(self)?;
347        bounded(&bytes)?;
348        Ok(bytes)
349    }
350
351    pub fn id(&self) -> Result<ContentHash> {
352        Ok(ContentHash::compute_typed(
353            OPERATION_FORMAT,
354            &self.encode()?,
355        ))
356    }
357
358    pub fn decode(bytes: &[u8]) -> Result<Self> {
359        bounded(bytes)?;
360        let operation: Self = rmp_serde::from_slice(bytes)?;
361        if operation.encode()? != bytes {
362            return Err(invalid("non-canonical Thread operation"));
363        }
364        Ok(operation)
365    }
366
367    /// Called after every parent is present. A peer cannot relabel a private
368    /// dependency as public or attach an operation to a different Thread.
369    pub fn validate_parents(&self, genesis: &ThreadGenesis, parents: &[Self]) -> Result<()> {
370        if self.thread != genesis.id()? || parents.len() != self.parents.len() {
371            return Err(invalid("Thread or causal parent set mismatch"));
372        }
373        let ids = parents
374            .iter()
375            .map(Self::id)
376            .collect::<Result<BTreeSet<_>>>()?;
377        if ids != self.parents
378            || parents
379                .iter()
380                .any(|p| p.thread != self.thread || p.facet() != self.facet())
381        {
382            return Err(invalid("causal parents cross Thread or disclosure facet"));
383        }
384        if self
385            .source_result()?
386            .is_some_and(|result| result.source_targets.is_none())
387        {
388            for parent in parents {
389                if parent
390                    .source_result()?
391                    .is_some_and(|result| result.source_targets.is_some())
392                {
393                    return Err(invalid(
394                        "source evolution drops inherited reference closure",
395                    ));
396                }
397            }
398        }
399        match &self.body {
400            ThreadOperationBody::Capture(bytes) => {
401                bytes.author.validate()?;
402                if let SourceAuthor::Account { spool, .. } = &bytes.author
403                    && spool.to_string() != genesis.spool
404                {
405                    return Err(invalid("original source author crosses Spool scope"));
406                }
407                let state = State::decode_current_msgpack(&bytes.result.state)?;
408                let mut source_parents = BTreeSet::new();
409                for parent in parents {
410                    source_parents.insert(
411                        parent
412                            .source_state()?
413                            .ok_or_else(|| invalid("capture parent is not source"))?
414                            .id(),
415                    );
416                }
417                let declared: BTreeSet<_> = state
418                    .parents
419                    .iter()
420                    .copied()
421                    .filter(|id| *id != genesis.base)
422                    .collect();
423                if declared != source_parents
424                    || state.parents.is_empty()
425                    || state.parents.len() != state.parents.iter().collect::<BTreeSet<_>>().len()
426                {
427                    return Err(invalid(
428                        "capture source ancestry differs from causal parents",
429                    ));
430                }
431            }
432            ThreadOperationBody::Integration(bytes) => {
433                let receipt = integration::HostedIntegration::decode(bytes)?;
434                receipt.validate_operation(self)?;
435                receipt.validate_parents(genesis, parents)?;
436            }
437            ThreadOperationBody::HostedImport(bytes) => {
438                let receipt = hosted_import::HostedImport::decode(bytes)?;
439                receipt.validate_operation(self)?;
440                receipt.validate_parents(genesis, parents)?;
441            }
442            ThreadOperationBody::LocalIntegration(bytes) => {
443                let receipt = local_integration::LocalIntegration::decode(bytes)?;
444                receipt.validate_operation(self)?;
445                receipt.validate_parents(genesis, parents)?;
446            }
447            ThreadOperationBody::Metadata(bytes) => {
448                metadata::ThreadControl::decode(bytes)?.validate_parents(genesis, parents)?;
449            }
450            ThreadOperationBody::Context(bytes) => {
451                let context = crate::object::ContextRevision::decode(bytes).map_err(invalid)?;
452                if context.metadata.scope.spool.to_string() != genesis.spool {
453                    return Err(invalid("context belongs to another spool"));
454                }
455                if parents.is_empty() && context.extracted_from.is_some() {
456                    return Err(invalid(
457                        "context extraction requires a signed discussion resolution",
458                    ));
459                }
460                for parent in parents {
461                    let parent = parent.context_revision()?.ok_or_else(|| {
462                        invalid("context parent is not a context revision or extraction")
463                    })?;
464                    if parent.id != context.id
465                        || parent.metadata.scope != context.metadata.scope
466                        || parent.extracted_from != context.extracted_from
467                    {
468                        return Err(invalid("context parents belong to another record or scope"));
469                    }
470                }
471            }
472            ThreadOperationBody::Discussion(bytes) => {
473                let operation = CollaborationOperationEnvelope::decode(bytes).map_err(invalid)?;
474                if operation
475                    .operation
476                    .metadata
477                    .as_ref()
478                    .is_some_and(|m| m.scope.spool.to_string() != genesis.spool)
479                {
480                    return Err(invalid("collaboration metadata belongs to another spool"));
481                }
482                let mut discussion_parents = BTreeSet::new();
483                for parent in parents {
484                    let ThreadOperationBody::Discussion(bytes) = &parent.body else {
485                        return Err(invalid("discussion parent is not discussion"));
486                    };
487                    let parent = CollaborationOperationEnvelope::decode(bytes).map_err(invalid)?;
488                    if parent.operation.discussion_id != operation.operation.discussion_id {
489                        return Err(invalid("parents cross discussions"));
490                    }
491                    discussion_parents.insert(parent.operation_id);
492                }
493                if discussion_parents != operation.operation.parents.iter().copied().collect() {
494                    return Err(invalid(
495                        "discussion causality differs from its canonical operation",
496                    ));
497                }
498            }
499        }
500        Ok(())
501    }
502}
503
504fn bounded(bytes: &[u8]) -> Result<()> {
505    if bytes.len() > MAX_OPERATION_BYTES {
506        return Err(invalid("Thread record exceeds the durable record bound"));
507    }
508    Ok(())
509}
510
511fn invalid(message: impl std::fmt::Display) -> HeddleError {
512    HeddleError::InvalidObject(message.to_string())
513}
514
515#[cfg(test)]
516mod capture_shape_tests {
517    use super::*;
518    use crate::object::{Attribution, Principal, Tree};
519    #[test]
520    fn byte_only_capture_shape_is_rejected_in_clean_cutover() {
521        #[derive(serde::Serialize)]
522        struct OldBody {
523            kind: &'static str,
524            canonical: Vec<u8>,
525        }
526        #[derive(serde::Serialize)]
527        struct OldOperation {
528            version: u16,
529            thread: ContentHash,
530            parents: BTreeSet<ContentHash>,
531            publisher: [u8; 32],
532            body: OldBody,
533        }
534        let state = State::new_snapshot(
535            Tree::new().hash(),
536            vec![],
537            Attribution::human(Principal::new("author", "")),
538        );
539        let bytes = rmp_serde::to_vec_named(&OldOperation {
540            version: 1,
541            thread: ContentHash::from_bytes([1; 32]),
542            parents: BTreeSet::new(),
543            publisher: [41; 32],
544            body: OldBody {
545                kind: "capture",
546                canonical: state.encode_current_msgpack().expect("state"),
547            },
548        })
549        .expect("old wire bytes");
550        assert!(
551            ThreadOperation::decode(&bytes).is_err(),
552            "capture has one typed shape, without a legacy byte decoder"
553        );
554    }
555}
heddle_object_model/object/thread_replication.rs

heddle_object_model/object/
thread_replication.rs
