""" I/O Tower - Human checkpoints where Users connect to The Grid. "The I/O Tower is sacred. It's where we commune with the Users." """ from dataclasses import dataclass, field from datetime import datetime from typing import Optional, Any, Callable from enum import Enum import uuid class CheckpointReason(Enum): """Reason for I/O Tower checkpoint.""" FISSION_DEPTH = "fission_depth" BEFORE_COMMIT = "before_commit" ERROR_OCCURRED = "error_occurred" LOW_ENERGY = "low_energy" RECOGNIZER_FAILED = "recognizer_failed" USER_REQUESTED = "user_requested" DECISION_REQUIRED = "decision_required" REVIEW_REQUIRED = "review_required" class UserDecision(Enum): """User decision at checkpoint.""" AUTHORIZE = "authorize" # Allow action to proceed DENY = "deny" # Block action MODIFY = "modify" # Modify before proceeding MERGE = "merge" # Merge operations (for fission) ABORT = "abort" # Abort entire operation DEFER = "defer" # Defer decision @dataclass class CheckpointOption: """An option presented at a checkpoint.""" key: str label: str description: str decision: UserDecision @dataclass class CheckpointContext: """Context for an I/O Tower checkpoint.""" reason: CheckpointReason program_id: str program_name: str depth: int energy_remaining: int energy_allocated: int pending_actions: list[str] = field(default_factory=list) details: dict = field(default_factory=dict) @dataclass class CheckpointResult: """Result of a checkpoint interaction.""" decision: UserDecision timestamp: datetime user_input: Optional[str] = None modifications: dict = field(default_factory=dict) class IOTower: """ I/O Tower - Human checkpoint interface. Provides human checkpoints at critical moments: - Fission depth >= configured threshold - Before commits - On errors - Energy running low - Recognizer flags issues """ # Default options for common checkpoint types DEFAULT_OPTIONS = { CheckpointReason.FISSION_DEPTH: [ CheckpointOption("A", "Authorize", "Allow fission to continue", UserDecision.AUTHORIZE), CheckpointOption("M", "Merge", "Combine into single Program (saves Energy)", UserDecision.MERGE), CheckpointOption("D", "Deny", "Block spawning, continue with current Program", UserDecision.DENY), CheckpointOption("V", "View Disc", "See full context before deciding", UserDecision.DEFER), ], CheckpointReason.BEFORE_COMMIT: [ CheckpointOption("C", "Commit", "Proceed with commit", UserDecision.AUTHORIZE), CheckpointOption("R", "Review", "Review changes first", UserDecision.DEFER), CheckpointOption("M", "Modify", "Modify commit message/contents", UserDecision.MODIFY), CheckpointOption("A", "Abort", "Cancel commit", UserDecision.ABORT), ], CheckpointReason.ERROR_OCCURRED: [ CheckpointOption("R", "Retry", "Retry the operation", UserDecision.AUTHORIZE), CheckpointOption("S", "Skip", "Skip this operation", UserDecision.DENY), CheckpointOption("A", "Abort", "Abort the entire process", UserDecision.ABORT), CheckpointOption("D", "Debug", "View debug information", UserDecision.DEFER), ], CheckpointReason.LOW_ENERGY: [ CheckpointOption("C", "Continue", "Continue with remaining energy", UserDecision.AUTHORIZE), CheckpointOption("A", "Allocate", "Allocate more energy", UserDecision.MODIFY), CheckpointOption("P", "Pause", "Pause and save state", UserDecision.DEFER), CheckpointOption("S", "Stop", "Stop execution", UserDecision.ABORT), ], CheckpointReason.RECOGNIZER_FAILED: [ CheckpointOption("F", "Fix", "Fix issues and retry", UserDecision.MODIFY), CheckpointOption("O", "Override", "Override and continue", UserDecision.AUTHORIZE), CheckpointOption("R", "Review", "Review issues in detail", UserDecision.DEFER), CheckpointOption("A", "Abort", "Abort due to validation failure", UserDecision.ABORT), ], } def __init__( self, require_at_depth: int = 3, require_before_commit: bool = True, require_on_error: bool = True, require_on_low_energy: bool = True, timeout_seconds: int = 300, ): self.require_at_depth = require_at_depth self.require_before_commit = require_before_commit self.require_on_error = require_on_error self.require_on_low_energy = require_on_low_energy self.timeout_seconds = timeout_seconds # Checkpoint history self.checkpoints: list[tuple[CheckpointContext, CheckpointResult]] = [] # Callback for user input (to be set by CLI) self.input_handler: Optional[Callable[[CheckpointContext, list[CheckpointOption]], CheckpointResult]] = None def should_checkpoint( self, reason: CheckpointReason, depth: int = 0, energy_percentage: float = 100.0, ) -> bool: """Determine if a checkpoint is required.""" if reason == CheckpointReason.FISSION_DEPTH: return depth >= self.require_at_depth elif reason == CheckpointReason.BEFORE_COMMIT: return self.require_before_commit elif reason == CheckpointReason.ERROR_OCCURRED: return self.require_on_error elif reason == CheckpointReason.LOW_ENERGY: return self.require_on_low_energy and energy_percentage < 10 elif reason == CheckpointReason.RECOGNIZER_FAILED: return True # Always checkpoint on validation failure elif reason == CheckpointReason.USER_REQUESTED: return True return False def checkpoint( self, context: CheckpointContext, options: list[CheckpointOption] = None, ) -> CheckpointResult: """ Trigger a checkpoint and await user decision. In actual use, this would pause execution and present options to the user through the CLI. """ if options is None: options = self.DEFAULT_OPTIONS.get(context.reason, []) # If we have an input handler, use it if self.input_handler: result = self.input_handler(context, options) else: # Default: authorize (for testing/non-interactive) result = CheckpointResult( decision=UserDecision.AUTHORIZE, timestamp=datetime.now(), ) self.checkpoints.append((context, result)) return result def render_checkpoint( self, context: CheckpointContext, options: list[CheckpointOption], ) -> str: """Render checkpoint display.""" lines = [ "╔" + "═" * 77 + "╗", "║ ⛯ I/O TOWER - Human Input Required" + " " * 40 + "║", "╠" + "═" * 77 + "╣", "║" + " " * 77 + "║", f"║ REASON: {context.reason.value:<65}║", "║" + " " * 77 + "║", "║ CONTEXT:" + " " * 67 + "║", ] # Program info lines.append(f"║ Program \"{context.program_name}\" (depth: {context.depth})" + " " * (48 - len(context.program_name)) + "║") lines.append(f"║ Energy: {context.energy_remaining}/{context.energy_allocated}" + " " * 55 + "║") # Pending actions if context.pending_actions: lines.append("║" + " " * 77 + "║") lines.append("║ Pending actions:" + " " * 59 + "║") for action in context.pending_actions[:5]: # Limit display lines.append(f"║ • {action:<71}║") # Details if context.details: lines.append("║" + " " * 77 + "║") for key, value in list(context.details.items())[:5]: lines.append(f"║ {key}: {str(value):<69}║") # Options lines.append("║" + " " * 77 + "║") lines.append("║ ┌" + "─" * 73 + "┐ ║") for opt in options: lines.append(f"║ │ [{opt.key}] {opt.label} - {opt.description:<55}│ ║") lines.append("║ └" + "─" * 73 + "┘ ║") lines.append("║" + " " * 77 + "║") lines.append("╚" + "═" * 77 + "╝") return "\n".join(lines) def get_history(self) -> list[dict]: """Get checkpoint history.""" return [ { "context": { "reason": ctx.reason.value, "program_id": ctx.program_id, "program_name": ctx.program_name, "depth": ctx.depth, }, "result": { "decision": res.decision.value, "timestamp": res.timestamp.isoformat(), "user_input": res.user_input, }, } for ctx, res in self.checkpoints ]