src/continuity/shared_mem.py

"""
Paradigm 4: Cyclic Bitmaps & Shared Memory (ctypes + multiprocessing)
Part of SOVEREIGN PYTHON LLM ENGINE

Zero-copy state sharing between orchestrator and daemon processes.
RAM-speed, no serialization, no disk I/O, no TCP overhead.

Two backends:
  A. ctypes.create_string_buffer()
     - Single process / multiple threads
     - Direct memory pointer sharing
     - Fastest possible: byte mutation at array index

  B. multiprocessing.shared_memory (Python 3.8+ stdlib)
     - Cross-process on same machine
     - Named shared memory block, accessible by name
     - Survives across process restarts (by name)

Memory layout (fixed 4KB block):
  Offset  Size  Field
  0       8     MAGIC = b'SOVSTATE'
  8       8     version (uint64 BE) — incremented on every write
  16      8     agent_id hash (uint64 BE) — FNV-1a of agent_id string
  24      8     step (uint64 BE)
  32      8     timestamp_ns (uint64 BE)
  40      8     bitmask (uint64 BE) — 64 boolean flags
  48      8     sequence (uint64 BE) — monotonic write counter
  56      8     checksum (uint64 BE) — FNV-1a of bytes 0-55
  64      960   payload (raw bytes, caller-defined)
  1024    3072  reserved

Total fixed block = 4096 bytes (one memory page)
"""

from __future__ import annotations

import ctypes
import struct
import time
from pathlib import Path
from typing import Any

# ─────────────────────────────────────────────
# Constants
# ─────────────────────────────────────────────

BLOCK_SIZE   = 4096
HEADER_SIZE  = 64
PAYLOAD_SIZE = 960
MAGIC        = b'SOVSTATE'

HDR_FMT      = '>8sQQQQQQQ'   # magic(8) ver(8) id_hash(8) step(8) ts(8) mask(8) seq(8) chk(8)
HDR_BYTES    = struct.calcsize(HDR_FMT)  # should be 64

assert HDR_BYTES == HEADER_SIZE, f"Header size mismatch: {HDR_BYTES}"


# ─────────────────────────────────────────────
# FNV-1a hash (fast, no imports)
# ─────────────────────────────────────────────

def _fnv1a_64(data: bytes) -> int:
    h = 0xcbf29ce484222325
    for b in data:
        h ^= b
        h = (h * 0x100000001b3) & 0xFFFFFFFFFFFFFFFF
    return h


# ─────────────────────────────────────────────
# Shared State Block (in-process, ctypes buffer)
# ─────────────────────────────────────────────

class SharedStateBlock:
    """
    Single 4KB memory block for agent state.
    Uses ctypes.create_string_buffer — works within one process
    across threads with no locking needed for reads
    (writes are single-byte atomic on x86/ARM).

    Usage (single process, multiple threads):
        block = SharedStateBlock(agent_id="react_1")
        block.set_flag(2)         # set bit 2
        block.set_step(5)
        block.write_payload(b"custom data")

        # From another thread:
        print(block.get_flags())  # fast bitmask read
        print(block.get_step())
    """

    def __init__(self, agent_id: str, block_size: int = BLOCK_SIZE):
        self._agent_id   = agent_id
        self._id_hash    = _fnv1a_64(agent_id.encode('utf-8'))
        self._version    = 0
        self._sequence   = 0
        self._buf        = ctypes.create_string_buffer(block_size)
        self._block_size = block_size
        self._flush_header(mask=0, step=0)

    def _flush_header(self, mask: int, step: int) -> None:
        self._version  += 1
        self._sequence += 1
        ts = time.time_ns()

        # Build header without checksum first
        hdr_no_chk = struct.pack(
            '>8sQQQQQQ',
            MAGIC,
            self._version,
            self._id_hash,
            step,
            ts,
            mask,
            self._sequence
        )
        checksum = _fnv1a_64(hdr_no_chk)
        hdr = hdr_no_chk + struct.pack('>Q', checksum)

        # Write to ctypes buffer
        self._buf[:HEADER_SIZE] = hdr

    def _read_header(self) -> tuple | None:
        raw = bytes(self._buf[:HEADER_SIZE])
        try:
            (magic, ver, id_hash, step, ts, mask, seq, chk) = struct.unpack(HDR_FMT, raw)
            if magic != MAGIC:
                return None
            # Verify checksum
            expected = _fnv1a_64(raw[:56])
            if chk != expected:
                return None
            return (ver, id_hash, step, ts, mask, seq)
        except Exception:
            return None

    # ── Flag operations ────────────────────────

    def set_flag(self, bit: int) -> None:
        hdr = self._read_header()
        mask = hdr[4] if hdr else 0
        step = hdr[2] if hdr else 0
        mask |= (1 << (bit & 63))
        self._flush_header(mask=mask, step=step)

    def clear_flag(self, bit: int) -> None:
        hdr = self._read_header()
        mask = hdr[4] if hdr else 0
        step = hdr[2] if hdr else 0
        mask &= ~(1 << (bit & 63))
        self._flush_header(mask=mask, step=step)

    def get_flags(self) -> int:
        hdr = self._read_header()
        return hdr[4] if hdr else 0

    def has_flag(self, bit: int) -> bool:
        return bool(self.get_flags() & (1 << (bit & 63)))

    def set_flags_from_set(self, bits: set[int]) -> None:
        mask = sum(1 << (b & 63) for b in bits)
        hdr = self._read_header()
        step = hdr[2] if hdr else 0
        self._flush_header(mask=mask, step=step)

    def active_bits(self) -> set[int]:
        mask = self.get_flags()
        return {i for i in range(64) if mask & (1 << i)}

    # ── Step ────────────────────────────────────

    def set_step(self, step: int) -> None:
        hdr = self._read_header()
        mask = hdr[4] if hdr else 0
        self._flush_header(mask=mask, step=step)

    def get_step(self) -> int:
        hdr = self._read_header()
        return hdr[2] if hdr else 0

    # ── Payload ─────────────────────────────────

    def write_payload(self, data: bytes) -> None:
        n = min(len(data), PAYLOAD_SIZE)
        self._buf[HEADER_SIZE:HEADER_SIZE + n] = data[:n]
        # Zero rest
        if n < PAYLOAD_SIZE:
            self._buf[HEADER_SIZE + n:HEADER_SIZE + PAYLOAD_SIZE] = b'\x00' * (PAYLOAD_SIZE - n)

    def read_payload(self, length: int | None = None) -> bytes:
        n = length if length is not None else PAYLOAD_SIZE
        n = min(n, PAYLOAD_SIZE)
        return bytes(self._buf[HEADER_SIZE:HEADER_SIZE + n])

    # ── Snapshot ────────────────────────────────

    def snapshot(self) -> dict[str, Any]:
        hdr = self._read_header()
        if not hdr:
            return {"valid": False}
        ver, id_hash, step, ts, mask, seq = hdr
        return {
            "valid":       True,
            "agent_id":    self._agent_id,
            "version":     ver,
            "step":        step,
            "timestamp_ns": ts,
            "flags_mask":  mask,
            "active_bits": list(self.active_bits()),
            "sequence":    seq,
        }

    def raw_bytes(self) -> bytes:
        return bytes(self._buf[:self._block_size])


