src/continuity/inode_state.py

"""
Paradigm 3: Filesystem Inode State (Zero-Byte File Gates)
Part of SOVEREIGN PYTHON LLM ENGINE

Boolean state flags stored as directory entries.
File existence = 1, absence = 0.
No file content ever written or read — pure stat() calls.

Structure:
  {base_dir}/
    {agent_id}/
      flags/
        IDLE          ← exists = agent is idle
        THINKING      ← exists = agent is thinking
        ACTING        ← exists = agent is acting
        OBSERVING     ← exists = agent is observing
        REFLECTING    ← exists = agent is reflecting
        DONE          ← exists = agent is done
        ERROR         ← exists = agent errored
      step/
        0000000000000005   ← filename = current step number (zero-padded)
      locks/
        tool:filesystem.read   ← exists = this tool is checked out

Why this is fast:
  - os.stat() is a single kernel syscall
  - Directory entries live in kernel dcache (hot in memory)
  - No open(), read(), write(), close() — just stat() + mkdir() + unlink()
  - Concurrent readers are safe (no file locking needed for boolean flags)
  - Corruption-proof: filesystem guarantees atomic rename/unlink

Use cases:
  - Agent state machine flags (7 states)
  - Tool checkout tracking (which tools are in use)
  - Mutex locks (file existence = lock held)
  - Rate limit gates (file mtime = last call timestamp)
"""

from __future__ import annotations

import os
import time
from pathlib import Path
from typing import Iterator


# ─────────────────────────────────────────────
# Inode State Gate
# ─────────────────────────────────────────────

class InodeGate:
    """
    Single boolean state gate backed by a zero-byte file.

    set()    → touch file (create if not exists)
    clear()  → unlink file (delete if exists)
    get()    → stat file (True if exists)
    mtime()  → file modification time (useful for rate limiting)
    """

    def __init__(self, path: Path):
        self.path = path
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def set(self) -> None:
        self.path.touch(exist_ok=True)

    def clear(self) -> None:
        try:
            self.path.unlink()
        except FileNotFoundError:
            pass

    def get(self) -> bool:
        return self.path.exists()

    def toggle(self) -> bool:
        if self.get():
            self.clear()
            return False
        else:
            self.set()
            return True

    def mtime(self) -> float | None:
        try:
            return self.path.stat().st_mtime
        except FileNotFoundError:
            return None

    def age_seconds(self) -> float | None:
        mt = self.mtime()
        return (time.time() - mt) if mt is not None else None

    def __bool__(self) -> bool:
        return self.get()

    def __repr__(self) -> str:
        return f"InodeGate({self.path.name}={'1' if self.get() else '0'})"


# ─────────────────────────────────────────────
# Inode State Machine
# ─────────────────────────────────────────────

# Standard agent state flags
AGENT_FLAGS = [
    "IDLE", "THINKING", "ACTING", "OBSERVING",
    "REFLECTING", "DONE", "ERROR"
]


