qualia_client_core/wellfair/
live_share.rs

1//! Pending companion live-section requests and usage agreements (owner approval gate).
2
3use std::fs::{self, OpenOptions};
4use std::io::{BufRead, BufReader, Write};
5use std::path::{Path, PathBuf};
6
7use serde::{Deserialize, Serialize};
8use wellfare_core::live_share::{LiveSectionDecision, LiveSectionRequest, UsageAgreement};
9
10use super::journal::{JournalEntry, WellfairJournal};
11use super::sanctuary::is_sanctuary_protected_kind;
12
13pub const LIVE_SHARE_REQUESTS_FILE: &str = "wellfair/live_share_requests.jsonl";
14pub const USAGE_AGREEMENTS_FILE: &str = "wellfair/usage_agreements.jsonl";
15pub const MAX_PENDING: usize = 64;
16
17#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
18#[serde(rename_all = "snake_case")]
19pub enum LiveShareRequestStatus {
20    Pending,
21    Approved,
22    Denied,
23}
24
25#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
26pub struct LiveShareRequestRecord {
27    pub request: LiveSectionRequest,
28    pub enqueued_at_unix: u64,
29    pub status: LiveShareRequestStatus,
30    /// Sanctuary-classified kinds present in the request (therapy_note, sanctuary_note, welfare_case).
31    pub classified_kinds: Vec<String>,
32    pub requires_owner_approval: bool,
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    pub approved: Option<bool>,
35    #[serde(default, skip_serializing_if = "Option::is_none")]
36    pub projection_kinds: Option<Vec<String>>,
37    #[serde(default, skip_serializing_if = "Option::is_none")]
38    pub decided_at_unix: Option<u64>,
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub deny_reason: Option<String>,
41}
42
43impl LiveShareRequestRecord {
44    pub fn from_request(request: LiveSectionRequest, enqueued_at_unix: u64) -> Self {
45        let classified_kinds: Vec<String> = request
46            .requested_kinds
47            .iter()
48            .filter(|k| is_sanctuary_protected_kind(k))
49            .cloned()
50            .collect();
51        let requires_owner_approval = !classified_kinds.is_empty();
52        Self {
53            request,
54            enqueued_at_unix,
55            status: LiveShareRequestStatus::Pending,
56            classified_kinds,
57            requires_owner_approval,
58            approved: None,
59            projection_kinds: None,
60            decided_at_unix: None,
61            deny_reason: None,
62        }
63    }
64}
65
66pub struct LiveShareStore {
67    requests_path: PathBuf,
68    agreements_path: PathBuf,
69}
70
71impl LiveShareStore {
72    pub fn open(storage_root: impl AsRef<Path>) -> std::io::Result<Self> {
73        let root = storage_root.as_ref();
74        let requests_path = root.join(LIVE_SHARE_REQUESTS_FILE);
75        let agreements_path = root.join(USAGE_AGREEMENTS_FILE);
76        for path in [&requests_path, &agreements_path] {
77            if let Some(parent) = path.parent() {
78                fs::create_dir_all(parent)?;
79            }
80            if !path.exists() {
81                OpenOptions::new().create(true).write(true).open(path)?;
82            }
83        }
84        Ok(Self {
85            requests_path,
86            agreements_path,
87        })
88    }
89
90    pub fn enqueue_request(
91        &self,
92        request: LiveSectionRequest,
93        now_unix: u64,
94    ) -> std::io::Result<LiveShareRequestRecord> {
95        let record = LiveShareRequestRecord::from_request(request, now_unix);
96        let line =
97            serde_json::to_string(&record).map_err(|e| std::io::Error::other(e.to_string()))?;
98        let mut file = OpenOptions::new().append(true).open(&self.requests_path)?;
99        writeln!(file, "{line}")?;
100        file.sync_all()?;
101        Ok(record)
102    }
103
104    pub fn list_pending(&self, limit: usize) -> std::io::Result<Vec<LiveSectionRequest>> {
105        let all = self.load_requests()?;
106        let mut pending: Vec<LiveSectionRequest> = all
107            .into_iter()
108            .filter(|r| r.status == LiveShareRequestStatus::Pending)
109            .map(|r| r.request)
110            .collect();
111        let keep = limit.min(MAX_PENDING).min(pending.len());
112        if pending.len() > keep {
113            pending.drain(0..pending.len() - keep);
114        }
115        pending.reverse();
116        Ok(pending)
117    }
118
119    pub fn get_request(&self, request_id: &str) -> std::io::Result<Option<LiveShareRequestRecord>> {
120        Ok(self
121            .load_requests()?
122            .into_iter()
123            .find(|r| r.request.id == request_id))
124    }
125
126    pub fn decide(
127        &self,
128        request_id: &str,
129        approved: bool,
130        projection_kinds: &[String],
131        decided_at_unix: u64,
132        deny_reason: Option<&str>,
133    ) -> std::io::Result<LiveShareRequestRecord> {
134        let all = self.load_requests()?;
135        let mut found: Option<LiveShareRequestRecord> = None;
136        let mut rewritten = Vec::with_capacity(all.len());
137        for mut record in all {
138            if record.request.id == request_id {
139                if record.status != LiveShareRequestStatus::Pending {
140                    return Err(std::io::Error::other(format!(
141                        "live share request '{request_id}' already decided"
142                    )));
143                }
144                record.status = if approved {
145                    LiveShareRequestStatus::Approved
146                } else {
147                    LiveShareRequestStatus::Denied
148                };
149                record.approved = Some(approved);
150                record.projection_kinds = if approved {
151                    Some(projection_kinds.to_vec())
152                } else {
153                    Some(vec![])
154                };
155                record.decided_at_unix = Some(decided_at_unix);
156                record.deny_reason = deny_reason.map(str::to_string);
157                found = Some(record.clone());
158            }
159            rewritten.push(record);
160        }
161        let updated = found.ok_or_else(|| {
162            std::io::Error::new(
163                std::io::ErrorKind::NotFound,
164                format!("live share request '{request_id}' not found"),
165            )
166        })?;
167        self.rewrite_requests(&rewritten)?;
168        Ok(updated)
169    }
170
171    pub fn save_usage_agreement(&self, agreement: &UsageAgreement) -> std::io::Result<()> {
172        let line =
173            serde_json::to_string(agreement).map_err(|e| std::io::Error::other(e.to_string()))?;
174        let mut file = OpenOptions::new()
175            .append(true)
176            .open(&self.agreements_path)?;
177        writeln!(file, "{line}")?;
178        file.sync_all()?;
179        Ok(())
180    }
181
182    pub fn get_usage_agreement(&self, device_id: &str) -> std::io::Result<Option<UsageAgreement>> {
183        let file = fs::File::open(&self.agreements_path)?;
184        let reader = BufReader::new(file);
185        let mut latest: Option<UsageAgreement> = None;
186        for line in reader.lines() {
187            let line = line?;
188            if line.trim().is_empty() {
189                continue;
190            }
191            if let Ok(agreement) = serde_json::from_str::<UsageAgreement>(&line) {
192                if agreement.device_id == device_id {
193                    latest = Some(agreement);
194                }
195            }
196        }
197        Ok(latest)
198    }
199
200    fn load_requests(&self) -> std::io::Result<Vec<LiveShareRequestRecord>> {
201        let file = fs::File::open(&self.requests_path)?;
202        let reader = BufReader::new(file);
203        let mut records = Vec::new();
204        for line in reader.lines() {
205            let line = line?;
206            if line.trim().is_empty() {
207                continue;
208            }
209            if let Ok(record) = serde_json::from_str::<LiveShareRequestRecord>(&line) {
210                records.push(record);
211            }
212        }
213        Ok(records)
214    }
215
216    fn rewrite_requests(&self, records: &[LiveShareRequestRecord]) -> std::io::Result<()> {
217        let tmp = self.requests_path.with_extension("jsonl.tmp");
218        {
219            let mut file = OpenOptions::new()
220                .create(true)
221                .write(true)
222                .truncate(true)
223                .open(&tmp)?;
224            for record in records {
225                let line = serde_json::to_string(record)
226                    .map_err(|e| std::io::Error::other(e.to_string()))?;
227                writeln!(file, "{line}")?;
228            }
229            file.sync_all()?;
230        }
231        fs::rename(&tmp, &self.requests_path)?;
232        Ok(())
233    }
234}
235
236/// Fail-closed validation before owner approval: projection kinds must be requested,
237/// and sanctuary-classified kinds require an unlocked (non-decoy) sanctuary session.
238pub fn validate_live_share_decision(
239    record: &LiveShareRequestRecord,
240    approved: bool,
241    projection_kinds: &[String],
242    sanctuary_unlocked: bool,
243) -> Result<(), String> {
244    if !approved {
245        return Ok(());
246    }
247    for kind in projection_kinds {
248        if !record.request.requested_kinds.iter().any(|k| k == kind) {
249            return Err(format!(
250                "projection kind '{kind}' was not in the companion request (fail closed)"
251            ));
252        }
253        if is_sanctuary_protected_kind(kind) && !sanctuary_unlocked {
254            return Err(format!(
255                "sanctuary protected kind '{kind}' requires explicit owner approval after sanctuary unlock"
256            ));
257        }
258    }
259    Ok(())
260}
261
262pub fn sanctuary_allows_classified_projection(prefs: &super::sanctuary::SanctuaryPrefs) -> bool {
263    prefs.enabled && !prefs.locked && !prefs.decoy_session
264}
265
266pub fn live_share_request_journal_entry(
267    record: &LiveShareRequestRecord,
268    committed_unix: u32,
269) -> JournalEntry {
270    let sensitivity = if record.requires_owner_approval {
271        "Classified"
272    } else {
273        "Restricted"
274    };
275    JournalEntry {
276        id: format!("urn:wellfair:live_share_request:{}", record.request.id),
277        kind: "live_share_request".into(),
278        asserted_time_unix: record.enqueued_at_unix as u32,
279        evidence_type: "SelfReported".into(),
280        sensitivity: sensitivity.into(),
281        blob_hash: None,
282        source: "wellfair:live_share".into(),
283        committed_unix,
284        summary: Some(
285            serde_json::json!({
286                "request_id": record.request.id,
287                "device_id": record.request.device_id,
288                "purpose": record.request.purpose,
289                "requested_kinds": record.request.requested_kinds,
290                "classified_kinds": record.classified_kinds,
291                "requires_owner_approval": record.requires_owner_approval,
292            })
293            .to_string(),
294        ),
295    }
296}
297
298pub fn live_share_decision_journal_entry(
299    record: &LiveShareRequestRecord,
300    committed_unix: u32,
301) -> JournalEntry {
302    let approved = record.approved.unwrap_or(false);
303    let projection = record.projection_kinds.clone().unwrap_or_default();
304    JournalEntry {
305        id: format!("urn:wellfair:live_share_decision:{}", record.request.id),
306        kind: "live_share_decision".into(),
307        asserted_time_unix: record.decided_at_unix.unwrap_or(0) as u32,
308        evidence_type: "SelfReported".into(),
309        sensitivity: if projection.iter().any(|k| is_sanctuary_protected_kind(k)) {
310            "Classified".into()
311        } else {
312            "Restricted".into()
313        },
314        blob_hash: None,
315        source: "wellfair:live_share".into(),
316        committed_unix,
317        summary: Some(
318            serde_json::json!({
319                "request_id": record.request.id,
320                "device_id": record.request.device_id,
321                "approved": approved,
322                "projection_kinds": projection,
323                "classified_kinds": record.classified_kinds,
324                "deny_reason": record.deny_reason,
325            })
326            .to_string(),
327        ),
328    }
329}
330
331/// Build the companion wire message after owner decision.
332pub fn live_section_decision_from_record(record: &LiveShareRequestRecord) -> LiveSectionDecision {
333    let decided_at = record.decided_at_unix.unwrap_or(0);
334    if record.approved.unwrap_or(false) {
335        LiveSectionDecision::approved(
336            &record.request.id,
337            record.projection_kinds.clone().unwrap_or_default(),
338            decided_at,
339        )
340    } else {
341        LiveSectionDecision::denied(
342            &record.request.id,
343            record
344                .deny_reason
345                .clone()
346                .unwrap_or_else(|| "owner denied live share request".into()),
347            decided_at,
348        )
349    }
350}
351
352pub fn append_live_share_journal(
353    storage_root: impl AsRef<Path>,
354    entry: &JournalEntry,
355) -> Result<(), String> {
356    WellfairJournal::open(storage_root.as_ref())
357        .map_err(|e| e.to_string())?
358        .append(entry)
359        .map_err(|e| e.to_string())
360}
361
362#[cfg(test)]
363mod tests {
364    use super::*;
365    use wellfare_core::live_share::LiveSectionRequest;
366
367    #[test]
368    fn enqueue_flags_classified_kinds() {
369        let dir = tempfile::tempdir().unwrap();
370        let store = LiveShareStore::open(dir.path()).unwrap();
371        let request = LiveSectionRequest::new(
372            "req-1",
373            "phone-1",
374            "Desktop",
375            "preview",
376            vec!["conditions".into(), "therapy_note".into()],
377            vec![],
378            300,
379        );
380        let record = store.enqueue_request(request, 1_700_000_000).unwrap();
381        assert!(record.requires_owner_approval);
382        assert_eq!(record.classified_kinds, vec!["therapy_note"]);
383        let pending = store.list_pending(8).unwrap();
384        assert_eq!(pending.len(), 1);
385        assert_eq!(pending[0].id, "req-1");
386    }
387
388    #[test]
389    fn usage_agreement_latest_per_device() {
390        let dir = tempfile::tempdir().unwrap();
391        let store = LiveShareStore::open(dir.path()).unwrap();
392        let first = UsageAgreement::new("phone-1", "vitals", vec!["sleep".into()], 100, 50);
393        let second = UsageAgreement::new(
394            "phone-1",
395            "vitals v2",
396            vec!["sleep".into(), "steps".into()],
397            200,
398            150,
399        );
400        store.save_usage_agreement(&first).unwrap();
401        store.save_usage_agreement(&second).unwrap();
402        let got = store.get_usage_agreement("phone-1").unwrap().unwrap();
403        assert_eq!(got.purpose, "vitals v2");
404    }
405}
qualia_client_core/wellfair/live_share.rs

qualia_client_core/wellfair/
live_share.rs
