#!/home/shire3d/ARMOR/venv/bin/python from __future__ import annotations import fcntl import json import os import signal import subprocess import sys import tempfile import time from datetime import datetime, timezone from pathlib import Path from typing import Any VERSION = "0001-supervised-full-throttle" PYTHON = Path("/home/shire3d/ARMOR/venv/bin/python") SCHEDULER = Path("/home/ray/shire-academy-safe-autonomous") STATE_ROOT = Path( "/home/ray/.local/state/shire-academy-full-throttle" ) STATUS_PATH = STATE_ROOT / "status.json" LOCK_PATH = STATE_ROOT / "worker.lock" BACKOFF_DELAYS_SECONDS = ( 120, 300, 900, ) SUCCESS_DELAY_SECONDS = 2 BUSY_DELAY_SECONDS = 10 INFRASTRUCTURE_ACTIONS = { "qualification_infrastructure_retry_scheduled", "counted_infrastructure_retry_scheduled", "safe_cadquery_infrastructure_retry", "scheduler_infrastructure_backoff", } SUPERVISED_STOP_ACTIONS = { "scheduler_paused_for_review", "scheduler_paused_manually", "counted_semantic_failure_paused", "safe_cadquery_semantic_failure_paused", "qualification_safety_stop", "counted_infrastructure_safety_stop", "permission_drift_safety_stop", "unexpected_scheduler_failure", "safe_supported_curriculum_complete", } stop_requested = False def now() -> str: return datetime.now( timezone.utc ).isoformat() def request_stop( signum: int, frame: Any, ) -> None: del signum, frame global stop_requested stop_requested = True def atomic_json_write( path: Path, payload: dict[str, Any], ) -> None: path.parent.mkdir( parents=True, exist_ok=True, ) descriptor, temporary_name = tempfile.mkstemp( prefix=path.name + ".", dir=str(path.parent), ) temporary = Path(temporary_name) try: with os.fdopen( descriptor, "w", encoding="utf-8", ) as handle: json.dump( payload, handle, indent=2, sort_keys=True, ensure_ascii=False, ) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace( temporary, path, ) finally: if temporary.exists(): temporary.unlink() def write_status( **payload: Any, ) -> dict[str, Any]: complete = { "schema": "shire.academy.supervised_full_throttle.v1", "version": VERSION, "updated_at": now(), "forge_unlock_allowed": False, **payload, } atomic_json_write( STATUS_PATH, complete, ) print( json.dumps( complete, sort_keys=True, ensure_ascii=False, ), flush=True, ) return complete def parse_payload( stdout: str, ) -> dict[str, Any] | None: stdout = stdout.strip() if not stdout: return None try: value = json.loads(stdout) except json.JSONDecodeError: return None return value if isinstance(value, dict) else None def payload_is_infrastructure( payload: Any, ) -> bool: pending = [payload] visited: set[int] = set() while pending: item = pending.pop() if not isinstance(item, dict): continue identity = id(item) if identity in visited: continue visited.add(identity) if item.get( "infrastructure_failure" ) is True: return True for value in item.values(): if isinstance(value, dict): pending.append(value) return False def sleep_interruptibly( seconds: int, ) -> None: deadline = time.monotonic() + seconds while ( not stop_requested and time.monotonic() < deadline ): time.sleep( min( 1, max( 0, deadline - time.monotonic(), ), ) ) def run() -> int: global stop_requested STATE_ROOT.mkdir( parents=True, exist_ok=True, ) with LOCK_PATH.open("a+") as lock: try: fcntl.flock( lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB, ) except BlockingIOError: write_status( ok=True, running=False, action= "another_full_throttle_worker_is_running", ) return 0 signal.signal( signal.SIGTERM, request_stop, ) signal.signal( signal.SIGINT, request_stop, ) infrastructure_failures = 0 completed_actions = 0 write_status( ok=True, running=True, action="full_throttle_started", completed_actions=completed_actions, infrastructure_failures=0, next_delay_seconds=0, ) while not stop_requested: started_at = now() process = subprocess.run( [ str(PYTHON), str(SCHEDULER), "tick", ], capture_output=True, text=True, check=False, ) stdout = process.stdout.strip() stderr = process.stderr.strip() payload = parse_payload(stdout) action = ( str(payload.get("action") or "") if isinstance(payload, dict) else "" ) infrastructure = ( action in INFRASTRUCTURE_ACTIONS or payload_is_infrastructure(payload) ) if process.returncode != 0 and not infrastructure: write_status( ok=False, running=False, action= "full_throttle_worker_command_failure", scheduler_action=action or None, scheduler_returncode= process.returncode, scheduler_stdout=stdout[-4000:], scheduler_stderr=stderr[-4000:], completed_actions= completed_actions, infrastructure_failures= infrastructure_failures, started_at=started_at, ) return 1 if ( action in SUPERVISED_STOP_ACTIONS or ( isinstance(payload, dict) and payload.get("paused") is True ) ): write_status( ok=True, running=False, supervised_stop=True, action= "full_throttle_supervised_stop", scheduler_action=action, scheduler_payload=payload, completed_actions= completed_actions, infrastructure_failures= infrastructure_failures, started_at=started_at, ) return 0 if action == "another_scheduler_tick_is_running": write_status( ok=True, running=True, action= "waiting_for_existing_scheduler_tick", scheduler_action=action, completed_actions= completed_actions, infrastructure_failures= infrastructure_failures, next_delay_seconds= BUSY_DELAY_SECONDS, started_at=started_at, ) sleep_interruptibly( BUSY_DELAY_SECONDS ) continue if infrastructure: infrastructure_failures += 1 delay = BACKOFF_DELAYS_SECONDS[ min( infrastructure_failures - 1, len( BACKOFF_DELAYS_SECONDS ) - 1, ) ] write_status( ok=True, running=True, action= "full_throttle_infrastructure_backoff", scheduler_action=action or None, scheduler_payload=payload, scheduler_returncode= process.returncode, scheduler_stderr=stderr[-4000:], completed_actions= completed_actions, infrastructure_failures= infrastructure_failures, next_delay_seconds=delay, started_at=started_at, ) sleep_interruptibly(delay) continue infrastructure_failures = 0 completed_actions += 1 write_status( ok=True, running=True, action= "full_throttle_action_completed", scheduler_action=action or None, scheduler_payload=payload, completed_actions= completed_actions, infrastructure_failures=0, next_delay_seconds= SUCCESS_DELAY_SECONDS, started_at=started_at, ) sleep_interruptibly( SUCCESS_DELAY_SECONDS ) write_status( ok=True, running=False, action="full_throttle_stopped", completed_actions=completed_actions, infrastructure_failures= infrastructure_failures, ) return 0 def status() -> int: if not STATUS_PATH.exists(): print(json.dumps({ "ok": True, "available": False, "running": False, "version": VERSION, "forge_unlock_allowed": False, }, indent=2, sort_keys=True)) return 0 print( STATUS_PATH.read_text( encoding="utf-8" ), end="", ) return 0 def self_test() -> int: assert BACKOFF_DELAYS_SECONDS == ( 120, 300, 900, ) assert SUCCESS_DELAY_SECONDS == 2 assert BUSY_DELAY_SECONDS == 10 assert payload_is_infrastructure({ "payload": { "infrastructure_failure": True, }, }) is True assert payload_is_infrastructure({ "payload": { "infrastructure_failure": False, }, }) is False assert parse_payload( '{"ok":true,"action":"test"}' ) == { "ok": True, "action": "test", } assert parse_payload( "not-json" ) is None assert ( "permission_drift_safety_stop" in SUPERVISED_STOP_ACTIONS ) assert ( "counted_semantic_failure_paused" in SUPERVISED_STOP_ACTIONS ) print(json.dumps({ "ok": True, "version": VERSION, "continuous_completion_driven": True, "success_delay_seconds": SUCCESS_DELAY_SECONDS, "infrastructure_backoff_seconds": list(BACKOFF_DELAYS_SECONDS), "single_worker_flock": True, "permission_drift_hard_stop": True, "semantic_review_supervised_stop": True, "forge_unlock_allowed": False, }, indent=2, sort_keys=True)) return 0 def main() -> int: command = ( sys.argv[1] if len(sys.argv) > 1 else "run" ) if command == "run": return run() if command == "status": return status() if command == "self-test": return self_test() raise SystemExit( "Usage: " + str(Path(sys.argv[0]).name) + " [run|status|self-test]" ) if __name__ == "__main__": raise SystemExit(main())