""" Program - Autonomous executor on The Grid. "Programs are living entities on The Grid. They execute, they adapt, they spawn." """ from dataclasses import dataclass, field from datetime import datetime from typing import Optional, Callable, Any from enum import Enum import uuid from .disc import IdentityDisc class ProgramStatus(Enum): """Status of a Program.""" INITIALIZING = "initializing" RUNNING = "running" SPAWNING = "spawning" WAITING = "waiting" COMPLETED = "completed" FAILED = "failed" DEREZZED = "derezzed" class ProgramType(Enum): """Type of Program.""" STANDARD = "standard" RECOGNIZER = "recognizer" LIGHT_CYCLE = "light_cycle" # Fast execution path @dataclass class SpawnRequest: """Request to spawn a child Program.""" name: str purpose: str energy_budget: int constraints: list[str] = field(default_factory=list) approved: bool = False class Program: """ Program - Autonomous executor (subagent) on The Grid. Programs can: - Execute tasks autonomously - Spawn child Programs (fission) - Carry context via Identity Disc - Consume energy from their budget """ def __init__( self, name: str, purpose: str, parent: Optional["Program"] = None, energy_budget: int = 100, program_type: ProgramType = ProgramType.STANDARD, ): self.id = str(uuid.uuid4())[:8] self.name = name self.purpose = purpose self.program_type = program_type self.status = ProgramStatus.INITIALIZING self.created_at = datetime.now() self.started_at: Optional[datetime] = None self.completed_at: Optional[datetime] = None # Fission lineage self.parent = parent self.children: list["Program"] = [] self.depth = 0 if parent is None else parent.depth + 1 # Identity Disc (context carrier) parent_disc = parent.disc if parent else None self.disc = IdentityDisc( program_id=f"program:{self.id}", purpose=purpose, parent_disc=parent_disc, ) self.disc.energy_allocated = energy_budget # Execution state self.current_action = "Initializing" self.output: Any = None self.error: Optional[str] = None # Spawn requests (pending fission) self.spawn_requests: list[SpawnRequest] = [] def start(self) -> None: """Start Program execution.""" self.status = ProgramStatus.RUNNING self.started_at = datetime.now() self.current_action = "Running" def update_action(self, action: str) -> None: """Update current action.""" self.current_action = action def request_spawn( self, name: str, purpose: str, energy_budget: int, constraints: list[str] = None, ) -> SpawnRequest: """Request to spawn a child Program.""" request = SpawnRequest( name=name, purpose=purpose, energy_budget=energy_budget, constraints=constraints or [], ) self.spawn_requests.append(request) self.status = ProgramStatus.SPAWNING return request def spawn(self, request: SpawnRequest) -> Optional["Program"]: """Spawn a child Program from approved request.""" if not request.approved: return None # Deduct energy for spawn if not self.disc.consume_energy(request.energy_budget): return None child = Program( name=request.name, purpose=request.purpose, parent=self, energy_budget=request.energy_budget, ) for constraint in request.constraints: child.disc.add_constraint(constraint) self.children.append(child) return child def consume_energy(self, amount: int, description: str = "") -> bool: """Consume energy for an operation.""" success = self.disc.consume_energy(amount) if success: self.disc.record_discovery( finding=f"Energy consumed: {amount}", source="energy_system", relevance="low", ) return success def complete(self, output: Any = None) -> None: """Mark Program as completed.""" self.status = ProgramStatus.COMPLETED self.completed_at = datetime.now() self.output = output self.current_action = "Completed" # Return unused energy to parent if self.parent: unused = self.disc.energy_remaining() if unused > 0: self.parent.disc.energy_allocated += unused def fail(self, error: str) -> None: """Mark Program as failed.""" self.status = ProgramStatus.FAILED self.completed_at = datetime.now() self.error = error self.current_action = f"Failed: {error}" def derez(self) -> None: """Derez (terminate) this Program and all children.""" self.status = ProgramStatus.DEREZZED self.disc.derez() for child in self.children: child.derez() def wait_for_children(self) -> None: """Set status to waiting for children to complete.""" self.status = ProgramStatus.WAITING self.current_action = "Waiting for child Programs" def all_children_completed(self) -> bool: """Check if all child Programs are completed.""" return all( c.status in (ProgramStatus.COMPLETED, ProgramStatus.DEREZZED) for c in self.children ) def any_child_failed(self) -> bool: """Check if any child Program failed.""" return any( c.status == ProgramStatus.FAILED for c in self.children ) def get_lineage(self) -> list[str]: """Get the lineage chain of Program names.""" return self.disc.get_lineage() def depth_display(self) -> str: """Get depth indicator for display.""" return " " * self.depth def status_icon(self) -> str: """Get status icon for display.""" icons = { ProgramStatus.INITIALIZING: "○", ProgramStatus.RUNNING: "◐", ProgramStatus.SPAWNING: "◎", ProgramStatus.WAITING: "◑", ProgramStatus.COMPLETED: "●", ProgramStatus.FAILED: "✗", ProgramStatus.DEREZZED: "◯", } return icons.get(self.status, "?") def duration(self) -> Optional[float]: """Get execution duration in seconds.""" if not self.started_at: return None end = self.completed_at or datetime.now() return (end - self.started_at).total_seconds() def energy_percentage(self) -> float: """Get energy as percentage.""" return self.disc.energy_percentage() def summary(self) -> str: """Get Program summary.""" lines = [ f"Program: {self.name} [{self.id}]", f" Purpose: {self.purpose}", f" Status: {self.status.value}", f" Depth: {self.depth}", f" Energy: {self.disc.energy_remaining()}/{self.disc.energy_allocated}", f" Children: {len(self.children)}", f" Action: {self.current_action}", ] return "\n".join(lines) def to_dict(self) -> dict: """Serialize Program to dictionary.""" return { "id": self.id, "name": self.name, "purpose": self.purpose, "type": self.program_type.value, "status": self.status.value, "depth": self.depth, "current_action": self.current_action, "energy_remaining": self.disc.energy_remaining(), "energy_allocated": self.disc.energy_allocated, "children": [c.to_dict() for c in self.children], "lineage": self.get_lineage(), "output": str(self.output) if self.output else None, "error": self.error, "created_at": self.created_at.isoformat(), "started_at": self.started_at.isoformat() if self.started_at else None, "completed_at": self.completed_at.isoformat() if self.completed_at else None, } def __repr__(self) -> str: return f"Program(id={self.id}, name={self.name}, depth={self.depth})"