heddle_thread_api/
thread_control.rs

1//! Prepare portable Thread edits directly from an observed field frontier.
2//! Signing is local; the resulting typed request is retained unchanged for retries.
3use std::collections::BTreeSet;
4
5use crypto::{Signer, thread_operation::SignedOperation};
6pub use heddle_object_model::object::thread_replication::metadata::{
7    Control, Destination, EndpointKind, Intent, Lifecycle, Review, ReviewCoverage, ReviewKind,
8    SharedFacet, SharingPolicy,
9    audience::{Audience, Invitee},
10    retention::{MaterialRetention, RetentionPolicy},
11};
12use heddle_object_model::object::{
13    CollaborationActor, ContentHash,
14    thread_replication::{
15        OPERATION_FORMAT, ThreadOperation, ThreadOperationBody,
16        metadata::{AUTHORITY_FORMAT, Property, ThreadControl, property_version},
17    },
18};
19use uuid::Uuid;
20
21use crate::{contract as wire, transport::Error};
22
23/// Verify canonical framing and the original operation signature before the
24/// receiver checks current original-author authority and causal admission.
25/// A signature alone never establishes Spool permissions.
26pub fn verify(record: &wire::SignedRecord) -> Result<ThreadOperation, Error> {
27    let operation = crate::replication::decode_record(record.clone())
28        .map_err(|_| Error::Protocol("invalid Thread control record"))?
29        .verify()
30        .map_err(|_| Error::Protocol("invalid Thread control signature"))?;
31    let ThreadOperationBody::Metadata(bytes) = &operation.body else {
32        return Err(Error::Protocol("Thread control requires metadata"));
33    };
34    ThreadControl::decode(bytes).map_err(|_| Error::Protocol("invalid Thread control"))?;
35    Ok(operation)
36}
37
38/// Public original-author evidence. Build the envelope with the shared capability
39/// verifier using independently enrolled account history and the original Biscuit.
40/// The receiving endpoint verifies it independently at admission.
41pub struct Author<'a> {
42    pub account: Uuid,
43    pub agent_id: Option<&'a str>,
44    pub authority_envelope: &'a [u8],
45}
46
47/// Canonical original operation, plus the exact observed versions for its RPC.
48/// A command never invents a frontier from a mutable projection or wall clock.
49pub struct PreparedControl {
50    pub record: wire::SignedRecord,
51    pub control: ThreadControl,
52    pub thread: wire::ThreadRef,
53    pub property_version: Vec<u8>,
54    pub thread_version: Vec<u8>,
55}
56impl PreparedControl {
57    pub fn sign(
58        observed: &wire::ThreadOverview,
59        control: Control,
60        author: Author<'_>,
61        operation_id: Uuid,
62        occurred_at_ms: i64,
63        signer: &impl Signer,
64    ) -> Result<Self, Error> {
65        let reference = observed
66            .r#ref
67            .as_ref()
68            .ok_or(Error::Protocol("Thread reference missing"))?;
69        let spool = reference
70            .spool
71            .as_ref()
72            .ok_or(Error::Protocol("Spool missing"))?;
73        let thread = hash(
74            &reference
75                .id
76                .as_ref()
77                .ok_or(Error::Protocol("Thread ID missing"))?
78                .value,
79        )?;
80        let control = ThreadControl {
81            version: 1,
82            spool: spool
83                .id
84                .parse()
85                .map_err(|_| Error::Protocol("Spool ID must be UUID"))?,
86            actor: CollaborationActor {
87                principal_id: author.account,
88                agent_id: author.agent_id.map(str::to_owned),
89            },
90            authority_digest: ContentHash::compute_typed(
91                AUTHORITY_FORMAT,
92                author.authority_envelope,
93            ),
94            authority_envelope: author.authority_envelope.to_vec(),
95            client_operation_id: operation_id,
96            occurred_at_ms,
97            control,
98        };
99        let property = control.property();
100        let (kind, record_id) = property_key(&property);
101        let mut matches = observed
102            .metadata_frontiers
103            .iter()
104            .filter(|frontier| frontier.property == kind as i32 && frontier.record_id == record_id);
105        let empty_review;
106        let frontier = match matches.next() {
107            Some(frontier) => frontier,
108            None if matches!(property, Property::Review(_)) => {
109                empty_review = wire::ThreadPropertyFrontier {
110                    property: kind as i32,
111                    record_id: record_id.clone(),
112                    version: property_version(thread, &property, &BTreeSet::new())
113                        .map_err(io_error)?
114                        .as_bytes()
115                        .to_vec(),
116                    operation_ids: vec![],
117                };
118                &empty_review
119            }
120            None => {
121                return Err(Error::Protocol(
122                    "observe the exact Thread property frontier before signing",
123                ));
124            }
125        };
126        if matches.next().is_some() || frontier.operation_ids.len() > 128 {
127            return Err(Error::Protocol(
128                "invalid or duplicate Thread property frontier",
129            ));
130        }
131        let parents = frontier
132            .operation_ids
133            .iter()
134            .map(|id| hash(id))
135            .collect::<Result<BTreeSet<_>, _>>()?;
136        if parents.len() != frontier.operation_ids.len()
137            || property_version(thread, &property, &parents)
138                .map_err(io_error)?
139                .as_bytes()
140                .as_slice()
141                != frontier.version
142        {
143            return Err(Error::Protocol(
144                "Thread property version does not bind its exact parents",
145            ));
146        }
147        let operation = ThreadOperation {
148            version: 1,
149            thread,
150            parents,
151            publisher: signer
152                .public_key()
153                .try_into()
154                .map_err(|_| Error::Protocol("Thread control requires Ed25519"))?,
155            body: ThreadOperationBody::Metadata(control.encode().map_err(io_error)?),
156        };
157        let signed = SignedOperation::sign(&operation, signer).map_err(io_error)?;
158        Ok(Self {
159            record: wire::SignedRecord {
160                format: OPERATION_FORMAT.into(),
161                canonical_record: signed.canonical,
162                signatures: vec![wire::RecordSignature {
163                    public_key: operation.publisher.to_vec(),
164                    signature: signed.signature,
165                }],
166            },
167            control,
168            thread: reference.clone(),
169            property_version: frontier.version.clone(),
170            thread_version: observed.version.clone(),
171        })
172    }
173    pub fn revise_intent(&self) -> Result<wire::ReviseIntentRequest, Error> {
174        let Control::Intent(value) = &self.control.control else {
175            return Err(Error::Protocol("control is not an intent"));
176        };
177        Ok(wire::ReviseIntentRequest {
178            client_operation_id: self.control.client_operation_id.to_string(),
179            thread: Some(self.thread.clone()),
180            expected_intent_version: self.property_version.clone(),
181            proposed_intent: Some(wire::ThreadIntent {
182                outcome: value.outcome.clone(),
183                acceptance_criteria: value.acceptance_criteria.clone(),
184                origin_urls: value.origin_urls.clone(),
185                principal_approved: value.principal_approved,
186                principal_id: self.control.actor.principal_id.to_string(),
187                agent_id: self.control.actor.agent_id.clone().unwrap_or_default(),
188                version: vec![],
189            }),
190            operation: Some(self.record.clone()),
191        })
192    }
193    pub fn rename(&self) -> Result<wire::RenameThreadRequest, Error> {
194        let Control::Name(name) = &self.control.control else {
195            return Err(Error::Protocol("control is not a name"));
196        };
197        if self.property_version.is_empty() {
198            return Err(Error::Protocol("observed property version missing"));
199        }
200        Ok(wire::RenameThreadRequest {
201            client_operation_id: self.control.client_operation_id.to_string(),
202            thread: Some(self.thread.clone()),
203            expected_version: self.property_version.clone(),
204            name: name.clone(),
205            operation: Some(self.record.clone()),
206        })
207    }
208    pub fn change_lifecycle(&self) -> Result<wire::ChangeThreadLifecycleRequest, Error> {
209        let Control::Lifecycle(value) = self.control.control else {
210            return Err(Error::Protocol("control is not lifecycle"));
211        };
212        if self.property_version.is_empty() {
213            return Err(Error::Protocol("observed property version missing"));
214        }
215        Ok(wire::ChangeThreadLifecycleRequest {
216            client_operation_id: self.control.client_operation_id.to_string(),
217            thread: Some(self.thread.clone()),
218            expected_version: self.property_version.clone(),
219            lifecycle: match value {
220                Lifecycle::Draft => wire::ThreadLifecycle::Draft,
221                Lifecycle::Active => wire::ThreadLifecycle::Active,
222                Lifecycle::Ready => wire::ThreadLifecycle::Ready,
223                Lifecycle::Abandoned => wire::ThreadLifecycle::Abandoned,
224            } as i32,
225            operation: Some(self.record.clone()),
226        })
227    }
228    pub fn record_review(&self) -> Result<wire::RecordReviewRequest, Error> {
229        let Control::Review(value) = &self.control.control else {
230            return Err(Error::Protocol("control is not a review"));
231        };
232        let record = wire::RecordRef {
233            spool: self.thread.spool.clone(),
234            id: value.id.to_string(),
235        };
236        let mut decision = wire::ReviewDecision {
237            r#ref: Some(record.clone()),
238            thread: Some(self.thread.clone()),
239            source: Some(self.revision(value.source)),
240            target: Some(self.revision(value.target)),
241            policy_version: value.policy_version.as_bytes().to_vec(),
242            principal_id: self.control.actor.principal_id.to_string(),
243            agent_id: self.control.actor.agent_id.clone().unwrap_or_default(),
244            kind: match value.kind {
245                ReviewKind::Opinion => wire::review_decision::Kind::Opinion,
246                ReviewKind::Approval => wire::review_decision::Kind::Approval,
247                ReviewKind::Rejection => wire::review_decision::Kind::Rejection,
248                ReviewKind::Revocation => wire::review_decision::Kind::Revocation,
249                ReviewKind::Read => wire::review_decision::Kind::Read,
250                ReviewKind::AgentPreview => wire::review_decision::Kind::AgentPreview,
251                ReviewKind::AgentCoReview => wire::review_decision::Kind::AgentCoReview,
252            } as i32,
253            explanation: value.explanation.clone(),
254            revokes: value.revokes.map(|id| wire::RecordRef {
255                spool: self.thread.spool.clone(),
256                id: id.to_string(),
257            }),
258            expires_at: None,
259            coverage: value
260                .coverage
261                .as_ref()
262                .map(|coverage| wire::ReviewCoverage {
263                    selection: Some(match coverage {
264                        ReviewCoverage::WholeSource => {
265                            wire::review_coverage::Selection::WholeSource(true)
266                        }
267                        ReviewCoverage::Symbols(anchors) => {
268                            wire::review_coverage::Selection::Symbols(wire::ReviewSymbols {
269                                anchors: anchors
270                                    .iter()
271                                    .map(|anchor| wire::ReviewSymbolAnchor {
272                                        path: anchor.file.clone(),
273                                        symbol: anchor.symbol.clone(),
274                                    })
275                                    .collect(),
276                            })
277                        }
278                    }),
279                }),
280        };
281        if let Some(seconds) = value.expires_at_unix_seconds {
282            decision.expires_at = Some(Default::default());
283            if let Some(timestamp) = decision.expires_at.as_mut() {
284                timestamp.seconds = seconds;
285            }
286        }
287        Ok(wire::RecordReviewRequest {
288            client_operation_id: self.control.client_operation_id.to_string(),
289            decision: Some(decision),
290            expected_versions: vec![wire::ExpectedVersion {
291                resource: Some(wire::EntityRef {
292                    entity: Some(wire::entity_ref::Entity::Review(record)),
293                }),
294                version: self.property_version.clone(),
295            }],
296            operation: Some(self.record.clone()),
297        })
298    }
299    fn revision(&self, state: heddle_object_model::object::StateId) -> wire::RevisionRef {
300        wire::RevisionRef {
301            spool: self.thread.spool.clone(),
302            revision: Some(wire::revision_ref::Revision::State(
303                api::heddle::api::common::StateId {
304                    value: state.as_bytes().to_vec(),
305                },
306            )),
307        }
308    }
309    pub fn set_audience(&self) -> Result<wire::SetThreadAudienceRequest, Error> {
310        let Control::Audience(value) = &self.control.control else {
311            return Err(Error::Protocol("control is not audience"));
312        };
313        let (kind, invitees) = match value {
314            Audience::Owner => (wire::thread_audience_policy::Kind::Owner, vec![]),
315            Audience::Spool => (wire::thread_audience_policy::Kind::Spool, vec![]),
316            Audience::Invited(invitees) => (
317                wire::thread_audience_policy::Kind::Invited,
318                invitees
319                    .iter()
320                    .map(|invitee| wire::ThreadInvitee {
321                        principal_id: invitee.principal_id.to_string(),
322                        agent_id: invitee.agent_id.clone().unwrap_or_default(),
323                    })
324                    .collect(),
325            ),
326        };
327        Ok(wire::SetThreadAudienceRequest {
328            client_operation_id: self.control.client_operation_id.to_string(),
329            expected_policy_version: self.property_version.clone(),
330            operation: Some(self.record.clone()),
331            policy: Some(wire::ThreadAudiencePolicy {
332                thread: Some(self.thread.clone()),
333                version: vec![],
334                kind: kind as i32,
335                invitees,
336            }),
337        })
338    }
339    pub fn set_retention(&self) -> Result<wire::SetThreadRetentionRequest, Error> {
340        let Control::Retention(value) = &self.control.control else {
341            return Err(Error::Protocol("control is not retention"));
342        };
343        Ok(wire::SetThreadRetentionRequest {
344            client_operation_id: self.control.client_operation_id.to_string(),
345            expected_policy_version: self.property_version.clone(),
346            operation: Some(self.record.clone()),
347            policy: Some(wire::ThreadRetentionPolicy {
348                thread: Some(self.thread.clone()),
349                version: vec![],
350                source: Some(retention_wire(value.source)),
351                collaboration: Some(retention_wire(value.collaboration)),
352                evidence: Some(retention_wire(value.evidence)),
353                scrubbed_timeline: Some(retention_wire(value.scrubbed_timeline)),
354                raw_transcripts: Some(retention_wire(value.raw_transcripts)),
355            }),
356        })
357    }
358    pub fn set_sharing(&self) -> Result<wire::SetThreadSharingRequest, Error> {
359        let Control::Sharing(value) = &self.control.control else {
360            return Err(Error::Protocol("control is not sharing"));
361        };
362        Ok(wire::SetThreadSharingRequest {
363            client_operation_id: self.control.client_operation_id.to_string(),
364            expected_policy_version: self.property_version.clone(),
365            operation: Some(self.record.clone()),
366            policy: Some(wire::ThreadSharingPolicy {
367                thread: Some(self.thread.clone()),
368                version: vec![],
369                ongoing: value.ongoing,
370                destinations: value
371                    .destinations
372                    .iter()
373                    .map(|destination| wire::SharingDestination {
374                        endpoint: Some(wire::EndpointRef {
375                            public_key: destination.endpoint.to_vec(),
376                            kind: match destination.kind {
377                                EndpointKind::Device => wire::EndpointKind::Device,
378                                EndpointKind::Weft => wire::EndpointKind::Weft,
379                            } as i32,
380                        }),
381                        spool: Some(wire::SpoolRef {
382                            id: destination.spool.to_string(),
383                        }),
384                        facets: destination
385                            .facets
386                            .iter()
387                            .map(|facet| match facet {
388                                SharedFacet::Source => wire::SharedFacet::Source,
389                                SharedFacet::Collaboration => wire::SharedFacet::Collaboration,
390                                SharedFacet::Evidence => wire::SharedFacet::Evidence,
391                                SharedFacet::ScrubbedTimeline => {
392                                    wire::SharedFacet::ScrubbedTimeline
393                                }
394                                SharedFacet::Metadata => wire::SharedFacet::Metadata,
395                            } as i32)
396                            .collect(),
397                    })
398                    .collect(),
399            }),
400        })
401    }
402}
403
404fn retention_wire(value: MaterialRetention) -> wire::MaterialRetention {
405    let (mode, seconds) = match value {
406        MaterialRetention::Discard => (wire::material_retention::Mode::Discard, 0),
407        MaterialRetention::Bounded(seconds) => (wire::material_retention::Mode::Bounded, seconds),
408        MaterialRetention::Retain => (wire::material_retention::Mode::Retain, 0),
409    };
410    wire::MaterialRetention {
411        mode: mode as i32,
412        seconds,
413    }
414}
415
416pub fn property_key(property: &Property) -> (wire::ThreadProperty, String) {
417    match property {
418        Property::Name => (wire::ThreadProperty::Name, String::new()),
419        Property::Intent => (wire::ThreadProperty::Intent, String::new()),
420        Property::Lifecycle => (wire::ThreadProperty::Lifecycle, String::new()),
421        Property::Sharing => (wire::ThreadProperty::Sharing, String::new()),
422        Property::Audience => (wire::ThreadProperty::Audience, String::new()),
423        Property::Retention => (wire::ThreadProperty::Retention, String::new()),
424        Property::Review(id) => (wire::ThreadProperty::Review, id.to_string()),
425    }
426}
427fn hash(bytes: &[u8]) -> Result<ContentHash, Error> {
428    Ok(ContentHash::from_bytes(bytes.try_into().map_err(|_| {
429        Error::Protocol("Thread hash must be 32 bytes")
430    })?))
431}
432fn io_error(error: impl std::fmt::Display) -> Error {
433    Error::Io(error.to_string())
434}
435
436#[cfg(test)]
437#[path = "thread_control_tests.rs"]
438mod tests;
heddle_thread_api/thread_control.rs

heddle_thread_api/
thread_control.rs
