heddle_object_model/object/thread_replication/
metadata.rs

1//! Caller-signed Thread controls with independent causal registers. Concurrent
2//! writes to one property remain visible candidates; no arrival-time winner is
3//! selected. Controls describe intent and policy, never grant recipient rights.
4use std::collections::BTreeSet;
5
6use serde::{Deserialize, Serialize};
7use uuid::Uuid;
8
9pub mod audience;
10pub mod retention;
11
12use super::{MAX_OPERATION_BYTES, ThreadGenesis, ThreadOperation, ThreadOperationBody, invalid};
13use crate::{
14    error::Result,
15    object::{CollaborationActor, ContentHash, StateId},
16};
17
18pub const CONTROL_FORMAT: &str = "heddle-thread-control-v1";
19pub const PROPERTY_VERSION_FORMAT: &str = "heddle-thread-property-v1";
20pub const AUTHORITY_FORMAT: &str = "heddle-thread-control-authority-v1";
21
22#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
23#[serde(rename_all = "snake_case")]
24pub enum Property {
25    Name,
26    Intent,
27    Lifecycle,
28    Sharing,
29    Audience,
30    Retention,
31    Review(Uuid),
32}
33
34#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(deny_unknown_fields)]
36pub struct Intent {
37    pub outcome: String,
38    pub acceptance_criteria: Vec<String>,
39    pub origin_urls: Vec<String>,
40    pub principal_approved: bool,
41}
42#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(rename_all = "snake_case")]
44pub enum Lifecycle {
45    Draft,
46    Active,
47    Ready,
48    Abandoned,
49}
50
51#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
52#[serde(rename_all = "snake_case")]
53pub enum SharedFacet {
54    Source,
55    Collaboration,
56    Evidence,
57    ScrubbedTimeline,
58    Metadata,
59}
60#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
61#[serde(rename_all = "snake_case")]
62pub enum EndpointKind {
63    Device,
64    Weft,
65}
66#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(deny_unknown_fields)]
68pub struct Destination {
69    pub endpoint: [u8; 32],
70    pub kind: EndpointKind,
71    pub spool: Uuid,
72    pub facets: BTreeSet<SharedFacet>,
73}
74#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
75#[serde(deny_unknown_fields)]
76pub struct SharingPolicy {
77    pub ongoing: bool,
78    pub destinations: Vec<Destination>,
79}
80#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
81#[serde(rename_all = "snake_case")]
82pub enum ReviewKind {
83    Opinion,
84    Approval,
85    Rejection,
86    Revocation,
87    Read,
88    AgentPreview,
89    AgentCoReview,
90}
91#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
92#[serde(rename_all = "snake_case")]
93pub enum ReviewCoverage {
94    WholeSource,
95    Symbols(Vec<ReviewSymbolAnchor>),
96}
97#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
98#[serde(deny_unknown_fields)]
99pub struct ReviewSymbolAnchor {
100    pub file: String,
101    pub symbol: String,
102}
103#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
104#[serde(deny_unknown_fields)]
105pub struct Review {
106    pub id: Uuid,
107    pub source: StateId,
108    pub target: StateId,
109    pub policy_version: ContentHash,
110    pub kind: ReviewKind,
111    pub explanation: String,
112    pub revokes: Option<Uuid>,
113    pub expires_at_unix_seconds: Option<i64>,
114    #[serde(default, skip_serializing_if = "Option::is_none")]
115    pub coverage: Option<ReviewCoverage>,
116}
117#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
118#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
119pub enum Control {
120    Name(String),
121    Intent(Intent),
122    Lifecycle(Lifecycle),
123    Sharing(SharingPolicy),
124    Audience(audience::Audience),
125    Retention(retention::RetentionPolicy),
126    Review(Review),
127}
128#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
129#[serde(deny_unknown_fields)]
130pub struct ThreadControl {
131    pub version: u16,
132    pub spool: Uuid,
133    pub actor: CollaborationActor,
134    /// Portable original authority, independently checked by the admitting host.
135    /// The envelope grants nothing merely because its digest is well formed.
136    pub authority_digest: ContentHash,
137    pub authority_envelope: Vec<u8>,
138    pub client_operation_id: Uuid,
139    pub occurred_at_ms: i64,
140    pub control: Control,
141}
142impl ThreadControl {
143    /// Exact mutation whose original authority must be checked at admission.
144    pub fn authorization_method(&self) -> &'static str {
145        match self.control {
146            Control::Name(_) => "/heddle.api.v1alpha2.ThreadService/RenameThread",
147            Control::Intent(_) => "/heddle.api.v1alpha2.ThreadService/ReviseIntent",
148            Control::Lifecycle(_) => "/heddle.api.v1alpha2.ThreadService/ChangeLifecycle",
149            Control::Sharing(_) => "/heddle.api.v1alpha2.ThreadService/SetSharingPolicy",
150            Control::Audience(_) => "/heddle.api.v1alpha2.ThreadService/SetAudiencePolicy",
151            Control::Retention(_) => "/heddle.api.v1alpha2.ThreadService/SetRetentionPolicy",
152            Control::Review(_) => "/heddle.api.v1alpha2.ThreadService/RecordReview",
153        }
154    }
155    pub fn property(&self) -> Property {
156        match &self.control {
157            Control::Name(_) => Property::Name,
158            Control::Intent(_) => Property::Intent,
159            Control::Lifecycle(_) => Property::Lifecycle,
160            Control::Sharing(_) => Property::Sharing,
161            Control::Audience(_) => Property::Audience,
162            Control::Retention(_) => Property::Retention,
163            Control::Review(review) => Property::Review(review.id),
164        }
165    }
166    pub fn encode(&self) -> Result<Vec<u8>> {
167        if self.version != 1
168            || self.spool.is_nil()
169            || self.actor.principal_id.is_nil()
170            || self.client_operation_id.is_nil()
171            || self.occurred_at_ms < 0
172        {
173            return Err(invalid("invalid Thread control identity"));
174        }
175        if self.authority_envelope.is_empty()
176            || self.authority_envelope.len() > 64 * 1024
177            || ContentHash::compute_typed(AUTHORITY_FORMAT, &self.authority_envelope)
178                != self.authority_digest
179        {
180            return Err(invalid("invalid Thread control authority binding"));
181        }
182        if let Some(agent) = &self.actor.agent_id {
183            text(agent, 256, false)?;
184        }
185        match &self.control {
186            Control::Name(name) => text(name, 1024, false)?,
187            Control::Intent(intent) => {
188                text(&intent.outcome, 32768, false)?;
189                if intent.acceptance_criteria.len() > 64
190                    || intent.origin_urls.len() > 64
191                    || (intent.principal_approved && self.actor.agent_id.is_some())
192                {
193                    return Err(invalid("invalid Thread intent approval or bounds"));
194                }
195                for criterion in &intent.acceptance_criteria {
196                    text(criterion, 4096, false)?;
197                }
198                for origin in &intent.origin_urls {
199                    text(origin, 4096, false)?;
200                }
201            }
202            Control::Lifecycle(_) => {}
203            Control::Audience(policy) => policy.validate()?,
204            Control::Retention(policy) => policy.validate()?,
205            Control::Sharing(policy) => {
206                if policy.destinations.len() > 64 {
207                    return Err(invalid("Thread sharing destination bound"));
208                }
209                let mut seen = BTreeSet::new();
210                for destination in &policy.destinations {
211                    if destination.spool.is_nil()
212                        || destination.endpoint == [0; 32]
213                        || destination.facets.is_empty()
214                        || !seen.insert((destination.endpoint, destination.spool))
215                    {
216                        return Err(invalid("invalid or duplicate Thread sharing destination"));
217                    }
218                }
219            }
220            Control::Review(review) => {
221                text(&review.explanation, 32768, true)?;
222                let attestation = matches!(
223                    review.kind,
224                    ReviewKind::Read | ReviewKind::AgentPreview | ReviewKind::AgentCoReview
225                );
226                if attestation != review.coverage.is_some() {
227                    return Err(invalid("review coverage must match attestation kind"));
228                }
229                if let Some(ReviewCoverage::Symbols(anchors)) = &review.coverage {
230                    if anchors.is_empty() || anchors.len() > 128 {
231                        return Err(invalid("review symbol coverage exceeds bounds"));
232                    }
233                    for anchor in anchors {
234                        text(&anchor.file, 4096, false)?;
235                        text(&anchor.symbol, 1024, false)?;
236                        if anchor.file.starts_with('/')
237                            || anchor
238                                .file
239                                .split('/')
240                                .any(|part| part.is_empty() || part == "." || part == "..")
241                        {
242                            return Err(invalid(
243                                "review symbol path must be relative and canonical",
244                            ));
245                        }
246                    }
247                }
248                if matches!(
249                    review.kind,
250                    ReviewKind::AgentPreview | ReviewKind::AgentCoReview
251                ) && self.actor.agent_id.is_none()
252                {
253                    return Err(invalid("agent review attestation requires an agent actor"));
254                }
255                if review.id.is_nil()
256                    || review
257                        .revokes
258                        .is_some_and(|id| id.is_nil() || id == review.id)
259                    || (review.kind == ReviewKind::Revocation) != review.revokes.is_some()
260                    || review
261                        .expires_at_unix_seconds
262                        .is_some_and(|time| time <= self.occurred_at_ms / 1000)
263                {
264                    return Err(invalid("invalid Thread review decision"));
265                }
266            }
267        }
268        let bytes = rmp_serde::to_vec_named(self)?;
269        if bytes.len() > MAX_OPERATION_BYTES / 2 {
270            return Err(invalid("Thread control exceeds record budget"));
271        }
272        Ok(bytes)
273    }
274    pub fn decode(bytes: &[u8]) -> Result<Self> {
275        if bytes.len() > MAX_OPERATION_BYTES / 2 {
276            return Err(invalid("Thread control exceeds record budget"));
277        }
278        let value: Self = rmp_serde::from_slice(bytes)?;
279        if value.encode()? != bytes {
280            return Err(invalid("non-canonical Thread control"));
281        }
282        Ok(value)
283    }
284    pub fn validate_operation(&self, operation: &ThreadOperation) -> Result<()> {
285        if operation.parents.len() > 128 {
286            return Err(invalid("Thread property frontier exceeds budget"));
287        }
288        let ThreadOperationBody::Metadata(bytes) = &operation.body else {
289            return Err(invalid("Thread control requires metadata operation"));
290        };
291        if self.encode()? != *bytes {
292            return Err(invalid("Thread control differs from signed operation"));
293        }
294        Ok(())
295    }
296    pub fn validate_parents(
297        &self,
298        genesis: &ThreadGenesis,
299        parents: &[ThreadOperation],
300    ) -> Result<()> {
301        if self.spool.to_string() != genesis.spool {
302            return Err(invalid("Thread control belongs to another Spool"));
303        }
304        for parent in parents {
305            let ThreadOperationBody::Metadata(bytes) = &parent.body else {
306                return Err(invalid("Thread property parent is not metadata"));
307            };
308            let parent = Self::decode(bytes)?;
309            if parent.spool != self.spool || parent.property() != self.property() {
310                return Err(invalid("Thread property causal parents cross fields"));
311            }
312            if matches!(self.control, Control::Review(_)) && parent.actor != self.actor {
313                return Err(invalid("review successor changes original actor"));
314            }
315        }
316        Ok(())
317    }
318}
319/// A field's exact version commits its scope and every concurrent candidate.
320/// Genesis/default values have an empty frontier, with a stable nonempty version.
321pub fn property_version(
322    thread: ContentHash,
323    property: &Property,
324    heads: &BTreeSet<ContentHash>,
325) -> Result<ContentHash> {
326    if heads.len() > 128 {
327        return Err(invalid("Thread property frontier exceeds budget"));
328    }
329    Ok(ContentHash::compute_typed(
330        PROPERTY_VERSION_FORMAT,
331        &rmp_serde::to_vec_named(&(thread, property, heads))?,
332    ))
333}
334fn text(value: &str, max: usize, empty: bool) -> Result<()> {
335    if value.len() > max || (!empty && value.trim().is_empty()) || value.contains('\0') {
336        return Err(invalid("invalid Thread control text"));
337    }
338    Ok(())
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344    fn fixture() -> (ThreadGenesis, ThreadControl) {
345        let spool = Uuid::from_u128(1);
346        (
347            ThreadGenesis {
348                version: 1,
349                spool: spool.to_string(),
350                parent: None,
351                base: StateId::from_bytes([1; 32]),
352                name: "thread".into(),
353                intent: "intent".into(),
354                creator: [3; 32],
355                owner: super::super::GenesisOwner::Account(Uuid::from_u128(2)),
356                nonce: vec![],
357            },
358            ThreadControl {
359                version: 1,
360                spool,
361                actor: CollaborationActor {
362                    principal_id: Uuid::from_u128(2),
363                    agent_id: None,
364                },
365                authority_digest: ContentHash::compute_typed(
366                    AUTHORITY_FORMAT,
367                    b"codec fixture, independently verified at admission",
368                ),
369                authority_envelope: b"codec fixture, independently verified at admission".to_vec(),
370                client_operation_id: Uuid::from_u128(3),
371                occurred_at_ms: 1000,
372                control: Control::Name("new name".into()),
373            },
374        )
375    }
376    fn operation(
377        genesis: &ThreadGenesis,
378        value: &ThreadControl,
379        parents: BTreeSet<ContentHash>,
380    ) -> ThreadOperation {
381        ThreadOperation {
382            version: 1,
383            thread: genesis.id().expect("Thread"),
384            parents,
385            publisher: [3; 32],
386            body: ThreadOperationBody::Metadata(value.encode().expect("control")),
387        }
388    }
389    #[test]
390    fn canonical_control_retains_actor_and_value_in_original_operation_identity() {
391        let (genesis, value) = fixture();
392        let original = operation(&genesis, &value, BTreeSet::new());
393        let encoded = original.encode().expect("canonical operation");
394        assert_eq!(ThreadOperation::decode(&encoded).expect("decode"), original);
395        assert_eq!(original.facet(), super::super::ThreadFacet::Metadata);
396        original
397            .validate_parents(&genesis, &[])
398            .expect("initial property");
399        let mut changed = value;
400        changed.actor.principal_id = Uuid::from_u128(4);
401        assert_ne!(
402            original.id().expect("ID"),
403            operation(&genesis, &changed, BTreeSet::new())
404                .id()
405                .expect("changed actor")
406        );
407        changed.control = Control::Name("another value".into());
408        assert_ne!(
409            original.id().expect("ID"),
410            operation(&genesis, &changed, BTreeSet::new())
411                .id()
412                .expect("changed value")
413        );
414    }
415    #[test]
416    fn independent_properties_cannot_causally_erase_each_other() {
417        let (genesis, mut value) = fixture();
418        let name = operation(&genesis, &value, BTreeSet::new());
419        let name_id = name.id().expect("name ID");
420        value.control = Control::Lifecycle(Lifecycle::Active);
421        let lifecycle = operation(&genesis, &value, BTreeSet::from([name_id]));
422        assert!(
423            lifecycle
424                .validate_parents(&genesis, std::slice::from_ref(&name))
425                .expect_err("cannot dominate another field")
426                .to_string()
427                .contains("cross fields")
428        );
429        value.control = Control::Name("resolved name".into());
430        operation(&genesis, &value, BTreeSet::from([name_id]))
431            .validate_parents(&genesis, &[name])
432            .expect("same property successor");
433    }
434    #[test]
435    fn property_version_binds_every_concurrent_candidate_and_exact_field() {
436        let (genesis, value) = fixture();
437        let a = operation(&genesis, &value, BTreeSet::new());
438        let mut other = value;
439        other.control = Control::Name("parallel".into());
440        let b = operation(&genesis, &other, BTreeSet::new());
441        let heads = BTreeSet::from([a.id().expect("a"), b.id().expect("b")]);
442        let version = property_version(a.thread, &Property::Name, &heads).expect("version");
443        assert_ne!(
444            version,
445            property_version(
446                a.thread,
447                &Property::Name,
448                &BTreeSet::from([a.id().expect("a")])
449            )
450            .expect("incomplete frontier")
451        );
452        assert_ne!(
453            version,
454            property_version(a.thread, &Property::Intent, &heads).expect("other field")
455        );
456        operation(&genesis, &other, heads)
457            .validate_parents(&genesis, &[b, a])
458            .expect("explicit resolution retains both observed parents");
459    }
460    #[test]
461    fn bounded_controls_cannot_claim_principal_approval_for_agent_attribution() {
462        let (_, mut value) = fixture();
463        value.control = Control::Intent(Intent {
464            outcome: "goal".into(),
465            acceptance_criteria: vec![],
466            origin_urls: vec![],
467            principal_approved: true,
468        });
469        value.actor.agent_id = Some("delegated agent".into());
470        assert!(
471            value
472                .encode()
473                .expect_err("human approval is an explicit actor assertion")
474                .to_string()
475                .contains("approval")
476        );
477        value.actor.agent_id = None;
478        value.encode().expect("principal statement");
479        value.control = Control::Name("n".repeat(1025));
480        assert!(
481            value
482                .encode()
483                .expect_err("name limit")
484                .to_string()
485                .contains("text")
486        );
487    }
488    #[test]
489    fn original_authority_envelope_is_bound_by_the_signed_control() {
490        let (_, mut control) = fixture();
491        control.authority_envelope.push(1);
492        assert!(
493            control
494                .encode()
495                .expect_err("substituted proof must fail before signing")
496                .to_string()
497                .contains("authority binding")
498        );
499    }
500    #[test]
501    fn review_successor_retains_exact_original_human_or_agent_actor() {
502        let (genesis, mut value) = fixture();
503        value.control = Control::Review(Review {
504            id: Uuid::from_u128(5),
505            source: StateId::from_bytes([7; 32]),
506            target: StateId::from_bytes([8; 32]),
507            policy_version: ContentHash::from_bytes([9; 32]),
508            kind: ReviewKind::Approval,
509            explanation: "original decision".into(),
510            revokes: None,
511            expires_at_unix_seconds: None,
512            coverage: None,
513        });
514        let original = operation(&genesis, &value, BTreeSet::new());
515        let parent = BTreeSet::from([original.id().expect("review ID")]);
516        value.actor.agent_id = Some("delegated agent".into());
517        assert!(
518            operation(&genesis, &value, parent.clone())
519                .validate_parents(&genesis, std::slice::from_ref(&original))
520                .expect_err("agent cannot rewrite human decision")
521                .to_string()
522                .contains("original actor")
523        );
524        value.actor.agent_id = None;
525        operation(&genesis, &value, parent)
526            .validate_parents(&genesis, &[original])
527            .expect("same accountable reviewer");
528    }
529    #[test]
530    fn read_coverage_cannot_be_relabelled_as_approval_or_name_hidden_paths() {
531        let (_, mut value) = fixture();
532        value.control = Control::Review(Review {
533            id: Uuid::from_u128(6),
534            source: StateId::from_bytes([7; 32]),
535            target: StateId::from_bytes([8; 32]),
536            policy_version: ContentHash::from_bytes([9; 32]),
537            kind: ReviewKind::Read,
538            explanation: String::new(),
539            revokes: None,
540            expires_at_unix_seconds: None,
541            coverage: Some(ReviewCoverage::Symbols(vec![ReviewSymbolAnchor {
542                file: "src/main.rs".into(),
543                symbol: "run".into(),
544            }])),
545        });
546        value.encode().expect("bounded read attestation");
547        let Control::Review(review) = &mut value.control else {
548            panic!("review fixture")
549        };
550        review.kind = ReviewKind::Approval;
551        assert!(
552            value
553                .encode()
554                .expect_err("approval cannot carry read coverage")
555                .to_string()
556                .contains("coverage")
557        );
558        let Control::Review(review) = &mut value.control else {
559            panic!("review fixture")
560        };
561        review.kind = ReviewKind::AgentPreview;
562        review.coverage = Some(ReviewCoverage::Symbols(vec![ReviewSymbolAnchor {
563            file: "../private.rs".into(),
564            symbol: "run".into(),
565        }]));
566        assert!(
567            value
568                .encode()
569                .expect_err("coverage path must be canonical")
570                .to_string()
571                .contains("path")
572        );
573    }
574}
heddle_object_model/object/thread_replication/metadata.rs

heddle_object_model/object/thread_replication/
metadata.rs
