File size: 11,349 Bytes
9abace2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 | """
Paradigm 1: Unix Environment State Machine
Part of SOVEREIGN PYTHON LLM ENGINE
Agent state persisted in OS environment variables and inherited file
descriptors. The kernel holds the state — zero disk I/O.
Two mechanisms:
A. EnvStateMachine — serialize/deserialize agent state as bitmask env var
B. DaemonRestarter — hot-restart a daemon via os.execv, passing state
through the process environment so the replacement
process inherits full context with no disk read
Bitmask encoding:
State flags packed into a single base64url string stored in
SOVEREIGN_STATE env var. Each flag is one bit. Up to 64 flags
in a single 8-byte (uint64) integer.
Example:
THINKING=0, ACTING=1, OBSERVING=2, REFLECTING=3, DONE=4, ERROR=5
Active flags {ACTING, OBSERVING} → bitmask = 0b000110 = 6
Encoded: SOVEREIGN_STATE=AAAAAAAAAAY=
"""
from __future__ import annotations
import base64
import os
import struct
import sys
from dataclasses import dataclass, field
from typing import Any
# ─────────────────────────────────────────────
# Bitmask state encoding
# ─────────────────────────────────────────────
ENV_KEY = "SOVEREIGN_STATE"
ENV_META_KEY = "SOVEREIGN_META"
ENV_AGENT_KEY = "SOVEREIGN_AGENT_ID"
ENV_STEP_KEY = "SOVEREIGN_STEP"
ENV_VERSION_KEY = "SOVEREIGN_VERSION"
# Built-in flag indices (0-based bit positions)
FLAG_IDLE = 0
FLAG_THINKING = 1
FLAG_ACTING = 2
FLAG_OBSERVING = 3
FLAG_REFLECTING = 4
FLAG_DONE = 5
FLAG_ERROR = 6
FLAG_NAMES = {
FLAG_IDLE: "IDLE",
FLAG_THINKING: "THINKING",
FLAG_ACTING: "ACTING",
FLAG_OBSERVING: "OBSERVING",
FLAG_REFLECTING: "REFLECTING",
FLAG_DONE: "DONE",
FLAG_ERROR: "ERROR",
}
def _encode_bitmask(flags: set[int]) -> str:
"""Encode set of flag indices → base64url string (8 bytes)."""
mask = 0
for f in flags:
if 0 <= f < 64:
mask |= (1 << f)
return base64.urlsafe_b64encode(struct.pack('>Q', mask)).decode('ascii')
def _decode_bitmask(encoded: str) -> set[int]:
"""Decode base64url string → set of active flag indices."""
try:
raw = base64.urlsafe_b64decode(encoded.encode('ascii'))
mask = struct.unpack('>Q', raw[:8])[0]
return {i for i in range(64) if mask & (1 << i)}
except Exception:
return set()
# ─────────────────────────────────────────────
# Env State Machine
# ─────────────────────────────────────────────
@dataclass
class EnvSnapshot:
agent_id: str
step: int
flags: set[int]
version: int
raw_env: dict[str, str] = field(default_factory=dict)
def flag_names(self) -> list[str]:
return [FLAG_NAMES.get(f, f"FLAG_{f}") for f in sorted(self.flags)]
def has_flag(self, flag: int) -> bool:
return flag in self.flags
class EnvStateMachine:
"""
Persist agent state in process environment variables.
State survives across os.execv() restarts because child processes
inherit the parent's environment. No disk I/O required.
Usage:
esm = EnvStateMachine(agent_id="react_1")
esm.set_flags({FLAG_THINKING})
esm.set_step(5)
# On daemon restart via exec:
snapshot = EnvStateMachine.load_from_env()
if snapshot:
esm.restore(snapshot)
"""
def __init__(self, agent_id: str):
self.agent_id = agent_id
self._flags: set[int] = {FLAG_IDLE}
self._step: int = 0
self._version: int = 0
# ── Write ──────────────────────────────────
def set_flags(self, flags: set[int]) -> None:
self._flags = flags
self._flush()
def add_flag(self, flag: int) -> None:
self._flags.add(flag)
self._flush()
def remove_flag(self, flag: int) -> None:
self._flags.discard(flag)
self._flush()
def set_step(self, step: int) -> None:
self._step = step
self._flush()
def transition(self, remove: set[int], add: set[int]) -> None:
self._flags -= remove
self._flags |= add
self._version += 1
self._flush()
def _flush(self) -> None:
os.environ[ENV_KEY] = _encode_bitmask(self._flags)
os.environ[ENV_AGENT_KEY] = self.agent_id
os.environ[ENV_STEP_KEY] = str(self._step)
os.environ[ENV_VERSION_KEY] = str(self._version)
# ── Read ───────────────────────────────────
@staticmethod
def load_from_env(env: dict[str, str] | None = None) -> EnvSnapshot | None:
"""
Load snapshot from current process environment (or supplied dict).
Returns None if no state found in environment.
"""
e = env if env is not None else dict(os.environ)
encoded = e.get(ENV_KEY)
if not encoded:
return None
return EnvSnapshot(
agent_id=e.get(ENV_AGENT_KEY, "unknown"),
step=int(e.get(ENV_STEP_KEY, "0")),
flags=_decode_bitmask(encoded),
version=int(e.get(ENV_VERSION_KEY, "0")),
raw_env={k: v for k, v in e.items() if k.startswith("SOVEREIGN_")}
)
def restore(self, snapshot: EnvSnapshot) -> None:
self._flags = snapshot.flags
self._step = snapshot.step
self._version = snapshot.version
self._flush()
def snapshot(self) -> EnvSnapshot:
return EnvSnapshot(
agent_id=self.agent_id,
step=self._step,
flags=set(self._flags),
version=self._version
)
# ── Properties ─────────────────────────────
@property
def current_flags(self) -> set[int]:
return set(self._flags)
@property
def step(self) -> int:
return self._step
@property
def is_idle(self) -> bool: return FLAG_IDLE in self._flags
@property
def is_thinking(self) -> bool: return FLAG_THINKING in self._flags
@property
def is_acting(self) -> bool: return FLAG_ACTING in self._flags
@property
def is_done(self) -> bool: return FLAG_DONE in self._flags
@property
def is_error(self) -> bool: return FLAG_ERROR in self._flags
# ─────────────────────────────────────────────
# Daemon hot-restart via execv
# ─────────────────────────────────────────────
class DaemonRestarter:
"""
Hot-restart a Python daemon via os.execv, passing state through
the process environment. The replacement process starts with full
context inherited — zero disk reads required.
Usage:
restarter = DaemonRestarter()
# Before restart — flush state to env
esm.set_step(current_step)
esm.set_flags(current_flags)
# Restart — replaces current process image
restarter.exec_restart()
# In new process — check for inherited state
snapshot = EnvStateMachine.load_from_env()
if snapshot:
print(f"Resumed from step {snapshot.step}")
"""
RESTART_FLAG = "SOVEREIGN_RESTARTED"
def exec_restart(
self,
argv: list[str] | None = None,
extra_env: dict[str, str] | None = None
) -> None:
"""
Replace current process with fresh copy of same script.
State in os.environ is inherited automatically.
Args:
argv: Command to exec (defaults to current sys.argv)
extra_env: Additional env vars to set before exec
"""
os.environ[self.RESTART_FLAG] = "1"
if extra_env:
os.environ.update(extra_env)
args = argv or sys.argv
executable = sys.executable
# os.execv replaces current process — no return
os.execv(executable, [executable] + args)
@staticmethod
def is_restarted() -> bool:
return os.environ.get(DaemonRestarter.RESTART_FLAG) == "1"
@staticmethod
def clear_restart_flag() -> None:
os.environ.pop(DaemonRestarter.RESTART_FLAG, None)
# ─────────────────────────────────────────────
# Pipe-based state transfer (parent→child)
# ─────────────────────────────────────────────
class PipeStateTransfer:
"""
Transfer agent state to child process via inherited pipe fd.
Child reads state from fd before doing any work.
Parent:
pst = PipeStateTransfer()
pid, fd_write = pst.spawn_with_state(cmd, state_bytes)
Child (reads from fd 3 by convention):
state_bytes = PipeStateTransfer.read_from_fd(3)
"""
STATE_FD = 3 # fd 3 by convention (0=stdin, 1=stdout, 2=stderr)
def spawn_with_state(
self,
cmd: list[str],
state_bytes: bytes
) -> tuple[int, int]:
"""
Spawn child process, passing state_bytes via inherited pipe.
Returns:
(pid, write_fd) — write_fd is the write end (parent owns it)
"""
import subprocess
r_fd, w_fd = os.pipe()
# Write state to pipe before spawning
os.write(w_fd, struct.pack('>I', len(state_bytes)))
os.write(w_fd, state_bytes)
proc = subprocess.Popen(
cmd,
close_fds=False, # allow fd inheritance
pass_fds=(r_fd,)
)
os.close(r_fd) # parent closes read end
return proc.pid, w_fd
@staticmethod
def read_from_fd(fd: int = STATE_FD) -> bytes | None:
"""
Child reads state from inherited fd.
Returns None if fd not available.
"""
try:
length_bytes = os.read(fd, 4)
if len(length_bytes) < 4:
return None
length = struct.unpack('>I', length_bytes)[0]
data = b''
while len(data) < length:
chunk = os.read(fd, length - len(data))
if not chunk:
break
data += chunk
os.close(fd)
return data if len(data) == length else None
except OSError:
return None
|