| """ |
| Bifrost WORM-Link — Append-only hash chain for Guardian Omega FSM. |
| Every state transition is sealed. No mutation. No deletion. |
| Authors: Ahmad Ali Parr, Jessica L. Williams (SNAPKITTYWEST) |
| """ |
|
|
| import hashlib |
| import json |
| import time |
| from dataclasses import dataclass, field |
| from typing import List, Optional |
|
|
|
|
| @dataclass(frozen=True) |
| class WORMEntry: |
| seq: int |
| timestamp: float |
| state_from: str |
| event: str |
| state_to: str |
| action: str |
| record_id: Optional[str] |
| prev_hash: str |
| entry_hash: str |
|
|
|
|
| class BifrostWORMChain: |
| """ |
| Append-only chain for FSM state transitions. |
| Properties: |
| - Monotonic: seq always increases |
| - Immutable: entries cannot be modified after append |
| - Verifiable: hash chain links every entry to genesis |
| - Deterministic: same inputs produce same hashes |
| """ |
|
|
| def __init__(self): |
| self._chain: List[WORMEntry] = [] |
| self._genesis_hash = hashlib.sha256(b"BIFROST_GENESIS_2026").hexdigest() |
|
|
| @property |
| def tip(self) -> str: |
| if not self._chain: |
| return self._genesis_hash |
| return self._chain[-1].entry_hash |
|
|
| @property |
| def length(self) -> int: |
| return len(self._chain) |
|
|
| def append(self, state_from: str, event: str, state_to: str, |
| action: str, record_id: Optional[str] = None) -> WORMEntry: |
| """Append a new entry. Returns the sealed entry.""" |
| seq = len(self._chain) |
| ts = time.time() |
| prev = self.tip |
|
|
| |
| payload = f"{seq}|{state_from}|{event}|{state_to}|{action}|{record_id}|{prev}" |
| entry_hash = hashlib.sha256(payload.encode()).hexdigest() |
|
|
| entry = WORMEntry( |
| seq=seq, |
| timestamp=ts, |
| state_from=state_from, |
| event=event, |
| state_to=state_to, |
| action=action, |
| record_id=record_id, |
| prev_hash=prev, |
| entry_hash=entry_hash |
| ) |
| self._chain.append(entry) |
| return entry |
|
|
| def verify(self) -> bool: |
| """Verify entire chain integrity.""" |
| if not self._chain: |
| return True |
|
|
| |
| if self._chain[0].prev_hash != self._genesis_hash: |
| return False |
|
|
| |
| for i in range(1, len(self._chain)): |
| if self._chain[i].prev_hash != self._chain[i - 1].entry_hash: |
| return False |
|
|
| |
| e = self._chain[i] |
| payload = f"{e.seq}|{e.state_from}|{e.event}|{e.state_to}|{e.action}|{e.record_id}|{e.prev_hash}" |
| expected = hashlib.sha256(payload.encode()).hexdigest() |
| if e.entry_hash != expected: |
| return False |
|
|
| |
| for i in range(1, len(self._chain)): |
| if self._chain[i].timestamp < self._chain[i - 1].timestamp: |
| return False |
|
|
| return True |
|
|
| def export_json(self) -> str: |
| """Export chain as JSON (for audit).""" |
| return json.dumps([ |
| { |
| "seq": e.seq, |
| "timestamp": e.timestamp, |
| "state_from": e.state_from, |
| "event": e.event, |
| "state_to": e.state_to, |
| "action": e.action, |
| "record_id": e.record_id, |
| "prev_hash": e.prev_hash, |
| "entry_hash": e.entry_hash |
| } |
| for e in self._chain |
| ], indent=2) |
|
|
| def rollback_point(self, seq: int) -> Optional[str]: |
| """ |
| Find the hash at a given sequence number for recovery. |
| Does NOT mutate the chain — rollback creates a NEW branch. |
| """ |
| if seq < 0 or seq >= len(self._chain): |
| return None |
| return self._chain[seq].entry_hash |
|
|
|
|
| class GuardianOmegaFSM: |
| """ |
| The sovereign FSM with WORM-sealed transitions. |
| Every state change is recorded immutably. |
| """ |
|
|
| TRANSITIONS = { |
| ("idle", "record_arrived"): ("absorbing", "absorb"), |
| ("absorbing", "schema_valid"): ("classifying", "classify"), |
| ("absorbing", "schema_invalid"): ("idle", "reject"), |
| ("classifying", "factual_pass"): ("gating", "noop"), |
| ("classifying", "factual_fail"): ("idle", "reject"), |
| ("gating", "plasma_pass"): ("sealing", "seal_worm"), |
| ("gating", "plasma_fail"): ("idle", "reject"), |
| ("sealing", "worm_sealed"): ("idle", "approve"), |
| ("sealing", "worm_failed"): ("recovering", "rollback"), |
| ("quarantined", "recovery_complete"): ("idle", "noop"), |
| ("recovering", "recovery_complete"): ("idle", "noop"), |
| } |
|
|
| def __init__(self): |
| self.state = "idle" |
| self.chain = BifrostWORMChain() |
|
|
| def process_event(self, event: str, record_id: Optional[str] = None) -> str: |
| """Process event, transition state, seal to WORM chain.""" |
| |
| if event == "byzantine_detected": |
| old_state = self.state |
| self.state = "quarantined" |
| self.chain.append(old_state, event, "quarantined", "quarantine", record_id) |
| return "quarantine" |
|
|
| key = (self.state, event) |
| if key not in self.TRANSITIONS: |
| |
| self.chain.append(self.state, event, self.state, "noop", record_id) |
| return "noop" |
|
|
| new_state, action = self.TRANSITIONS[key] |
| self.chain.append(self.state, event, new_state, action, record_id) |
| self.state = new_state |
| return action |
|
|
| def verify_integrity(self) -> bool: |
| """Verify the WORM chain is intact.""" |
| return self.chain.verify() |
|
|
