heddle_object_model/object/thread_replication/
local_integration.rs

1//! Authored device integration. This is ordinary source work, never a hosted
2//! executor's policy approval. Admission separately authorizes both Threads and
3//! verifies the referenced original source operation before storing this record.
4use std::collections::BTreeSet;
5
6use serde::{Deserialize, Serialize};
7
8use super::{ThreadGenesis, ThreadOperation, bounded, invalid};
9use crate::{
10    error::Result,
11    object::{ContentHash, State, StateId, VisibilityTier},
12};
13
14#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(deny_unknown_fields)]
16pub struct LocalIntegration {
17    pub version: u16,
18    pub spool: uuid::Uuid,
19    pub device: [u8; 32],
20    pub author: super::SourceAuthor,
21    pub source_thread: ContentHash,
22    pub source_operation: ContentHash,
23    pub source_revision: StateId,
24    pub target_thread: ContentHash,
25    pub expected_target_frontier: BTreeSet<ContentHash>,
26    pub result: super::Capture,
27    pub result_visibility: VisibilityTier,
28    /// Audit binding to whatever initiated the integration: the verified
29    /// device-RPC request proof on the device path, a digest of the request
30    /// tuple on the CLI path. Never an authority proof; heddle admission does
31    /// not read it.
32    pub initiating_request_proof: ContentHash,
33    pub local_policy_version: ContentHash,
34    pub executed_at_ms: i64,
35}
36impl LocalIntegration {
37    pub fn encode(&self) -> Result<Vec<u8>> {
38        self.author.validate()?;
39        if matches!(&self.author, super::SourceAuthor::Account { spool, .. } if *spool != self.spool)
40        {
41            return Err(invalid("local integration author crosses Spool scope"));
42        }
43        if self.version != 1
44            || self.spool.is_nil()
45            || self.source_thread == self.target_thread
46            || self.expected_target_frontier.len() > 128
47            || self.executed_at_ms < 0
48        {
49            return Err(invalid("invalid local integration receipt"));
50        }
51        let state = self.result.validated_state()?;
52        if state.encode_current_msgpack()? != self.result.state {
53            return Err(invalid("non-canonical local integration State"));
54        }
55        let bytes = rmp_serde::to_vec_named(self)?;
56        bounded(&bytes)?;
57        Ok(bytes)
58    }
59    pub fn decode(bytes: &[u8]) -> Result<Self> {
60        bounded(bytes)?;
61        let receipt: Self = rmp_serde::from_slice(bytes)?;
62        if receipt.encode()? != bytes {
63            return Err(invalid("non-canonical local integration receipt"));
64        }
65        Ok(receipt)
66    }
67    pub fn resulting_state(&self) -> Result<State> {
68        self.result.validated_state()
69    }
70    /// Supply the independently authenticated original record from the named
71    /// source Thread, never a record selected solely by a claimed revision hash.
72    pub fn validate_source(&self, source: &ThreadOperation) -> Result<()> {
73        if self.result.source_targets.is_none()
74            && source
75                .source_result()?
76                .is_some_and(|result| result.source_targets.is_some())
77        {
78            return Err(invalid("integration drops source reference closure"));
79        }
80        if source.thread != self.source_thread
81            || source.id()? != self.source_operation
82            || source
83                .source_state()?
84                .is_none_or(|state| state.id() != self.source_revision)
85        {
86            return Err(invalid(
87                "local integration source differs from original operation",
88            ));
89        }
90        Ok(())
91    }
92    pub(super) fn validate_operation(&self, operation: &ThreadOperation) -> Result<()> {
93        if operation.thread != self.target_thread
94            || operation.publisher != self.device
95            || operation.parents != self.expected_target_frontier
96        {
97            return Err(invalid(
98                "local integration changes device, target or observed frontier",
99            ));
100        }
101        Ok(())
102    }
103    pub(super) fn validate_parents(
104        &self,
105        genesis: &ThreadGenesis,
106        parents: &[ThreadOperation],
107    ) -> Result<()> {
108        if genesis.spool != self.spool.to_string() {
109            return Err(invalid("local integration belongs to another Spool"));
110        }
111        let state = self.resulting_state()?;
112        let mut expected = BTreeSet::from([self.source_revision]);
113        if parents.is_empty() {
114            expected.insert(genesis.base);
115        }
116        for parent in parents {
117            expected.insert(
118                parent
119                    .source_state()?
120                    .ok_or_else(|| invalid("local integration parent is not source"))?
121                    .id(),
122            );
123        }
124        // Even a same-tree local landing creates an attributed merge record;
125        // its cross-Thread ancestry is explicit rather than executor-attested.
126        if state.parents.iter().copied().collect::<BTreeSet<_>>() != expected
127            || state.parents.len() != expected.len()
128        {
129            return Err(invalid("local integration drops source or target ancestry"));
130        }
131        Ok(())
132    }
133}
134
135/// Intersection of representable disclosure sets. Different named audiences
136/// require an explicit policy decision, never a rank-only lateral downgrade.
137pub fn intersect_visibility(
138    left: &VisibilityTier,
139    right: &VisibilityTier,
140) -> Result<VisibilityTier> {
141    use VisibilityTier::*;
142    if left == right {
143        return Ok(left.clone());
144    }
145    match (left, right) {
146        (Public, other) | (other, Public) => Ok(other.clone()),
147        (Internal, other) | (other, Internal) => Ok(other.clone()),
148        (Private { scope_label: a }, Restricted { scope_label: b })
149        | (Restricted { scope_label: b }, Private { scope_label: a })
150            if a == b =>
151        {
152            Ok(Private {
153                scope_label: a.clone(),
154            })
155        }
156        _ => Err(invalid("integration combines incomparable named audiences")),
157    }
158}
heddle_object_model/object/thread_replication/local_integration.rs

heddle_object_model/object/thread_replication/
local_integration.rs
