qualia_core_db/
wal.rs

1#![allow(unused)]
2#[cfg(not(target_arch = "wasm32"))]
3use std::fs::{File, OpenOptions};
4use std::io::{self, Read, Seek, SeekFrom, Write};
5use std::path::Path;
6
7use ed25519_dalek::{Signature, SigningKey};
8
9use crate::agency::{scrub_quin_volatile, sign_graph_mutation, stamp_fiduciary_metadata};
10use crate::crdt::SuspendedTransactionQueue;
11use crate::modalities::logic::core::WebizenOpcode;
12use crate::NQuin;
13use crate::PermissiveRoutingLane;
14
15// ── WAL file layout ──────────────────────────────────────────────────────────
16//
17// Byte 0..32   — `prev_dag_hash`: SHA-256 of the last DagNode committed from this WAL.
18//                All-zero = no prior checkpoint.
19// Byte 32..    — Packed 48-byte NQuin records (append-only).
20//
21// On `checkpoint_to_dag()` the current quins are committed to the DagStore, then
22// `prev_dag_hash` is rewritten in-place and the NQuin region is truncated.
23// On crash-recovery the `prev_dag_hash` survives so the new checkpoint chains
24// correctly onto the previous DAG node.
25
26/// Magic sentinel written as the message for WAL checkpoint DagNodes.
27const WAL_CHECKPOINT_MSG: &str = "wal:checkpoint";
28/// Size of the fixed WAL header (just the prev_dag_hash).
29const WAL_HEADER_SIZE: u64 = 32;
30
31/// The Write-Ahead Log (WAL) ensures mobile fault tolerance by appending all
32/// 48-byte Quin mutations directly to flash memory synchronously before they are
33/// packed into the larger 40KB SuperBlock structures.
34///
35/// The WAL also maintains a `prev_dag_hash` linking each checkpoint into the
36/// `git_bridge::DagStore` Merkle-DAG so the full write history is content-addressed.
37pub struct WriteAheadLog {
38    #[cfg(not(target_arch = "wasm32"))]
39    file: File,
40    #[cfg(target_arch = "wasm32")]
41    file: Vec<u8>,
42    /// SHA-256 of the most recent DagNode committed from this WAL.
43    /// `[0u8; 32]` means no prior checkpoint — next call to `checkpoint_to_dag` triggers `genesis_node`.
44    pub prev_dag_hash: [u8; 32],
45}
46
47#[cfg(not(target_arch = "wasm32"))]
48impl WriteAheadLog {
49    /// Opens or creates the append-only WAL file at `path`.
50    ///
51    /// If the file already has a 32-byte header (prior checkpoint hash), it is read
52    /// back so the chain is preserved across restarts.
53    pub fn open<P: AsRef<Path>>(path: P) -> std::io::Result<Self> {
54        let mut file = OpenOptions::new()
55            .create(true)
56            .write(true)
57            .read(true)
58            .open(path)?;
59
60        let file_len = file.seek(SeekFrom::End(0))?;
61
62        // Read existing prev_dag_hash from the header, or write a zero header for new files.
63        let prev_dag_hash = if file_len >= WAL_HEADER_SIZE {
64            file.seek(SeekFrom::Start(0))?;
65            let mut h = [0u8; 32];
66            file.read_exact(&mut h)?;
67            // Seek back to end for appending.
68            file.seek(SeekFrom::End(0))?;
69            h
70        } else {
71            // New or empty file: write zero header so the layout is established.
72            file.seek(SeekFrom::Start(0))?;
73            file.write_all(&[0u8; 32])?;
74            file.sync_all()?;
75            file.seek(SeekFrom::End(0))?;
76            [0u8; 32]
77        };
78
79        Ok(Self {
80            file,
81            prev_dag_hash,
82        })
83    }
84
85    /// Synchronously appends a NQuin to the log and flushes to disk.
86    /// This prevents data loss if the OS kills the process.
87    pub fn append_mutation(&mut self, quin: &NQuin) -> std::io::Result<()> {
88        // Always append — the file cursor is already at the end after open().
89        self.file.write_all(quin_as_bytes(quin))?;
90        self.file.sync_all()?;
91        Ok(())
92    }
93
94    /// Append with volatile field scrub after durable sync (wipes transient reasoning state).
95    pub fn append_mutation_volatile(&mut self, quin: &mut NQuin) -> std::io::Result<()> {
96        self.file.write_all(quin_as_bytes(quin))?;
97        self.file.sync_all()?;
98        scrub_quin_volatile(quin);
99        Ok(())
100    }
101
102    /// Reconstructs uncommitted NQuins from the WAL (skips the 32-byte header).
103    pub fn recover(&mut self) -> std::io::Result<Vec<NQuin>> {
104        self.file.seek(SeekFrom::Start(WAL_HEADER_SIZE))?;
105
106        let mut buffer = Vec::new();
107        self.file.read_to_end(&mut buffer)?;
108
109        let quin_size = std::mem::size_of::<NQuin>();
110        let mut recovered = Vec::with_capacity(buffer.len() / quin_size);
111
112        // Only read complete 48-byte chunks — partial chunks mean a mid-write crash; discard.
113        for chunk in buffer.chunks_exact(quin_size) {
114            let quin: NQuin = unsafe { std::ptr::read_unaligned(chunk.as_ptr() as *const NQuin) };
115            recovered.push(quin);
116        }
117
118        Ok(recovered)
119    }
120
121    /// Wipes the NQuin region of the WAL after a SuperBlock commit, preserving the header.
122    ///
123    /// Call `checkpoint_to_dag` **before** this so the hash chain is updated first.
124    pub fn truncate(&mut self) -> std::io::Result<()> {
125        self.file.set_len(WAL_HEADER_SIZE)?;
126        self.file.seek(SeekFrom::End(0))?;
127        self.file.sync_all()?;
128        Ok(())
129    }
130
131    /// Commit the current WAL contents as a DagNode, updating `prev_dag_hash`.
132    ///
133    /// If the WAL is empty (no quins since the last checkpoint), this is a no-op and
134    /// returns the existing `prev_dag_hash`.
135    ///
136    /// Typical call sequence:
137    /// ```text
138    /// wal.checkpoint_to_dag(&mut dag_store, author_did, now_ms())?;
139    /// wal.truncate()?;
140    /// ```
141    pub fn checkpoint_to_dag(
142        &mut self,
143        dag_store: &mut crate::git_bridge::DagStore,
144        author_did: u64,
145        timestamp_ms: u64,
146    ) -> std::io::Result<[u8; 32]> {
147        let quins = self.recover()?;
148        if quins.is_empty() {
149            return Ok(self.prev_dag_hash);
150        }
151
152        let new_hash = if self.prev_dag_hash == [0u8; 32] {
153            dag_store.genesis_node(&quins, author_did, timestamp_ms, WAL_CHECKPOINT_MSG)
154        } else {
155            dag_store.commit_node(
156                self.prev_dag_hash,
157                &quins,
158                author_did,
159                timestamp_ms,
160                WAL_CHECKPOINT_MSG,
161            )
162        };
163
164        // Persist the new hash into the WAL header so it survives a crash.
165        self.file.seek(SeekFrom::Start(0))?;
166        self.file.write_all(&new_hash)?;
167        self.file.sync_all()?;
168        // Restore cursor to end for continued appending.
169        self.file.seek(SeekFrom::End(0))?;
170
171        self.prev_dag_hash = new_hash;
172        Ok(new_hash)
173    }
174
175    /// Return the number of NQuin records currently buffered in the WAL.
176    pub fn buffered_count(&mut self) -> std::io::Result<usize> {
177        let file_len = self.file.seek(SeekFrom::End(0))?;
178        let data_len = file_len.saturating_sub(WAL_HEADER_SIZE);
179        Ok((data_len as usize) / std::mem::size_of::<NQuin>())
180    }
181}
182
183#[cfg(target_arch = "wasm32")]
184impl WriteAheadLog {
185    pub fn open<P: AsRef<Path>>(_path: P) -> std::io::Result<Self> {
186        Ok(Self {
187            file: Vec::new(),
188            prev_dag_hash: [0; 32],
189        })
190    }
191    pub fn append_mutation(&mut self, _quin: &NQuin) -> std::io::Result<()> {
192        Ok(())
193    }
194    pub fn append_mutation_volatile(&mut self, quin: &mut NQuin) -> std::io::Result<()> {
195        scrub_quin_volatile(quin);
196        Ok(())
197    }
198    pub fn recover(&mut self) -> std::io::Result<Vec<NQuin>> {
199        Ok(Vec::new())
200    }
201    pub fn truncate(&mut self) -> std::io::Result<()> {
202        Ok(())
203    }
204    pub fn checkpoint_to_dag(
205        &mut self,
206        _dag_store: &mut crate::git_bridge::DagStore,
207        _author_did: u64,
208        _timestamp_ms: u64,
209    ) -> std::io::Result<[u8; 32]> {
210        Ok(self.prev_dag_hash)
211    }
212    pub fn buffered_count(&mut self) -> std::io::Result<usize> {
213        Ok(0)
214    }
215}
216
217#[inline]
218fn quin_as_bytes(quin: &NQuin) -> &[u8] {
219    unsafe {
220        std::slice::from_raw_parts(
221            (quin as *const NQuin) as *const u8,
222            std::mem::size_of::<NQuin>(),
223        )
224    }
225}
226
227/// Outcome of routing a sieve-emitted Quin into the ledger pipeline.
228#[derive(Debug, Clone, Copy, PartialEq, Eq)]
229pub enum WalHandoffResult {
230    Committed,
231    Suspended { agreement_id: u64 },
232}
233
234/// Fiduciary-stamped, signed WAL handoff for neuro-symbolic graph mutations (zero String parse).
235pub fn commit_semantic_mutation(
236    wal: &mut WriteAheadLog,
237    quin: &mut NQuin,
238    principal_did_hash: u64,
239    agent_did_hash: u64,
240    signing_key: &SigningKey,
241    suspended: &mut SuspendedTransactionQueue,
242) -> io::Result<WalHandoffResult> {
243    stamp_fiduciary_metadata(quin, principal_did_hash, agent_did_hash);
244    let _sig: Signature = sign_graph_mutation(signing_key, quin);
245
246    if quin.identify_routing_lane() == PermissiveRoutingLane::EnforceBilateralMicroCommons {
247        let agreement_id = quin.context;
248        let tx = crate::crdt::SuspendedTransaction {
249            agreement_id,
250            threshold: 2,
251            collected_signatures: 1,
252            registers: [None; 16],
253            bytecode_buffer: [None; 64],
254            yielded_op: Some(WebizenOpcode::LoadModel(0)),
255            suspended_quin: *quin,
256        };
257        if suspended.push(tx).is_err() {
258            return Err(std::io::Error::new(
259                std::io::ErrorKind::Other,
260                "suspended transaction queue full",
261            ));
262        }
263        scrub_quin_volatile(quin);
264        return Ok(WalHandoffResult::Suspended { agreement_id });
265    }
266
267    wal.append_mutation_volatile(quin)?;
268    Ok(WalHandoffResult::Committed)
269}
270
271/// Appends a mutation to the global Write-Ahead Log.
272/// For the MVP/MCP, we write to a default location or stderr.
273pub fn append_mutation(quin: &NQuin) -> io::Result<()> {
274    // In a real implementation this would use a globally managed WAL lock.
275    // For now we open it locally or just pass.
276    let mut wal = WriteAheadLog::open("qualia_global.wal")?;
277    wal.append_mutation(quin)
278}
279
280/// Logs an adversarial conduct violation to the WAL when the Sentinel VM halts execution.
281pub fn log_adversarial_conduct(
282    intent_quin: &NQuin,
283    violation_code: u8,
284    vm_cycles: u64,
285) -> io::Result<()> {
286    let violation_quin = NQuin {
287        subject: intent_quin.subject,
288        predicate: crate::q_hash("q42:conductViolation"),
289        object: intent_quin.object,
290        context: intent_quin.context,
291        metadata: (violation_code as u64) | (vm_cycles << 8),
292        parity: 0,
293    };
294    append_mutation(&violation_quin)
295}
296
297/// Logs a rule-evaluation audit event to the WAL.
298///
299/// For each `RuleResult` in `results`, appends a `q42:ruleEvaluation` Quin to the
300/// WAL with:
301/// - `subject`    = the evaluated Quin's subject
302/// - `predicate`  = `q_hash("q42:ruleEvaluation")`
303/// - `object`     = `q_hash(rule_name)` (the rule that was evaluated)
304/// - `context`    = the contract graph hash
305/// - `metadata`   = bit 0 = passed/failed; bits [8..15] = result count
306///
307/// This ensures every rule evaluation is durable, replayable, and auditable —
308/// the general-purpose event API complement to `log_adversarial_conduct`.
309#[cfg(any(
310    not(target_arch = "wasm32"),
311    feature = "wasm-scientific",
312    feature = "wasm-full"
313))]
314pub fn log_rule_evaluation(
315    input_quin: &NQuin,
316    results: &[crate::modalities::logic::rules::RuleResult],
317    contract_hash: u64,
318) -> io::Result<()> {
319    let eval_predicate = crate::modalities::logic::rules::RULE_EVAL_PREDICATE;
320    let result_count = results.len().min(255) as u64;
321    for (i, result) in results.iter().enumerate().take(255) {
322        let rule_hash = crate::q_hash(&result.rule_name);
323        let mut metadata = if result.passed { 1u64 } else { 0u64 };
324        metadata |= (i as u64) << 16;
325        metadata |= result_count << 8;
326        let eval_quin = NQuin {
327            subject: input_quin.subject,
328            predicate: eval_predicate,
329            object: rule_hash,
330            context: contract_hash,
331            metadata,
332            parity: 0,
333        };
334        append_mutation(&eval_quin)?;
335    }
336    Ok(())
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342    use tempfile::NamedTempFile;
343
344    fn make_quin(subject: u64, object: u64) -> NQuin {
345        NQuin {
346            subject,
347            predicate: 2,
348            object,
349            context: 4,
350            metadata: 0,
351            parity: 0,
352        }
353    }
354
355    #[test]
356    fn wal_append_and_recover() {
357        let tmp = NamedTempFile::new().unwrap();
358        let mut wal = WriteAheadLog::open(tmp.path()).unwrap();
359
360        wal.append_mutation(&make_quin(1, 3)).unwrap();
361        wal.append_mutation(&make_quin(10, 30)).unwrap();
362
363        let recovered = wal.recover().unwrap();
364        assert_eq!(recovered.len(), 2, "WAL must recover 2 quins");
365        assert_eq!(recovered[0].subject, 1);
366        assert_eq!(recovered[1].object, 30);
367
368        wal.truncate().unwrap();
369        assert_eq!(wal.recover().unwrap().len(), 0, "WAL truncation failed");
370    }
371
372    #[test]
373    fn wal_header_persists_across_reopen() {
374        let tmp = NamedTempFile::new().unwrap();
375        {
376            let mut wal = WriteAheadLog::open(tmp.path()).unwrap();
377            assert_eq!(wal.prev_dag_hash, [0u8; 32]);
378            wal.append_mutation(&make_quin(42, 99)).unwrap();
379        }
380        // Reopen — header must still be there; quin must be recoverable.
381        {
382            let mut wal = WriteAheadLog::open(tmp.path()).unwrap();
383            assert_eq!(wal.prev_dag_hash, [0u8; 32]);
384            let quins = wal.recover().unwrap();
385            assert_eq!(quins.len(), 1);
386            assert_eq!(quins[0].subject, 42);
387        }
388    }
389
390    #[test]
391    fn wal_checkpoint_to_dag_chains_nodes() {
392        let tmp = NamedTempFile::new().unwrap();
393        let mut wal = WriteAheadLog::open(tmp.path()).unwrap();
394        let mut dag = crate::git_bridge::DagStore::new();
395        const AUTHOR: u64 = 0xA007_0001;
396
397        // First checkpoint — should produce a genesis node.
398        wal.append_mutation(&make_quin(1, 1)).unwrap();
399        wal.append_mutation(&make_quin(2, 2)).unwrap();
400        let hash1 = wal.checkpoint_to_dag(&mut dag, AUTHOR, 1000).unwrap();
401        assert_ne!(hash1, [0u8; 32], "genesis hash must be non-zero");
402        assert_eq!(wal.prev_dag_hash, hash1);
403        wal.truncate().unwrap();
404        assert_eq!(wal.recover().unwrap().len(), 0);
405
406        // Second checkpoint — should produce a commit node chained from hash1.
407        wal.append_mutation(&make_quin(3, 3)).unwrap();
408        let hash2 = wal.checkpoint_to_dag(&mut dag, AUTHOR, 2000).unwrap();
409        assert_ne!(hash2, hash1, "second checkpoint must have a different hash");
410        assert_eq!(wal.prev_dag_hash, hash2);
411        wal.truncate().unwrap();
412
413        // Verify DAG has 2 nodes.
414        let serialized = dag.serialize();
415        assert!(!serialized.is_empty());
416    }
417
418    #[test]
419    fn wal_checkpoint_empty_wal_is_noop() {
420        let tmp = NamedTempFile::new().unwrap();
421        let mut wal = WriteAheadLog::open(tmp.path()).unwrap();
422        let mut dag = crate::git_bridge::DagStore::new();
423
424        // Empty WAL — checkpoint must return the existing (zero) prev_dag_hash.
425        let hash = wal.checkpoint_to_dag(&mut dag, 0, 0).unwrap();
426        assert_eq!(hash, [0u8; 32]);
427        // An empty DagStore serializes to just the 8-byte node-count header (count=0).
428        let serialized = dag.serialize();
429        let node_count = u64::from_le_bytes(serialized[..8].try_into().unwrap());
430        assert_eq!(node_count, 0, "no DagNodes should be created for empty WAL");
431    }
432
433    #[test]
434    fn wal_buffered_count() {
435        let tmp = NamedTempFile::new().unwrap();
436        let mut wal = WriteAheadLog::open(tmp.path()).unwrap();
437        assert_eq!(wal.buffered_count().unwrap(), 0);
438        wal.append_mutation(&make_quin(1, 1)).unwrap();
439        wal.append_mutation(&make_quin(2, 2)).unwrap();
440        assert_eq!(wal.buffered_count().unwrap(), 2);
441    }
442}
qualia_core_db/wal.rs

qualia_core_db/
wal.rs