class InodeStateMachine:
    """
    Agent state machine using zero-byte files as boolean flags.

    State transitions are atomic at the filesystem level:
      transition(remove={"THINKING"}, add={"ACTING"}) uses
      a rename-based swap to avoid intermediate invalid states.

    Directory structure:
      {base}/{agent_id}/flags/IDLE
      {base}/{agent_id}/flags/THINKING
      ...
      {base}/{agent_id}/step/{zero_padded_step}
      {base}/{agent_id}/locks/{lock_name}

    Usage:
        ism = InodeStateMachine(base_dir, agent_id="react_1")
        ism.transition(remove={"IDLE"}, add={"THINKING"})
        ism.set_step(1)
        print(ism.active_flags())   # ["THINKING"]
    """

    def __init__(self, base_dir: Path, agent_id: str):
        self.agent_id = agent_id
        self._root    = base_dir / agent_id
        self._flags   = self._root / "flags"
        self._step_dir = self._root / "step"
        self._locks   = self._root / "locks"

        # Create directory structure
        for d in (self._flags, self._step_dir, self._locks):
            d.mkdir(parents=True, exist_ok=True)

        # Initialize to IDLE if no flags set
        if not any(True for _ in self._flags.iterdir()):
            (self._flags / "IDLE").touch()

    # ── Flags ──────────────────────────────────

    def set_flag(self, flag: str) -> None:
        (self._flags / flag.upper()).touch(exist_ok=True)

    def clear_flag(self, flag: str) -> None:
        try:
            (self._flags / flag.upper()).unlink()
        except FileNotFoundError:
            pass

    def has_flag(self, flag: str) -> bool:
        return (self._flags / flag.upper()).exists()

    def active_flags(self) -> list[str]:
        return sorted(p.name for p in self._flags.iterdir() if p.is_file())

    def transition(self, remove: set[str], add: set[str]) -> None:
        """
        Atomic-ish transition: clear old flags, set new flags.
        Uses individual unlink/touch — not a true atomic swap,
        but fast enough for single-process agents.
        For true atomicity, use rename-based swap (future extension).
        """
        for flag in remove:
            self.clear_flag(flag)
        for flag in add:
            self.set_flag(flag)

    def clear_all(self) -> None:
        for p in self._flags.iterdir():
            if p.is_file():
                p.unlink()

    # ── Step counter ───────────────────────────

    def set_step(self, step: int) -> None:
        """Store current step as filename (zero-padded 16 digits)."""
        # Remove old step files
        for p in self._step_dir.iterdir():
            if p.is_file():
                p.unlink()
        # Create new step file
        (self._step_dir / f"{step:016d}").touch()

    def get_step(self) -> int:
        """Read current step from directory entry."""
        files = sorted(p.name for p in self._step_dir.iterdir() if p.is_file())
        if not files:
            return 0
        return int(files[-1])

    # ── Locks ──────────────────────────────────

    def acquire_lock(self, lock_name: str) -> bool:
        """
        Try to acquire a named lock.
        Returns True if acquired, False if already held.
        Uses exclusive creation (O_CREAT | O_EXCL at filesystem level).
        """
        lock_path = self._locks / lock_name.replace('/', '_').replace(':', '-')
        try:
            fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            os.close(fd)
            return True
        except FileExistsError:
            return False

    def release_lock(self, lock_name: str) -> None:
        lock_path = self._locks / lock_name.replace('/', '_').replace(':', '-')
        try:
            lock_path.unlink()
        except FileNotFoundError:
            pass

    def held_locks(self) -> list[str]:
        return [p.name for p in self._locks.iterdir() if p.is_file()]

    # ── Rate limiting via mtime ─────────────────

    def rate_gate(self, gate_name: str, min_interval_s: float) -> bool:
        """
        Rate limit gate using file mtime.
        Returns True if allowed (enough time elapsed), False if rate-limited.
        Touching the gate file resets the timer.
        """
        gate = InodeGate(self._locks / f"rate_{gate_name}")
        age  = gate.age_seconds()

        if age is None or age >= min_interval_s:
            gate.set()  # reset timer
            return True  # allowed

        return False  # rate limited

    # ── Introspection ──────────────────────────

    def snapshot(self) -> dict:
        return {
            "agent_id":     self.agent_id,
            "active_flags": self.active_flags(),
            "step":         self.get_step(),
            "held_locks":   self.held_locks(),
        }

    def destroy(self) -> None:
        """Remove all inode state for this agent."""
        import shutil
        shutil.rmtree(self._root, ignore_errors=True)


# ─────────────────────────────────────────────
# Multi-agent inode registry
# ─────────────────────────────────────────────

class InodeRegistry:
    """
    Registry of all active agents tracked via inode state.
    Scan base_dir to discover all agents without any central index.
    The directory listing IS the index.
    """

    def __init__(self, base_dir: Path):
        self.base_dir = base_dir
        base_dir.mkdir(parents=True, exist_ok=True)

    def get_agent(self, agent_id: str) -> InodeStateMachine:
        return InodeStateMachine(self.base_dir, agent_id)

    def list_agents(self) -> list[str]:
        return [
            d.name for d in self.base_dir.iterdir()
            if d.is_dir() and (d / "flags").exists()
        ]

    def all_snapshots(self) -> list[dict]:
        snapshots = []
        for agent_id in self.list_agents():
            ism = self.get_agent(agent_id)
            snapshots.append(ism.snapshot())
        return snapshots

    def active_agents(self) -> list[str]:
        """Agents that are not IDLE and not DONE."""
        result = []
        for agent_id in self.list_agents():
            ism = self.get_agent(agent_id)
            flags = set(ism.active_flags())
            if flags and not flags.issubset({"IDLE", "DONE"}):
                result.append(agent_id)
        return result