# ─────────────────────────────────────────────
# Cross-process shared memory (multiprocessing.shared_memory)
# ─────────────────────────────────────────────

class SharedMemoryState:
    """
    Cross-process state block using Python 3.8+ multiprocessing.shared_memory.

    The memory block is identified by name — any process on the same
    machine can attach to it by name.

    Survives process restarts: if the block already exists,
    the new process attaches and reads existing state.

    Usage:
        # Process A (orchestrator)
        sm = SharedMemoryState("sovereign_react_1", create=True)
        sm.set_step(5)
        sm.set_flag(2)

        # Process B (daemon) — attaches to existing block
        sm2 = SharedMemoryState("sovereign_react_1", create=False)
        print(sm2.get_step())   # 5
        print(sm2.has_flag(2))  # True
    """

    def __init__(self, name: str, create: bool = True, size: int = BLOCK_SIZE):
        from multiprocessing import shared_memory

        self._name = name
        self._size = size

        try:
            if create:
                try:
                    self._shm = shared_memory.SharedMemory(name=name, create=True, size=size)
                    # Initialize with zeros
                    self._shm.buf[:size] = b'\x00' * size
                except FileExistsError:
                    # Already exists — attach
                    self._shm = shared_memory.SharedMemory(name=name, create=False, size=size)
            else:
                self._shm = shared_memory.SharedMemory(name=name, create=False, size=size)
        except Exception as e:
            # Fallback to ctypes buffer if shared memory unavailable
            self._shm = None
            self._fallback = ctypes.create_string_buffer(size)
            self._fallback_mode = True
            return

        self._fallback_mode = False

    def _read(self, offset: int, length: int) -> bytes:
        if self._fallback_mode:
            return bytes(self._fallback[offset:offset + length])
        return bytes(self._shm.buf[offset:offset + length])

    def _write(self, offset: int, data: bytes) -> None:
        if self._fallback_mode:
            self._fallback[offset:offset + len(data)] = data
        else:
            self._shm.buf[offset:offset + len(data)] = data

    def set_step(self, step: int) -> None:
        self._write(24, struct.pack('>Q', step))

    def get_step(self) -> int:
        raw = self._read(24, 8)
        return struct.unpack('>Q', raw)[0]

    def set_flag(self, bit: int) -> None:
        raw = self._read(40, 8)
        mask = struct.unpack('>Q', raw)[0]
        mask |= (1 << (bit & 63))
        self._write(40, struct.pack('>Q', mask))

    def clear_flag(self, bit: int) -> None:
        raw = self._read(40, 8)
        mask = struct.unpack('>Q', raw)[0]
        mask &= ~(1 << (bit & 63))
        self._write(40, struct.pack('>Q', mask))

    def has_flag(self, bit: int) -> bool:
        raw = self._read(40, 8)
        mask = struct.unpack('>Q', raw)[0]
        return bool(mask & (1 << (bit & 63)))

    def get_flags(self) -> int:
        raw = self._read(40, 8)
        return struct.unpack('>Q', raw)[0]

    def write_payload(self, data: bytes) -> None:
        n = min(len(data), PAYLOAD_SIZE)
        self._write(HEADER_SIZE, data[:n])

    def read_payload(self, length: int | None = None) -> bytes:
        n = length if length is not None else PAYLOAD_SIZE
        return self._read(HEADER_SIZE, min(n, PAYLOAD_SIZE))

    def close(self) -> None:
        if not self._fallback_mode and self._shm:
            self._shm.close()

    def unlink(self) -> None:
        if not self._fallback_mode and self._shm:
            self._shm.unlink()

    @property
    def name(self) -> str:
        return self._name