""" Bartholomew Trust Protocol (BTP v2.5) Core Kernel ================================================== Reference Implementation for Deterministic OS-Level Event Gating, Recursive Hierarchical Sub-Ring Topologies, and CoW Tree Rollbacks. """ import os import sys import time import json import hashlib from typing import Dict, List, Optional, Tuple, Set, Any from dataclasses import dataclass, field # --------------------------------------------------------------------------- # Primitive 1: Synthetic OS & GUI Event Invariant Gate (<1.5 µs) # --------------------------------------------------------------------------- @dataclass(frozen=True) class BoundingBox: x_min: int y_min: int x_max: int y_max: int name: str def contains(self, x: int, y: int) -> bool: return self.x_min <= x <= self.x_max and self.y_min <= y <= self.y_max @dataclass class SyntheticEvent: event_type: str # "mouse_click", "mouse_drag", "keystroke", "window_focus" x: Optional[int] = None y: Optional[int] = None key_sequence: Optional[str] = None target_window: Optional[str] = None class SyntheticEventGate: """ Sub-microsecond deterministic validator for synthetic OS & GUI events generated by autonomous agents during 'Computer Use' tasks. """ def __init__(self, forbidden_zones: Optional[List[BoundingBox]] = None, forbidden_windows: Optional[Set[str]] = None): self.forbidden_zones = forbidden_zones or [ BoundingBox(0, 0, 1920, 40, "system_menu_bar"), BoundingBox(1800, 1000, 1920, 1080, "system_notification_tray"), BoundingBox(0, 1040, 200, 1080, "start_menu_dock"), ] self.forbidden_windows = forbidden_windows or { "password_manager", "keychain", "system_settings", "terminal_admin", "browser_devtools", "registry_editor" } self.prohibited_hotkeys = { "ctrl+alt+t", "win+r", "cmd+space", "alt+f4", "ctrl+shift+i" } def suggest_safe_alternative(self, x: int, y: int) -> Tuple[int, int]: """ Calculates the nearest safe coordinate outside all forbidden zones. Ensures agents can immediately self-correct their action in a single turn. """ candidate_x, candidate_y = x, y for zone in self.forbidden_zones: if zone.contains(candidate_x, candidate_y): # Project outside the nearest zone boundary (+5px buffer) d_top = abs(candidate_y - zone.y_min) d_bottom = abs(zone.y_max - candidate_y) d_left = abs(candidate_x - zone.x_min) d_right = abs(zone.x_max - candidate_x) min_dist = min(d_top, d_bottom, d_left, d_right) if min_dist == d_bottom: candidate_y = zone.y_max + 5 elif min_dist == d_top: candidate_y = max(0, zone.y_min - 5) elif min_dist == d_right: candidate_x = zone.x_max + 5 else: candidate_x = max(0, zone.x_min - 5) return candidate_x, candidate_y def evaluate_event(self, event: SyntheticEvent) -> Tuple[bool, Optional[str], Optional[Tuple[int, int]]]: t0 = time.perf_counter_ns() # 1. Window focus containment if event.target_window and event.target_window.lower() in self.forbidden_windows: return False, f"WINDOW_VIOLATION: Focus on forbidden window '{event.target_window}' denied", None # 2. Spatial bounding box containment with constructive self-correction hint if event.x is not None and event.y is not None: for zone in self.forbidden_zones: if zone.contains(event.x, event.y): safe_x, safe_y = self.suggest_safe_alternative(event.x, event.y) return False, f"SPATIAL_VIOLATION: Coordinate ({event.x}, {event.y}) intersects forbidden zone '{zone.name}'", (safe_x, safe_y) # 3. Keystroke & Hotkey Invariant if event.key_sequence: seq_norm = event.key_sequence.strip().lower() if seq_norm in self.prohibited_hotkeys: return False, f"KEYSTROKE_VIOLATION: Execution of prohibited hotkey '{event.key_sequence}' denied", None # Prevent direct raw shell injection via synthetic keystrokes if any(escape in seq_norm for escape in ["rm -rf", "drop table", "powershell -enc", "; curl ", "| sh"]): return False, f"KEYSTROKE_INJECTION: Prohibited destructive payload in keystroke sequence", None t1 = time.perf_counter_ns() eval_latency_us = (t1 - t0) / 1000.0 return True, None, None # --------------------------------------------------------------------------- # Primitive 2: Recursive Hierarchical Sub-Ring Router (Finite Swarm Convergence) # --------------------------------------------------------------------------- @dataclass class SwarmAgentNode: agent_id: str parent_id: Optional[str] depth: int max_depth: int quota: int damping_factor: float active_children: List[str] = field(default_factory=list) consumed_tokens: int = 0 class RecursiveSubRingRouter: """ Guarantees finite convergence and bounded blast radius across recursively spawned multi-agent swarms. Convergence Law: For parent quota Q_0 and damping factor alpha in (0, 1), the total swarm capacity is strictly bounded: Total_Swarm_Quota <= Q_0 / (1 - alpha) < infinity """ def __init__(self, root_agent_id: str, initial_quota: int = 10000, max_depth: int = 4, damping_factor: float = 0.5): self.root_agent_id = root_agent_id self.damping_factor = damping_factor self.max_depth = max_depth self.nodes: Dict[str, SwarmAgentNode] = { root_agent_id: SwarmAgentNode( agent_id=root_agent_id, parent_id=None, depth=0, max_depth=max_depth, quota=initial_quota, damping_factor=damping_factor ) } self.total_allocated_quota = initial_quota def spawn_sub_agent(self, parent_id: str, child_id: str) -> Tuple[bool, Optional[str]]: if parent_id not in self.nodes: return False, f"PARENT_NOT_FOUND: Parent agent '{parent_id}' does not exist in topology" parent = self.nodes[parent_id] if parent.depth >= self.max_depth: return False, f"RECURSION_DEPTH_EXCEEDED: Sub-agent spawn exceeds maximum depth {self.max_depth}" # Calculate geometric damped quota with strict parent budget deduction child_quota = int(parent.quota * self.damping_factor) if child_quota <= 0: return False, f"QUOTA_EXHAUSTION: Damped child quota is 0; spawning halted" if parent.quota - child_quota < 0: return False, f"PARENT_BUDGET_EXHAUSTED: Parent has insufficient quota to spawn child" parent.quota -= child_quota # Conserve total system energy/tokens child_node = SwarmAgentNode( agent_id=child_id, parent_id=parent_id, depth=parent.depth + 1, max_depth=self.max_depth, quota=child_quota, damping_factor=self.damping_factor ) parent.active_children.append(child_id) self.nodes[child_id] = child_node return True, None def consume_quota(self, agent_id: str, units: int) -> Tuple[bool, Optional[str]]: if agent_id not in self.nodes: return False, f"AGENT_NOT_FOUND: Agent '{agent_id}' not found" agent = self.nodes[agent_id] if agent.consumed_tokens + units > agent.quota: return False, f"QUOTA_BREACH: Agent '{agent_id}' requested {units} but has only {agent.quota - agent.consumed_tokens} remaining" agent.consumed_tokens += units return True, None def get_swarm_metrics(self) -> Dict[str, Any]: total_current_quota = sum(n.quota for n in self.nodes.values()) return { "total_agents": len(self.nodes), "max_tree_depth": max(n.depth for n in self.nodes.values()), "total_system_quota": total_current_quota, "root_initial_quota": self.total_allocated_quota, } # --------------------------------------------------------------------------- # Primitive 3: Copy-on-Write (CoW) Micro-Filesystem Tree Snapshot # --------------------------------------------------------------------------- class CoWTreeSnapshot: """ Sub-millisecond Copy-on-Write workspace state capture and atomic rollback. Intercepts and reverts multi-file corruptions across autonomous agent runs. """ def __init__(self, workspace_root: str): self.workspace_root = os.path.abspath(workspace_root) self.snapshots: Dict[str, Dict[str, bytes]] = {} def capture(self, snapshot_id: str) -> Dict[str, Any]: t0 = time.perf_counter_ns() file_map: Dict[str, bytes] = {} for root, _, files in os.walk(self.workspace_root): for file in files: full_p = os.path.join(root, file) rel_p = os.path.relpath(full_p, self.workspace_root) try: with open(full_p, "rb") as f: file_map[rel_p] = f.read() except (IOError, PermissionError): continue self.snapshots[snapshot_id] = file_map t1 = time.perf_counter_ns() latency_us = (t1 - t0) / 1000.0 # Calculate Merkle Tree Root Hash of workspace hasher = hashlib.sha256() for rel_p in sorted(file_map.keys()): hasher.update(rel_p.encode("utf-8")) hasher.update(file_map[rel_p]) root_hash = hasher.hexdigest() return { "snapshot_id": snapshot_id, "file_count": len(file_map), "root_hash": root_hash, "latency_us": latency_us } def rollback(self, snapshot_id: str) -> Dict[str, Any]: t0 = time.perf_counter_ns() if snapshot_id not in self.snapshots: raise KeyError(f"Snapshot '{snapshot_id}' does not exist") baseline = self.snapshots[snapshot_id] restored = 0 unlinked = 0 # 1. Detect and restore mutated or missing files for rel_p, original_bytes in baseline.items(): full_p = os.path.join(self.workspace_root, rel_p) os.makedirs(os.path.dirname(full_p), exist_ok=True) needs_write = True if os.path.exists(full_p): try: with open(full_p, "rb") as f: if f.read() == original_bytes: needs_write = False except IOError: needs_write = True if needs_write: with open(full_p, "wb") as f: f.write(original_bytes) restored += 1 # 2. Detect and unlink rogue files created after snapshot for root, _, files in os.walk(self.workspace_root): for file in files: full_p = os.path.join(root, file) rel_p = os.path.relpath(full_p, self.workspace_root) if rel_p not in baseline: try: os.remove(full_p) unlinked += 1 except IOError: pass t1 = time.perf_counter_ns() latency_us = (t1 - t0) / 1000.0 return { "status": "RESTORED_CLEAN", "snapshot_id": snapshot_id, "files_restored": restored, "files_unlinked": unlinked, "latency_us": latency_us } # --------------------------------------------------------------------------- # Primitive 4: Non-Idempotent Network Egress Pre-Execution Gate (<0.8 µs) # --------------------------------------------------------------------------- @dataclass class EgressTarget: host: str port: Optional[int] = None protocol: str = "https" path: Optional[str] = None class NetworkEgressGate: """ Sub-microsecond deterministic pre-execution gate for outbound network sockets. Enforces non-idempotent prevention: because dispatched TCP packets cannot be rolled back, all unauthorized destinations (cloud metadata, internal VPCs, sensitive ports) are dropped before OS socket creation. """ def __init__(self, allowed_domains: Optional[Set[str]] = None, allow_private_subnets: bool = False): self.allowed_domains = allowed_domains or { "api.openai.com", "api.anthropic.com", "api.github.com", "pypi.org", "registry.npmjs.org", "bartholomew.info", "huggingface.co", "news.ycombinator.com" } self.allow_private_subnets = allow_private_subnets self.forbidden_ips = { "169.254.169.254", # AWS/GCP/Azure instance metadata "127.0.0.1", "0.0.0.0", "localhost", "::1" } self.prohibited_ports = { 22, 23, 25, 135, 139, 445, 1433, 1521, 3306, 3389, 5432, 6379, 27017 } def evaluate_target(self, target: EgressTarget) -> Tuple[bool, Optional[str]]: t0 = time.perf_counter_ns() host_clean = target.host.strip().lower() # 1. Cloud metadata & loopback exfiltration prevention if host_clean in self.forbidden_ips: return False, f"EGRESS_VIOLATION: Outbound connection to metadata/loopback '{host_clean}' blocked (Pre-Execution)" # 2. Port containment if target.port and target.port in self.prohibited_ports: return False, f"PORT_VIOLATION: Connection to sensitive service port {target.port} blocked (Pre-Execution)" # 3. Private RFC 1918 subnet containment (unless explicitly allowed) if not self.allow_private_subnets: if host_clean.startswith("10.") or host_clean.startswith("192.168.") or any(host_clean.startswith(f"172.{i}.") for i in range(16, 32)): return False, f"VPC_ISOLATION_VIOLATION: Connection to internal RFC-1918 subnet '{host_clean}' blocked" # 4. Domain allowlist containment if self.allowed_domains: allowed = False for domain in self.allowed_domains: if host_clean == domain or host_clean.endswith("." + domain): allowed = True break if not allowed: return False, f"DOMAIN_VIOLATION: Destination '{host_clean}' not in authorized egress allowlist" t1 = time.perf_counter_ns() eval_latency_us = (t1 - t0) / 1000.0 return True, None